|
|
|
|
package msg_transfer
|
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
//import (
|
|
|
|
|
// "dcg/config"
|
|
|
|
|
// "dcg/game/manager"
|
|
|
|
|
// pbCommon "dcg/game/pb/common"
|
|
|
|
|
// pbMq "dcg/game/pb/mq"
|
|
|
|
|
// "dcg/game/svc"
|
|
|
|
|
// kfk "dcg/pkg/kafka"
|
|
|
|
|
// "git.noahlan.cn/northlan/ntools-go/kafka"
|
|
|
|
|
// "git.noahlan.cn/northlan/ntools-go/logger"
|
|
|
|
|
// "github.com/Shopify/sarama"
|
|
|
|
|
// "google.golang.org/protobuf/proto"
|
|
|
|
|
//)
|
|
|
|
|
//
|
|
|
|
|
//type RewardTransferHandler struct {
|
|
|
|
|
// svcCtx *svc.ServiceContext
|
|
|
|
|
// msgHandle map[string]kafkaMsgHandlerFunc
|
|
|
|
|
// ConsumerGroup *kafka.ConsumerGroup
|
|
|
|
|
//}
|
|
|
|
|
//
|
|
|
|
|
//func (h *RewardTransferHandler) Init(svcCtx *svc.ServiceContext) {
|
|
|
|
|
// h.svcCtx = svcCtx
|
|
|
|
|
// cfg := config.Config.Kafka.RewardPool
|
|
|
|
|
// h.msgHandle = make(map[string]kafkaMsgHandlerFunc)
|
|
|
|
|
// h.msgHandle[cfg.Topic] = h.handleTopic
|
|
|
|
|
//
|
|
|
|
|
// var err error
|
|
|
|
|
// h.ConsumerGroup, err = kafka.NewConsumerGroup(&kafka.ConsumerGroupConfig{
|
|
|
|
|
// KafkaVersion: sarama.V3_1_0_0,
|
|
|
|
|
// OffsetsInitial: sarama.OffsetNewest,
|
|
|
|
|
// IsReturnErr: false,
|
|
|
|
|
// UnMarshaler: kfk.ProtobufMarshaler,
|
|
|
|
|
// }, cfg.Addr, []string{cfg.Topic}, cfg.ConsumerGroup)
|
|
|
|
|
//
|
|
|
|
|
// if err != nil {
|
|
|
|
|
// logger.SLog.Error(err)
|
|
|
|
|
// }
|
|
|
|
|
//}
|
|
|
|
|
//
|
|
|
|
|
//func (h *RewardTransferHandler) handleTopic(data []byte, _ string) {
|
|
|
|
|
// // msg proto
|
|
|
|
|
// var msgFromMq pbMq.MqRewardPool
|
|
|
|
|
// err := proto.Unmarshal(data, &msgFromMq)
|
|
|
|
|
// if err != nil {
|
|
|
|
|
// logger.SLog.Error("unmarshal msg err", err)
|
|
|
|
|
// return
|
|
|
|
|
// }
|
|
|
|
|
// room, err := manager.GameManager.RoomByBattleId(msgFromMq.BattleId)
|
|
|
|
|
// if err != nil {
|
|
|
|
|
// return
|
|
|
|
|
// }
|
|
|
|
|
// room.Broadcast("game.rewardPool", &pbCommon.RewardPoolMsg{
|
|
|
|
|
// WelfarePool: msgFromMq.WelfarePool,
|
|
|
|
|
// BattleId: msgFromMq.BattleId,
|
|
|
|
|
// InitReward: msgFromMq.InitReward,
|
|
|
|
|
// GiftReward: msgFromMq.GiftReward,
|
|
|
|
|
// BattleReward: msgFromMq.BattleReward,
|
|
|
|
|
// OtherReward: msgFromMq.OtherReward,
|
|
|
|
|
// AllRewards: msgFromMq.AllRewards,
|
|
|
|
|
// })
|
|
|
|
|
//}
|
|
|
|
|
//
|
|
|
|
|
//func (RewardTransferHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
|
|
|
|
//func (RewardTransferHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
|
|
|
|
//func (h *RewardTransferHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
|
|
|
|
|
// for msg := range claim.Messages() {
|
|
|
|
|
// //logger.SLog.Infow("kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
|
|
|
|
|
// if hFunc, ok := h.msgHandle[msg.Topic]; ok {
|
|
|
|
|
// hFunc(msg.Value, string(msg.Key))
|
|
|
|
|
// }
|
|
|
|
|
// sess.MarkMessage(msg, "")
|
|
|
|
|
// }
|
|
|
|
|
// return nil
|
|
|
|
|
//}
|
