live-gateway/bilibili/msg_handler/guard_buy.go

package msg_handler

import (
	"encoding/json"
	"git.noahlan.cn/northlan/ntools-go/kafka"
	"git.noahlan.cn/northlan/ntools-go/logger"
	"live-gateway/config"
	"live-gateway/pb"
	kfk "live-gateway/pkg/kafka"
	"strconv"
)

type GuardBuy struct {
	Uid        int64  `json:"uid"`         // 用户ID
	Uname      string `json:"username"`    // 用户名
	GuardLevel int64  `json:"guard_level"` // 等级(3舰长 1总督)
	Num        int32  `json:"num"`         // 数量
	Price      int64  `json:"price"`       // 价格 金瓜子
	GiftId     int64  `json:"gift_id"`     // 礼物ID
	GiftName   string `json:"gift_name"`   // 礼物名
	StartTime  int64  `json:"start_time"`  // 开始时间
	EndTime    int64  `json:"end_time"`    // 结束时间
}

type GuardBuyHandler struct {
	producer   *kafka.Producer
	liveRoomId int64
}

func NewGuardBuyHandler(liveRoomId int64) *GuardBuyHandler {
	cfg := config.Config.Kafka.GuardBuy
	return &GuardBuyHandler{
		producer:   kafka.NewKafkaProducer(kfk.DefaultProducerConfig, cfg.Addr, cfg.Topic),
		liveRoomId: liveRoomId,
	}
}

func (h *GuardBuyHandler) CMD() string {
	return "GUARD_BUY"
}

func (h *GuardBuyHandler) HandlerMessage(data []byte) {
	var baseMsg struct {
		CMD  string    `json:"cmd"`
		Data *GuardBuy `json:"data"`
	}
	if err := json.Unmarshal(data, &baseMsg); err != nil {
		return
	}

	logger.SLog.Infof("%s 购买 %s x %d", baseMsg.Data.Uname, baseMsg.Data.GiftName, baseMsg.Data.Num)

	dmMsg := &pbMq.MqNobilityBuy{
		Platform:      pbMq.Platform_name[int32(pbMq.Platform_bilibili)],
		LiveRoomId:    h.liveRoomId,
		Uid:           baseMsg.Data.Uid,
		Uname:         baseMsg.Data.Uname,
		NobilityLevel: baseMsg.Data.GuardLevel,
		Price:         baseMsg.Data.Price,
		StartTime:     baseMsg.Data.StartTime,
		EndTime:       baseMsg.Data.EndTime,
	}

	_ = h.producer.SendMessageAsync(dmMsg, strconv.FormatInt(dmMsg.Uid, 10))
}