live-gateway/douyu/live.go

package douyu

import (
	"git.noahlan.cn/northlan/ntools-go/logger"
	"github.com/pkg/errors"
	"live-gateway/config"
	"live-gateway/douyu/msg_handler"
	"live-gateway/live"
	"live-gateway/ws"
	"time"
)

type MsgHandler interface {
	TypedData
	HandlerMessage(data []byte)
}

// 实现 live.Handler 接口
var _ live.Handler = (*LiveDouyu)(nil)

type (
	LiveDouyu struct {
		*live.Live
		sequenceId uint32

		msgHandlerMapper map[string]MsgHandler

		//loginResChan chan struct{} // login res
		entered chan struct{} // join group
	}
)

func NewLiveDouyu() *LiveDouyu {
	bl := &LiveDouyu{
		msgHandlerMapper: make(map[string]MsgHandler, 6),
		entered:          make(chan struct{}),
	}

	l := live.NewLive(
		live.WithWsOptions(
			ws.WithPacker(NewPackDouyu()),
			ws.WithCodec(NewCodecDouyu()),
		),
	)
	l.PreConnect(bl.preConnect)

	l.Init(bl.Init)
	l.Handler(bl)

	bl.Live = l
	return bl
}

func (l *LiveDouyu) registerMessageHandler(h ...MsgHandler) {
	for _, handler := range h {
		l.msgHandlerMapper[handler.DataType()] = handler
	}
}

func (l *LiveDouyu) preConnect() (url string, err error) {
	cfg := config.Config.Douyu
	url = cfg.Custom.Url
	// 注册监听器
	l.registerMessageHandler(
		msg_handler.NewMsgDanmakuHandler(cfg.RoomId),
		msg_handler.NewMsgGiftHandler(cfg.RoomId),
	)
	return
}

func (l *LiveDouyu) Init(conn *ws.NWebsocket) (err error) {
	cfg := config.Config.Douyu
	switch cfg.Type {
	case config.TypeOfficial:
	case config.TypeCustom:
		// 1. login
		err = conn.SendBinaryMessage(&MsgLoginReq{
			RoomId: cfg.RoomId,
			//Dfl:      "sn@AA=105@ASss@AA=1",
			//Dfl: MsgLoginReqDfl{},
			// sn@=105/ss@=1
			Username: "auto_EHUwJCggl7",
			Uid:      cfg.UserId,
			Ver:      "20190610",
			AVer:     "218101901",
			Ct:       0,
		})
		if err != nil {
			return errors.Wrap(err, "发送login消息到wss失败")
		}
		logger.SLog.Debug("发送登录消息")
		// 2. wait login resp
		<-l.entered

		err = conn.SendBinaryMessage(&MsgJoinGroup{
			RoomId: cfg.RoomId,
			GId:    -9999, // 海量弹幕模式
		})
		if err != nil {
			return errors.Wrap(err, "发送joingroup消息到wss失败")
		}
		logger.SLog.Debug("发送入组消息")
	}
	// 心跳
	go l.heartbeat(conn, cfg.HeartbeatInterval*time.Second)
	logger.SLog.Debug("开始心跳...")
	return
}

func (l *LiveDouyu) heartbeat(conn *ws.NWebsocket, t time.Duration) {
	hb := func(conn *ws.NWebsocket) {
		//logger.SLog.Debug("heartbeat !!!")
		data := &MsgHeartbeat{}
		l.sequenceId++
		err := conn.SendBinaryMessage(data)
		if err != nil {
			return
		}
	}
	hb(conn)

	ticker := time.NewTicker(t)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			hb(conn)
		}
	}
}

func (l *LiveDouyu) HandlerMessage(v interface{}) {
	entry, ok := v.(*WsEntry)
	if !ok {
		logger.SLog.Warnf("读取消息错误, 数据类型不匹配 %T %T", v, &WsEntry{})
		return
	}
	//logger.SLog.Debugf("接收消息 %s", string(entry.data))
	switch entry.dataType {
	case "loginres":
		go func() {
			select {
			case l.entered <- struct{}{}:
			}
		}()
	default:
		handler, ok := l.msgHandlerMapper[entry.dataType]
		if !ok {
			return
		}
		handler.HandlerMessage(entry.data)
		//logger.SLog.Infof("接收消息 %v\n%v", string(entry.data), entry.MapData)
	}

}