mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-01 07:35:58 +08:00
errcode
This commit is contained in:
wangchuxiao
@@ -0,0 +1,29 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/db"
|
||||
"Open_IM/pkg/common/log"
|
||||
pbMsg "Open_IM/pkg/proto/msg"
|
||||
"Open_IM/pkg/utils"
|
||||
)
|
||||
|
||||
func saveUserChat(uid string, msg *pbMsg.MsgDataToMQ) error {
|
||||
time := utils.GetCurrentTimestampByMill()
|
||||
seq, err := db.DB.IncrUserSeq(uid)
|
||||
if err != nil {
|
||||
log.NewError(msg.OperationID, "data insert to redis err", err.Error(), msg.String())
|
||||
return err
|
||||
}
|
||||
msg.MsgData.Seq = uint32(seq)
|
||||
pbSaveData := pbMsg.MsgDataToDB{}
|
||||
pbSaveData.MsgData = msg.MsgData
|
||||
log.NewInfo(msg.OperationID, "IncrUserSeq cost time", utils.GetCurrentTimestampByMill()-time)
|
||||
return db.DB.SaveUserChatMongo2(uid, pbSaveData.MsgData.SendTime, &pbSaveData)
|
||||
// return db.DB.SaveUserChatMongo2(uid, pbSaveData.MsgData.SendTime, &pbSaveData)
|
||||
}
|
||||
|
||||
func saveUserChatList(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) (error, uint64) {
|
||||
log.Info(operationID, utils.GetSelfFuncName(), "args ", userID, len(msgList))
|
||||
//return db.DB.BatchInsertChat(userID, msgList, operationID)
|
||||
return db.DB.BatchInsertChat2Cache(userID, msgList, operationID)
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/config"
|
||||
"Open_IM/pkg/common/constant"
|
||||
"Open_IM/pkg/common/kafka"
|
||||
promePkg "Open_IM/pkg/common/prometheus"
|
||||
"Open_IM/pkg/statistics"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
const OnlineTopicBusy = 1
|
||||
const OnlineTopicVacancy = 0
|
||||
const Msg = 2
|
||||
const ConsumerMsgs = 3
|
||||
const AggregationMessages = 4
|
||||
const MongoMessages = 5
|
||||
const ChannelNum = 100
|
||||
|
||||
var (
|
||||
persistentCH PersistentConsumerHandler
|
||||
historyCH OnlineHistoryRedisConsumerHandler
|
||||
historyMongoCH OnlineHistoryMongoConsumerHandler
|
||||
modifyCH ModifyMsgConsumerHandler
|
||||
producer *kafka.Producer
|
||||
producerToModify *kafka.Producer
|
||||
producerToMongo *kafka.Producer
|
||||
cmdCh chan Cmd2Value
|
||||
onlineTopicStatus int
|
||||
w *sync.Mutex
|
||||
singleMsgSuccessCount uint64
|
||||
groupMsgCount uint64
|
||||
singleMsgFailedCount uint64
|
||||
|
||||
singleMsgSuccessCountMutex sync.Mutex
|
||||
)
|
||||
|
||||
func Init() {
|
||||
cmdCh = make(chan Cmd2Value, 10000)
|
||||
w = new(sync.Mutex)
|
||||
if config.Config.Prometheus.Enable {
|
||||
initPrometheus()
|
||||
}
|
||||
persistentCH.Init() // ws2mschat save mysql
|
||||
historyCH.Init(cmdCh) //
|
||||
historyMongoCH.Init()
|
||||
modifyCH.Init()
|
||||
onlineTopicStatus = OnlineTopicVacancy
|
||||
//offlineHistoryCH.Init(cmdCh)
|
||||
statistics.NewStatistics(&singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
|
||||
statistics.NewStatistics(&groupMsgCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second groupMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
|
||||
producer = kafka.NewKafkaProducer(config.Config.Kafka.Ms2pschat.Addr, config.Config.Kafka.Ms2pschat.Topic)
|
||||
producerToModify = kafka.NewKafkaProducer(config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.MsgToModify.Topic)
|
||||
producerToMongo = kafka.NewKafkaProducer(config.Config.Kafka.MsgToMongo.Addr, config.Config.Kafka.MsgToMongo.Topic)
|
||||
}
|
||||
func Run(promethuesPort int) {
|
||||
//register mysqlConsumerHandler to
|
||||
if config.Config.ChatPersistenceMysql {
|
||||
go persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(&persistentCH)
|
||||
} else {
|
||||
fmt.Println("not start mysql consumer")
|
||||
}
|
||||
go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH)
|
||||
go historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyMongoCH)
|
||||
go modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(&modifyCH)
|
||||
//go offlineHistoryCH.historyConsumerGroup.RegisterHandleAndConsumer(&offlineHistoryCH)
|
||||
go func() {
|
||||
err := promePkg.StartPromeSrv(promethuesPort)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
func SetOnlineTopicStatus(status int) {
|
||||
w.Lock()
|
||||
defer w.Unlock()
|
||||
onlineTopicStatus = status
|
||||
}
|
||||
func GetOnlineTopicStatus() int {
|
||||
w.Lock()
|
||||
defer w.Unlock()
|
||||
return onlineTopicStatus
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/api_struct"
|
||||
"Open_IM/pkg/common/config"
|
||||
"Open_IM/pkg/common/constant"
|
||||
"Open_IM/pkg/common/db"
|
||||
"Open_IM/pkg/common/db/mongo"
|
||||
kfk "Open_IM/pkg/common/kafka"
|
||||
"Open_IM/pkg/common/log"
|
||||
pbMsg "Open_IM/pkg/proto/msg"
|
||||
server_api_params "Open_IM/pkg/proto/sdk_ws"
|
||||
"Open_IM/pkg/utils"
|
||||
"encoding/json"
|
||||
"github.com/Shopify/sarama"
|
||||
|
||||
"github.com/golang/protobuf/proto"
|
||||
)
|
||||
|
||||
type ModifyMsgConsumerHandler struct {
|
||||
msgHandle map[string]fcb
|
||||
modifyMsgConsumerGroup *kfk.MConsumerGroup
|
||||
}
|
||||
|
||||
func (mmc *ModifyMsgConsumerHandler) Init() {
|
||||
mmc.msgHandle = make(map[string]fcb)
|
||||
mmc.msgHandle[config.Config.Kafka.MsgToModify.Topic] = mmc.ModifyMsg
|
||||
mmc.modifyMsgConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
|
||||
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToModify.Topic},
|
||||
config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.ConsumerGroupID.MsgToModify)
|
||||
}
|
||||
|
||||
func (ModifyMsgConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (ModifyMsgConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
claim sarama.ConsumerGroupClaim) error {
|
||||
for msg := range claim.Messages() {
|
||||
log.NewDebug("", "kafka get info to mysql", "ModifyMsgConsumerHandler", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
|
||||
if len(msg.Value) != 0 {
|
||||
mmc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
|
||||
} else {
|
||||
log.Error("", "msg get from kafka but is nil", msg.Key)
|
||||
}
|
||||
sess.MarkMessage(msg, "")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
|
||||
log.NewInfo("msg come here ModifyMsg!!!", "", "msg", string(cMsg.Value), msgKey)
|
||||
msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
|
||||
err := proto.Unmarshal(cMsg.Value, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.NewError(msgFromMQ.TriggerID, "msg_transfer Unmarshal msg err", "msg", string(cMsg.Value), "err", err.Error())
|
||||
return
|
||||
}
|
||||
log.Debug(msgFromMQ.TriggerID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String())
|
||||
for _, msgDataToMQ := range msgFromMQ.MessageList {
|
||||
isReactionFromCache := utils.GetSwitchFromOptions(msgDataToMQ.MsgData.Options, constant.IsReactionFromCache)
|
||||
if !isReactionFromCache {
|
||||
continue
|
||||
}
|
||||
if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageModifier {
|
||||
notification := &apistruct.ReactionMessageModifierNotification{}
|
||||
if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
|
||||
continue
|
||||
}
|
||||
if notification.IsExternalExtensions {
|
||||
log.NewInfo(msgDataToMQ.OperationID, "msg:", notification, "this is external extensions")
|
||||
continue
|
||||
}
|
||||
if !notification.IsReact {
|
||||
// first time to modify
|
||||
var reactionExtensionList = make(map[string]mongoDB.KeyValue)
|
||||
extendMsg := mongoDB.ExtendMsg{
|
||||
ReactionExtensionList: reactionExtensionList,
|
||||
ClientMsgID: notification.ClientMsgID,
|
||||
MsgFirstModifyTime: notification.MsgFirstModifyTime,
|
||||
}
|
||||
for _, v := range notification.SuccessReactionExtensionList {
|
||||
reactionExtensionList[v.TypeKey] = mongoDB.KeyValue{
|
||||
TypeKey: v.TypeKey,
|
||||
Value: v.Value,
|
||||
LatestUpdateTime: v.LatestUpdateTime,
|
||||
}
|
||||
}
|
||||
|
||||
if err := db.DB.InsertExtendMsg(notification.SourceID, notification.SessionType, &extendMsg); err != nil {
|
||||
log.NewError(msgDataToMQ.OperationID, "MsgFirstModify InsertExtendMsg failed", notification.SourceID, notification.SessionType, extendMsg, err.Error())
|
||||
continue
|
||||
}
|
||||
} else {
|
||||
var reactionExtensionList = make(map[string]*server_api_params.KeyValue)
|
||||
for _, v := range notification.SuccessReactionExtensionList {
|
||||
reactionExtensionList[v.TypeKey] = &server_api_params.KeyValue{
|
||||
TypeKey: v.TypeKey,
|
||||
Value: v.Value,
|
||||
LatestUpdateTime: v.LatestUpdateTime,
|
||||
}
|
||||
}
|
||||
// is already modify
|
||||
if err := db.DB.InsertOrUpdateReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, reactionExtensionList); err != nil {
|
||||
log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
|
||||
}
|
||||
}
|
||||
} else if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageDeleter {
|
||||
notification := &apistruct.ReactionMessageDeleteNotification{}
|
||||
if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
|
||||
continue
|
||||
}
|
||||
if err := db.DB.DeleteReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, notification.SuccessReactionExtensionList); err != nil {
|
||||
log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func UnMarshallSetReactionMsgContent(content []byte) (notification *apistruct.ReactionMessageModifierNotification, err error) {
|
||||
|
||||
return notification, nil
|
||||
}
|
||||
@@ -0,0 +1,579 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/config"
|
||||
"Open_IM/pkg/common/constant"
|
||||
"Open_IM/pkg/common/db"
|
||||
kfk "Open_IM/pkg/common/kafka"
|
||||
"Open_IM/pkg/common/log"
|
||||
"Open_IM/pkg/getcdv3"
|
||||
pbMsg "Open_IM/pkg/proto/msg"
|
||||
pbPush "Open_IM/pkg/proto/push"
|
||||
"Open_IM/pkg/utils"
|
||||
"context"
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/golang/protobuf/proto"
|
||||
"hash/crc32"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type MsgChannelValue struct {
|
||||
aggregationID string //maybe userID or super groupID
|
||||
triggerID string
|
||||
msgList []*pbMsg.MsgDataToMQ
|
||||
lastSeq uint64
|
||||
}
|
||||
type TriggerChannelValue struct {
|
||||
triggerID string
|
||||
cmsgList []*sarama.ConsumerMessage
|
||||
}
|
||||
type fcb func(cMsg *sarama.ConsumerMessage, msgKey string, sess sarama.ConsumerGroupSession)
|
||||
type Cmd2Value struct {
|
||||
Cmd int
|
||||
Value interface{}
|
||||
}
|
||||
type OnlineHistoryRedisConsumerHandler struct {
|
||||
msgHandle map[string]fcb
|
||||
historyConsumerGroup *kfk.MConsumerGroup
|
||||
chArrays [ChannelNum]chan Cmd2Value
|
||||
msgDistributionCh chan Cmd2Value
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) Init(cmdCh chan Cmd2Value) {
|
||||
och.msgHandle = make(map[string]fcb)
|
||||
och.msgDistributionCh = make(chan Cmd2Value) //no buffer channel
|
||||
go och.MessagesDistributionHandle()
|
||||
for i := 0; i < ChannelNum; i++ {
|
||||
och.chArrays[i] = make(chan Cmd2Value, 50)
|
||||
go och.Run(i)
|
||||
}
|
||||
if config.Config.ReliableStorage {
|
||||
och.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = och.handleChatWs2Mongo
|
||||
} else {
|
||||
och.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = och.handleChatWs2MongoLowReliability
|
||||
|
||||
}
|
||||
och.historyConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
|
||||
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
|
||||
config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToRedis)
|
||||
|
||||
}
|
||||
func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
|
||||
for {
|
||||
select {
|
||||
case cmd := <-och.chArrays[channelID]:
|
||||
switch cmd.Cmd {
|
||||
case AggregationMessages:
|
||||
msgChannelValue := cmd.Value.(MsgChannelValue)
|
||||
msgList := msgChannelValue.msgList
|
||||
triggerID := msgChannelValue.triggerID
|
||||
storageMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
|
||||
notStoragePushMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
|
||||
log.Debug(triggerID, "msg arrived channel", "channel id", channelID, msgList, msgChannelValue.aggregationID, len(msgList))
|
||||
var modifyMsgList []*pbMsg.MsgDataToMQ
|
||||
for _, v := range msgList {
|
||||
log.Debug(triggerID, "msg come to storage center", v.String())
|
||||
isHistory := utils.GetSwitchFromOptions(v.MsgData.Options, constant.IsHistory)
|
||||
isSenderSync := utils.GetSwitchFromOptions(v.MsgData.Options, constant.IsSenderSync)
|
||||
if isHistory {
|
||||
storageMsgList = append(storageMsgList, v)
|
||||
//log.NewWarn(triggerID, "storageMsgList to mongodb client msgID: ", v.MsgData.ClientMsgID)
|
||||
} else {
|
||||
if !(!isSenderSync && msgChannelValue.aggregationID == v.MsgData.SendID) {
|
||||
notStoragePushMsgList = append(notStoragePushMsgList, v)
|
||||
}
|
||||
}
|
||||
|
||||
if v.MsgData.ContentType == constant.ReactionMessageModifier || v.MsgData.ContentType == constant.ReactionMessageDeleter {
|
||||
modifyMsgList = append(modifyMsgList, v)
|
||||
}
|
||||
}
|
||||
if len(modifyMsgList) > 0 {
|
||||
sendMessageToModifyMQ(msgChannelValue.aggregationID, triggerID, modifyMsgList)
|
||||
}
|
||||
//switch msgChannelValue.msg.MsgData.SessionType {
|
||||
//case constant.SingleChatType:
|
||||
//case constant.GroupChatType:
|
||||
//case constant.NotificationChatType:
|
||||
//default:
|
||||
// log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String())
|
||||
// return
|
||||
//}
|
||||
log.Debug(triggerID, "msg storage length", len(storageMsgList), "push length", len(notStoragePushMsgList))
|
||||
if len(storageMsgList) > 0 {
|
||||
err, lastSeq := saveUserChatList(msgChannelValue.aggregationID, storageMsgList, triggerID)
|
||||
if err != nil {
|
||||
singleMsgFailedCount += uint64(len(storageMsgList))
|
||||
log.NewError(triggerID, "single data insert to redis err", err.Error(), storageMsgList)
|
||||
} else {
|
||||
singleMsgSuccessCountMutex.Lock()
|
||||
singleMsgSuccessCount += uint64(len(storageMsgList))
|
||||
singleMsgSuccessCountMutex.Unlock()
|
||||
och.SendMessageToMongoCH(msgChannelValue.aggregationID, triggerID, storageMsgList, lastSeq)
|
||||
|
||||
for _, v := range storageMsgList {
|
||||
sendMessageToPushMQ(v, msgChannelValue.aggregationID)
|
||||
}
|
||||
for _, x := range notStoragePushMsgList {
|
||||
sendMessageToPushMQ(x, msgChannelValue.aggregationID)
|
||||
}
|
||||
}
|
||||
|
||||
} else {
|
||||
for _, x := range notStoragePushMsgList {
|
||||
sendMessageToPushMQ(x, msgChannelValue.aggregationID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) SendMessageToMongoCH(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ, lastSeq uint64) {
|
||||
if len(messages) > 0 {
|
||||
pid, offset, err := producerToMongo.SendMessage(&pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
|
||||
if err != nil {
|
||||
log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
|
||||
} else {
|
||||
// log.NewWarn(m.OperationID, "sendMsgToKafka client msgID ", m.MsgData.ClientMsgID)
|
||||
}
|
||||
}
|
||||
//hashCode := getHashCode(aggregationID)
|
||||
//channelID := hashCode % ChannelNum
|
||||
//log.Debug(triggerID, "generate channelID", hashCode, channelID, aggregationID)
|
||||
////go func(cID uint32, userID string, messages []*pbMsg.MsgDataToMQ) {
|
||||
//och.chMongoArrays[channelID] <- Cmd2Value{Cmd: MongoMessages, Value: MsgChannelValue{aggregationID: aggregationID, msgList: messages, triggerID: triggerID, lastSeq: lastSeq}}
|
||||
}
|
||||
|
||||
//func (och *OnlineHistoryRedisConsumerHandler) MongoMessageRun(channelID int) {
|
||||
// for {
|
||||
// select {
|
||||
// case cmd := <-och.chMongoArrays[channelID]:
|
||||
// switch cmd.Cmd {
|
||||
// case MongoMessages:
|
||||
// msgChannelValue := cmd.Value.(MsgChannelValue)
|
||||
// msgList := msgChannelValue.msgList
|
||||
// triggerID := msgChannelValue.triggerID
|
||||
// aggregationID := msgChannelValue.aggregationID
|
||||
// lastSeq := msgChannelValue.lastSeq
|
||||
// err := db.DB.BatchInsertChat2DB(aggregationID, msgList, triggerID, lastSeq)
|
||||
// if err != nil {
|
||||
// log.NewError(triggerID, "single data insert to mongo err", err.Error(), msgList)
|
||||
// }
|
||||
// for _, v := range msgList {
|
||||
// if v.MsgData.ContentType == constant.DeleteMessageNotification {
|
||||
// tips := server_api_params.TipsComm{}
|
||||
// DeleteMessageTips := server_api_params.DeleteMessageTips{}
|
||||
// err := proto.Unmarshal(v.MsgData.Content, &tips)
|
||||
// if err != nil {
|
||||
// log.NewError(triggerID, "tips unmarshal err:", err.Error(), v.String())
|
||||
// continue
|
||||
// }
|
||||
// err = proto.Unmarshal(tips.Detail, &DeleteMessageTips)
|
||||
// if err != nil {
|
||||
// log.NewError(triggerID, "deleteMessageTips unmarshal err:", err.Error(), v.String())
|
||||
// continue
|
||||
// }
|
||||
// if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
|
||||
// log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
|
||||
// }
|
||||
//
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
//}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
|
||||
for {
|
||||
aggregationMsgs := make(map[string][]*pbMsg.MsgDataToMQ, ChannelNum)
|
||||
select {
|
||||
case cmd := <-och.msgDistributionCh:
|
||||
switch cmd.Cmd {
|
||||
case ConsumerMsgs:
|
||||
triggerChannelValue := cmd.Value.(TriggerChannelValue)
|
||||
triggerID := triggerChannelValue.triggerID
|
||||
consumerMessages := triggerChannelValue.cmsgList
|
||||
//Aggregation map[userid]message list
|
||||
log.Debug(triggerID, "batch messages come to distribution center", len(consumerMessages))
|
||||
for i := 0; i < len(consumerMessages); i++ {
|
||||
msgFromMQ := pbMsg.MsgDataToMQ{}
|
||||
err := proto.Unmarshal(consumerMessages[i].Value, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.Error(triggerID, "msg_transfer Unmarshal msg err", "msg", string(consumerMessages[i].Value), "err", err.Error())
|
||||
return
|
||||
}
|
||||
log.Debug(triggerID, "single msg come to distribution center", msgFromMQ.String(), string(consumerMessages[i].Key))
|
||||
if oldM, ok := aggregationMsgs[string(consumerMessages[i].Key)]; ok {
|
||||
oldM = append(oldM, &msgFromMQ)
|
||||
aggregationMsgs[string(consumerMessages[i].Key)] = oldM
|
||||
} else {
|
||||
m := make([]*pbMsg.MsgDataToMQ, 0, 100)
|
||||
m = append(m, &msgFromMQ)
|
||||
aggregationMsgs[string(consumerMessages[i].Key)] = m
|
||||
}
|
||||
}
|
||||
log.Debug(triggerID, "generate map list users len", len(aggregationMsgs))
|
||||
for aggregationID, v := range aggregationMsgs {
|
||||
if len(v) >= 0 {
|
||||
hashCode := getHashCode(aggregationID)
|
||||
channelID := hashCode % ChannelNum
|
||||
log.Debug(triggerID, "generate channelID", hashCode, channelID, aggregationID)
|
||||
//go func(cID uint32, userID string, messages []*pbMsg.MsgDataToMQ) {
|
||||
och.chArrays[channelID] <- Cmd2Value{Cmd: AggregationMessages, Value: MsgChannelValue{aggregationID: aggregationID, msgList: v, triggerID: triggerID}}
|
||||
//}(channelID, userID, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
func (mc *OnlineHistoryRedisConsumerHandler) handleChatWs2Mongo(cMsg *sarama.ConsumerMessage, msgKey string, sess sarama.ConsumerGroupSession) {
|
||||
msg := cMsg.Value
|
||||
now := time.Now()
|
||||
msgFromMQ := pbMsg.MsgDataToMQ{}
|
||||
err := proto.Unmarshal(msg, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.Error("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
|
||||
return
|
||||
}
|
||||
operationID := msgFromMQ.OperationID
|
||||
log.NewInfo(operationID, "msg come mongo!!!", "", "msg", string(msg))
|
||||
//Control whether to store offline messages (mongo)
|
||||
isHistory := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsHistory)
|
||||
//Control whether to store history messages (mysql)
|
||||
isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
|
||||
isSenderSync := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsSenderSync)
|
||||
switch msgFromMQ.MsgData.SessionType {
|
||||
case constant.SingleChatType:
|
||||
log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = SingleChatType", isHistory, isPersist)
|
||||
if isHistory {
|
||||
err := saveUserChat(msgKey, &msgFromMQ)
|
||||
if err != nil {
|
||||
singleMsgFailedCount++
|
||||
log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
|
||||
return
|
||||
}
|
||||
singleMsgSuccessCountMutex.Lock()
|
||||
singleMsgSuccessCount++
|
||||
singleMsgSuccessCountMutex.Unlock()
|
||||
log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
|
||||
}
|
||||
if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {
|
||||
} else {
|
||||
go sendMessageToPush(&msgFromMQ, msgKey)
|
||||
}
|
||||
log.NewDebug(operationID, "saveSingleMsg cost time ", time.Since(now))
|
||||
case constant.GroupChatType:
|
||||
log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = GroupChatType", isHistory, isPersist)
|
||||
if isHistory {
|
||||
err := saveUserChat(msgFromMQ.MsgData.RecvID, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.NewError(operationID, "group data insert to mongo err", msgFromMQ.String(), msgFromMQ.MsgData.RecvID, err.Error())
|
||||
return
|
||||
}
|
||||
groupMsgCount++
|
||||
}
|
||||
go sendMessageToPush(&msgFromMQ, msgFromMQ.MsgData.RecvID)
|
||||
log.NewDebug(operationID, "saveGroupMsg cost time ", time.Since(now))
|
||||
|
||||
case constant.NotificationChatType:
|
||||
log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = NotificationChatType", isHistory, isPersist)
|
||||
if isHistory {
|
||||
err := saveUserChat(msgKey, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
|
||||
return
|
||||
}
|
||||
log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
|
||||
}
|
||||
if !isSenderSync && msgKey == msgFromMQ.MsgData.SendID {
|
||||
} else {
|
||||
go sendMessageToPush(&msgFromMQ, msgKey)
|
||||
}
|
||||
log.NewDebug(operationID, "saveUserChat cost time ", time.Since(now))
|
||||
default:
|
||||
log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String())
|
||||
return
|
||||
}
|
||||
sess.MarkMessage(cMsg, "")
|
||||
log.NewDebug(msgFromMQ.OperationID, "msg_transfer handle topic data to database success...", msgFromMQ.String())
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) handleChatWs2MongoLowReliability(cMsg *sarama.ConsumerMessage, msgKey string, sess sarama.ConsumerGroupSession) {
|
||||
msg := cMsg.Value
|
||||
msgFromMQ := pbMsg.MsgDataToMQ{}
|
||||
err := proto.Unmarshal(msg, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.Error("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
|
||||
return
|
||||
}
|
||||
operationID := msgFromMQ.OperationID
|
||||
log.NewInfo(operationID, "msg come mongo!!!", "", "msg", string(msg))
|
||||
//Control whether to store offline messages (mongo)
|
||||
isHistory := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsHistory)
|
||||
isSenderSync := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsSenderSync)
|
||||
if isHistory {
|
||||
seq, err := db.DB.IncrUserSeq(msgKey)
|
||||
if err != nil {
|
||||
log.NewError(operationID, "data insert to redis err", err.Error(), string(msg))
|
||||
return
|
||||
}
|
||||
sess.MarkMessage(cMsg, "")
|
||||
msgFromMQ.MsgData.Seq = uint32(seq)
|
||||
log.Debug(operationID, "send ch msg is ", msgFromMQ.String())
|
||||
//och.msgCh <- Cmd2Value{Cmd: Msg, Value: MsgChannelValue{msgKey, msgFromMQ}}
|
||||
//err := saveUserChat(msgKey, &msgFromMQ)
|
||||
//if err != nil {
|
||||
// singleMsgFailedCount++
|
||||
// log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
|
||||
// return
|
||||
//}
|
||||
//singleMsgSuccessCountMutex.Lock()
|
||||
//singleMsgSuccessCount++
|
||||
//singleMsgSuccessCountMutex.Unlock()
|
||||
//log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", time.Since(now))
|
||||
} else {
|
||||
if !(!isSenderSync && msgKey == msgFromMQ.MsgData.SendID) {
|
||||
go sendMessageToPush(&msgFromMQ, msgKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (OnlineHistoryRedisConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (OnlineHistoryRedisConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
|
||||
//func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
// claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
|
||||
// log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
|
||||
// for msg := range claim.Messages() {
|
||||
// SetOnlineTopicStatus(OnlineTopicBusy)
|
||||
// //och.TriggerCmd(OnlineTopicBusy)
|
||||
// log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
|
||||
// och.msgHandle[msg.Topic](msg, string(msg.Key), sess)
|
||||
// if claim.HighWaterMarkOffset()-msg.Offset <= 1 {
|
||||
// log.Debug("", "online msg consume end", claim.HighWaterMarkOffset(), msg.Offset)
|
||||
// SetOnlineTopicStatus(OnlineTopicVacancy)
|
||||
// och.TriggerCmd(OnlineTopicVacancy)
|
||||
// }
|
||||
// }
|
||||
// return nil
|
||||
//}
|
||||
|
||||
func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
|
||||
|
||||
for {
|
||||
if sess == nil {
|
||||
log.NewWarn("", " sess == nil, waiting ")
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
rwLock := new(sync.RWMutex)
|
||||
log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
|
||||
cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
t := time.NewTicker(time.Duration(100) * time.Millisecond)
|
||||
var triggerID string
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
//case :
|
||||
// triggerID = utils.OperationIDGenerator()
|
||||
//
|
||||
// log.NewDebug(triggerID, "claim.Messages ", msg)
|
||||
// cMsg = append(cMsg, msg)
|
||||
// if len(cMsg) >= 1000 {
|
||||
// ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// for _, v := range cMsg {
|
||||
// ccMsg = append(ccMsg, v)
|
||||
// }
|
||||
// log.Debug(triggerID, "length trigger msg consumer start", len(ccMsg))
|
||||
// och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
|
||||
// triggerID: triggerID, cmsgList: ccMsg}}
|
||||
// sess.MarkMessage(msg, "")
|
||||
// cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// log.Debug(triggerID, "length trigger msg consumer end", len(cMsg))
|
||||
// }
|
||||
|
||||
case <-t.C:
|
||||
if len(cMsg) > 0 {
|
||||
rwLock.Lock()
|
||||
ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
for _, v := range cMsg {
|
||||
ccMsg = append(ccMsg, v)
|
||||
}
|
||||
cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
rwLock.Unlock()
|
||||
split := 1000
|
||||
triggerID = utils.OperationIDGenerator()
|
||||
log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
|
||||
for i := 0; i < len(ccMsg)/split; i++ {
|
||||
//log.Debug()
|
||||
och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
|
||||
triggerID: triggerID, cmsgList: ccMsg[i*split : (i+1)*split]}}
|
||||
}
|
||||
if (len(ccMsg) % split) > 0 {
|
||||
och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
|
||||
triggerID: triggerID, cmsgList: ccMsg[split*(len(ccMsg)/split):]}}
|
||||
}
|
||||
//sess.MarkMessage(ccMsg[len(cMsg)-1], "")
|
||||
|
||||
log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}()
|
||||
for msg := range claim.Messages() {
|
||||
//msgFromMQ := pbMsg.MsgDataToMQ{}
|
||||
//err := proto.Unmarshal(msg.Value, &msgFromMQ)
|
||||
//if err != nil {
|
||||
// log.Error(triggerID, "msg_transfer Unmarshal msg err", "msg", string(msg.Value), "err", err.Error())
|
||||
//}
|
||||
//userID := string(msg.Key)
|
||||
//hashCode := getHashCode(userID)
|
||||
//channelID := hashCode % ChannelNum
|
||||
//log.Debug(triggerID, "generate channelID", hashCode, channelID, userID)
|
||||
////go func(cID uint32, userID string, messages []*pbMsg.MsgDataToMQ) {
|
||||
//och.chArrays[channelID] <- Cmd2Value{Cmd: UserMessages, Value: MsgChannelValue{userID: userID, msgList: []*pbMsg.MsgDataToMQ{&msgFromMQ}, triggerID: msgFromMQ.OperationID}}
|
||||
//sess.MarkMessage(msg, "")
|
||||
rwLock.Lock()
|
||||
if len(msg.Value) != 0 {
|
||||
cMsg = append(cMsg, msg)
|
||||
}
|
||||
rwLock.Unlock()
|
||||
sess.MarkMessage(msg, "")
|
||||
//och.TriggerCmd(OnlineTopicBusy)
|
||||
|
||||
//log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
|
||||
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
//func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
// claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
|
||||
//
|
||||
// for {
|
||||
// if sess == nil {
|
||||
// log.NewWarn("", " sess == nil, waiting ")
|
||||
// time.Sleep(100 * time.Millisecond)
|
||||
// } else {
|
||||
// break
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
|
||||
// cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// t := time.NewTicker(time.Duration(100) * time.Millisecond)
|
||||
// var triggerID string
|
||||
// for msg := range claim.Messages() {
|
||||
// cMsg = append(cMsg, msg)
|
||||
// //och.TriggerCmd(OnlineTopicBusy)
|
||||
// select {
|
||||
// //case :
|
||||
// // triggerID = utils.OperationIDGenerator()
|
||||
// //
|
||||
// // log.NewDebug(triggerID, "claim.Messages ", msg)
|
||||
// // cMsg = append(cMsg, msg)
|
||||
// // if len(cMsg) >= 1000 {
|
||||
// // ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// // for _, v := range cMsg {
|
||||
// // ccMsg = append(ccMsg, v)
|
||||
// // }
|
||||
// // log.Debug(triggerID, "length trigger msg consumer start", len(ccMsg))
|
||||
// // och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
|
||||
// // triggerID: triggerID, cmsgList: ccMsg}}
|
||||
// // sess.MarkMessage(msg, "")
|
||||
// // cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// // log.Debug(triggerID, "length trigger msg consumer end", len(cMsg))
|
||||
// // }
|
||||
//
|
||||
// case <-t.C:
|
||||
// if len(cMsg) > 0 {
|
||||
// ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// for _, v := range cMsg {
|
||||
// ccMsg = append(ccMsg, v)
|
||||
// }
|
||||
// triggerID = utils.OperationIDGenerator()
|
||||
// log.Debug(triggerID, "timer trigger msg consumer start", len(ccMsg))
|
||||
// och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
|
||||
// triggerID: triggerID, cmsgList: ccMsg}}
|
||||
// sess.MarkMessage(cMsg[len(cMsg)-1], "")
|
||||
// cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
|
||||
// log.Debug(triggerID, "timer trigger msg consumer end", len(cMsg))
|
||||
// }
|
||||
// default:
|
||||
//
|
||||
// }
|
||||
// //log.NewDebug("", "online kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "online", msg.Offset, claim.HighWaterMarkOffset())
|
||||
//
|
||||
// }
|
||||
// return nil
|
||||
//}
|
||||
|
||||
func sendMessageToPush(message *pbMsg.MsgDataToMQ, pushToUserID string) {
|
||||
log.Info(message.OperationID, "msg_transfer send message to push", "message", message.String())
|
||||
rpcPushMsg := pbPush.PushMsgReq{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
|
||||
mqPushMsg := pbMsg.PushMsgDataToMQ{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
|
||||
grpcConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImPushName, message.OperationID)
|
||||
if grpcConn != nil {
|
||||
log.Error(rpcPushMsg.OperationID, "rpc dial failed", "push data", rpcPushMsg.String())
|
||||
pid, offset, err := producer.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
|
||||
if err != nil {
|
||||
log.Error(mqPushMsg.OperationID, "kafka send failed", "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
|
||||
}
|
||||
return
|
||||
}
|
||||
msgClient := pbPush.NewPushMsgServiceClient(grpcConn)
|
||||
_, err := msgClient.PushMsg(context.Background(), &rpcPushMsg)
|
||||
if err != nil {
|
||||
log.Error(rpcPushMsg.OperationID, "rpc send failed", rpcPushMsg.OperationID, "push data", rpcPushMsg.String(), "err", err.Error())
|
||||
pid, offset, err := producer.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
|
||||
if err != nil {
|
||||
log.Error(message.OperationID, "kafka send failed", mqPushMsg.OperationID, "send data", mqPushMsg.String(), "pid", pid, "offset", offset, "err", err.Error())
|
||||
}
|
||||
} else {
|
||||
log.Info(message.OperationID, "rpc send success", rpcPushMsg.OperationID, "push data", rpcPushMsg.String())
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
func sendMessageToPushMQ(message *pbMsg.MsgDataToMQ, pushToUserID string) {
|
||||
log.Info(message.OperationID, utils.GetSelfFuncName(), "msg ", message.String(), pushToUserID)
|
||||
rpcPushMsg := pbPush.PushMsgReq{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
|
||||
mqPushMsg := pbMsg.PushMsgDataToMQ{OperationID: message.OperationID, MsgData: message.MsgData, PushToUserID: pushToUserID}
|
||||
pid, offset, err := producer.SendMessage(&mqPushMsg, mqPushMsg.PushToUserID, rpcPushMsg.OperationID)
|
||||
if err != nil {
|
||||
log.Error(mqPushMsg.OperationID, "kafka send failed", "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func sendMessageToModifyMQ(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ) {
|
||||
if len(messages) > 0 {
|
||||
pid, offset, err := producerToModify.SendMessage(&pbMsg.MsgDataToModifyByMQ{AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
|
||||
if err != nil {
|
||||
log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
|
||||
} else {
|
||||
// log.NewWarn(m.OperationID, "sendMsgToKafka client msgID ", m.MsgData.ClientMsgID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// String hashes a string to a unique hashcode.
|
||||
//
|
||||
// crc32 returns a uint32, but for our use we need
|
||||
// and non negative integer. Here we cast to an integer
|
||||
// and invert it if the result is negative.
|
||||
func getHashCode(s string) uint32 {
|
||||
return crc32.ChecksumIEEE([]byte(s))
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/config"
|
||||
"Open_IM/pkg/common/constant"
|
||||
"Open_IM/pkg/common/db"
|
||||
kfk "Open_IM/pkg/common/kafka"
|
||||
"Open_IM/pkg/common/log"
|
||||
pbMsg "Open_IM/pkg/proto/msg"
|
||||
server_api_params "Open_IM/pkg/proto/sdk_ws"
|
||||
"Open_IM/pkg/utils"
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/golang/protobuf/proto"
|
||||
)
|
||||
|
||||
type OnlineHistoryMongoConsumerHandler struct {
|
||||
msgHandle map[string]fcb
|
||||
historyConsumerGroup *kfk.MConsumerGroup
|
||||
}
|
||||
|
||||
func (och *OnlineHistoryMongoConsumerHandler) Init() {
|
||||
och.msgHandle = make(map[string]fcb)
|
||||
och.msgHandle[config.Config.Kafka.MsgToMongo.Topic] = och.handleChatWs2Mongo
|
||||
och.historyConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
|
||||
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToMongo.Topic},
|
||||
config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMongo)
|
||||
|
||||
}
|
||||
func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
|
||||
msg := cMsg.Value
|
||||
msgFromMQ := pbMsg.MsgDataToMongoByMQ{}
|
||||
err := proto.Unmarshal(msg, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.Error("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
|
||||
return
|
||||
}
|
||||
log.Info(msgFromMQ.TriggerID, "BatchInsertChat2DB userID: ", msgFromMQ.AggregationID, "msgFromMQ.LastSeq: ", msgFromMQ.LastSeq)
|
||||
err = db.DB.BatchInsertChat2DB(msgFromMQ.AggregationID, msgFromMQ.MessageList, msgFromMQ.TriggerID, msgFromMQ.LastSeq)
|
||||
if err != nil {
|
||||
log.NewError(msgFromMQ.TriggerID, "single data insert to mongo err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
|
||||
} else {
|
||||
err = db.DB.DeleteMessageFromCache(msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.GetTriggerID())
|
||||
if err != nil {
|
||||
log.NewError(msgFromMQ.TriggerID, "remove cache msg from redis err", err.Error(), msgFromMQ.MessageList, msgFromMQ.AggregationID, msgFromMQ.TriggerID)
|
||||
}
|
||||
}
|
||||
for _, v := range msgFromMQ.MessageList {
|
||||
if v.MsgData.ContentType == constant.DeleteMessageNotification {
|
||||
tips := server_api_params.TipsComm{}
|
||||
DeleteMessageTips := server_api_params.DeleteMessageTips{}
|
||||
err := proto.Unmarshal(v.MsgData.Content, &tips)
|
||||
if err != nil {
|
||||
log.NewError(msgFromMQ.TriggerID, "tips unmarshal err:", err.Error(), v.String())
|
||||
continue
|
||||
}
|
||||
err = proto.Unmarshal(tips.Detail, &DeleteMessageTips)
|
||||
if err != nil {
|
||||
log.NewError(msgFromMQ.TriggerID, "deleteMessageTips unmarshal err:", err.Error(), v.String())
|
||||
continue
|
||||
}
|
||||
if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
|
||||
log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (OnlineHistoryMongoConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
|
||||
func (och *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
claim sarama.ConsumerGroupClaim) error { // a instance in the consumer group
|
||||
log.NewDebug("", "online new session msg come", claim.HighWaterMarkOffset(), claim.Topic(), claim.Partition())
|
||||
for msg := range claim.Messages() {
|
||||
log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
|
||||
if len(msg.Value) != 0 {
|
||||
och.msgHandle[msg.Topic](msg, string(msg.Key), sess)
|
||||
} else {
|
||||
log.Error("", "mongo msg get from kafka but is nil", msg.Key)
|
||||
}
|
||||
sess.MarkMessage(msg, "")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
/*
|
||||
** description("").
|
||||
** copyright('tuoyun,www.tuoyun.net').
|
||||
** author("fg,Gordon@tuoyun.net").
|
||||
** time(2021/5/11 15:37).
|
||||
*/
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/pkg/common/config"
|
||||
"Open_IM/pkg/common/constant"
|
||||
"Open_IM/pkg/common/db/mysql_model/im_mysql_msg_model"
|
||||
kfk "Open_IM/pkg/common/kafka"
|
||||
"Open_IM/pkg/common/log"
|
||||
pbMsg "Open_IM/pkg/proto/msg"
|
||||
"Open_IM/pkg/utils"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/golang/protobuf/proto"
|
||||
|
||||
promePkg "Open_IM/pkg/common/prometheus"
|
||||
)
|
||||
|
||||
type PersistentConsumerHandler struct {
|
||||
msgHandle map[string]fcb
|
||||
persistentConsumerGroup *kfk.MConsumerGroup
|
||||
}
|
||||
|
||||
func (pc *PersistentConsumerHandler) Init() {
|
||||
pc.msgHandle = make(map[string]fcb)
|
||||
pc.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = pc.handleChatWs2Mysql
|
||||
pc.persistentConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
|
||||
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
|
||||
config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql)
|
||||
|
||||
}
|
||||
|
||||
func initPrometheus() {
|
||||
promePkg.NewSeqGetSuccessCounter()
|
||||
promePkg.NewSeqGetFailedCounter()
|
||||
promePkg.NewSeqSetSuccessCounter()
|
||||
promePkg.NewSeqSetFailedCounter()
|
||||
promePkg.NewMsgInsertRedisSuccessCounter()
|
||||
promePkg.NewMsgInsertRedisFailedCounter()
|
||||
promePkg.NewMsgInsertMongoSuccessCounter()
|
||||
promePkg.NewMsgInsertMongoFailedCounter()
|
||||
}
|
||||
|
||||
func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
|
||||
msg := cMsg.Value
|
||||
log.NewInfo("msg come here mysql!!!", "", "msg", string(msg), msgKey)
|
||||
var tag bool
|
||||
msgFromMQ := pbMsg.MsgDataToMQ{}
|
||||
err := proto.Unmarshal(msg, &msgFromMQ)
|
||||
if err != nil {
|
||||
log.NewError(msgFromMQ.OperationID, "msg_transfer Unmarshal msg err", "msg", string(msg), "err", err.Error())
|
||||
return
|
||||
}
|
||||
log.Debug(msgFromMQ.OperationID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String())
|
||||
//Control whether to store history messages (mysql)
|
||||
isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
|
||||
//Only process receiver data
|
||||
if isPersist {
|
||||
switch msgFromMQ.MsgData.SessionType {
|
||||
case constant.SingleChatType, constant.NotificationChatType:
|
||||
if msgKey == msgFromMQ.MsgData.RecvID {
|
||||
tag = true
|
||||
}
|
||||
case constant.GroupChatType:
|
||||
if msgKey == msgFromMQ.MsgData.SendID {
|
||||
tag = true
|
||||
}
|
||||
case constant.SuperGroupChatType:
|
||||
tag = true
|
||||
}
|
||||
if tag {
|
||||
log.NewInfo(msgFromMQ.OperationID, "msg_transfer msg persisting", string(msg))
|
||||
if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil {
|
||||
log.NewError(msgFromMQ.OperationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String())
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
func (PersistentConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (PersistentConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
claim sarama.ConsumerGroupClaim) error {
|
||||
for msg := range claim.Messages() {
|
||||
log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
|
||||
if len(msg.Value) != 0 {
|
||||
pc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
|
||||
} else {
|
||||
log.Error("", "msg get from kafka but is nil", msg.Key)
|
||||
}
|
||||
sess.MarkMessage(msg, "")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user