IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-12 13:05:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

kafka key

Browse Source
This commit is contained in:
wangchuxiao
2023-05-19 19:43:43 +08:00
parent ca2eee17bc
commit 5fab4d3138
15 changed files with 501 additions and 667 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/msgtransfer/init.go
+1 -1
View File
@@ -87,7 +87,7 @@ func (m *MsgTransfer) Start(prometheusPort int) error {
wg.Add(1)
fmt.Println("start msg transfer", "prometheusPort:", prometheusPort)
if config.Config.ChatPersistenceMysql {
go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(m.persistentCH)
// go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(m.persistentCH)
} else {
fmt.Println("msg transfer not start mysql consumer")
}
internal/msgtransfer/modify_msg_handler.go
+5 -8
View File
@@ -53,15 +53,14 @@ func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
}
func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
log.NewInfo("msg come here ModifyMsg!!!", "", "msg", string(cMsg.Value), msgKey)
msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
operationID := mcontext.GetOperationID(ctx)
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())
log.ZError(ctx, "msg_transfer Unmarshal msg err", err, "msg", string(cMsg.Value))
return
}
log.Debug(msgFromMQ.TriggerID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String())
log.ZDebug(ctx, "proto.Unmarshal MsgDataToMQ", "msgs", msgFromMQ.String())
for _, msg := range msgFromMQ.Messages {
isReactionFromCache := utils.GetSwitchFromOptions(msg.Options, constant.IsReactionFromCache)
if !isReactionFromCache {
@@ -74,7 +73,6 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
continue
}
if notification.IsExternalExtensions {
log.NewInfo(operationID, "msg:", notification, "this is external extensions")
continue
}
if !notification.IsReact {
@@ -94,12 +92,12 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
}
if err := mmc.extendMsgDatabase.InsertExtendMsg(ctx, notification.ConversationID, notification.SessionType, &extendMsg); err != nil {
log.NewError(operationID, "MsgFirstModify InsertExtendMsg failed", notification.ConversationID, notification.SessionType, extendMsg, err.Error())
// log.ZError(ctx, "MsgFirstModify InsertExtendMsg failed", notification.ConversationID, notification.SessionType, extendMsg, err.Error())
continue
}
} else {
if err := mmc.extendMsgDatabase.InsertOrUpdateReactionExtendMsgSet(ctx, notification.ConversationID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
// log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
}
}
} else if msg.ContentType == constant.ReactionMessageDeleter {
@@ -108,9 +106,8 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(ctx context.Context, cMsg *sarama
continue
}
if err := mmc.extendMsgDatabase.DeleteReactionExtendMsgSet(ctx, notification.ConversationID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, mmc.extendSetMsgModel.Pb2Model(notification.SuccessReactionExtensions)); err != nil {
log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
// log.NewError(operationID, "InsertOrUpdateReactionExtendMsgSet failed")
}
}
}
}
internal/msgtransfer/online_history_msg_handler.go
+24 -22
View File
@@ -29,9 +29,9 @@ const MongoMessages = 5
const ChannelNum = 100
type MsgChannelValue struct {
conversationID string //maybe userID or super groupID
ctx context.Context
ctxMsgList []*ContextMsg
uniqueKey string
ctx context.Context
ctxMsgList []*ContextMsg
}
type TriggerChannelValue struct {
@@ -90,14 +90,16 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
msgChannelValue := cmd.Value.(MsgChannelValue)
ctxMsgList := msgChannelValue.ctxMsgList
ctx := msgChannelValue.ctx
log.ZDebug(ctx, "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMsgList), "conversationID", msgChannelValue.conversationID)
log.ZDebug(ctx, "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMsgList), "uniqueKey", msgChannelValue.uniqueKey)
storageMsgList, notStorageMsgList, storageNotificationList, notStorageNotificationList, modifyMsgList := och.getPushStorageMsgList(ctxMsgList)
log.ZDebug(ctx, "msg lens", "storageMsgList", len(storageMsgList), "notStorageMsgList", len(notStorageMsgList),
"storageNotificationList", len(storageNotificationList), "notStorageNotificationList", len(notStorageNotificationList), "modifyMsgList", len(modifyMsgList))
och.handleMsg(ctx, utils.GetChatConversationIDByMsg(ctxMsgList[0].message), storageMsgList, notStorageMsgList)
och.handleNotification(ctx, utils.GetNotificationConversationID(ctxMsgList[0].message), storageNotificationList, notStorageNotificationList)
if err := och.msgDatabase.MsgToModifyMQ(ctx, msgChannelValue.conversationID, modifyMsgList); err != nil {
log.ZError(ctx, "msg to modify mq error", err, "conversationID", msgChannelValue.conversationID, "modifyMsgList", modifyMsgList)
conversationIDMsg := utils.GetChatConversationIDByMsg(ctxMsgList[0].message)
conversationIDNotification := utils.GetNotificationConversationID(ctxMsgList[0].message)
och.handleMsg(ctx, msgChannelValue.uniqueKey, conversationIDMsg, storageMsgList, notStorageMsgList)
och.handleNotification(ctx, msgChannelValue.uniqueKey, conversationIDNotification, storageNotificationList, notStorageNotificationList)
if err := och.msgDatabase.MsgToModifyMQ(ctx, msgChannelValue.uniqueKey, conversationIDNotification, modifyMsgList); err != nil {
log.ZError(ctx, "msg to modify mq error", err, "uniqueKey", msgChannelValue.uniqueKey, "modifyMsgList", modifyMsgList)
}
}
}
@@ -156,8 +158,8 @@ func (och *OnlineHistoryRedisConsumerHandler) getPushStorageMsgList(totalMsgs []
return
}
func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
och.toPushTopic(ctx, conversationID, notStorageList)
func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Context, key, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
och.toPushTopic(ctx, key, conversationID, notStorageList)
if len(storageList) > 0 {
lastSeq, _, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
if err != nil {
@@ -165,19 +167,19 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(ctx context.Con
return
}
log.ZDebug(ctx, "success to next topic", "conversationID", conversationID)
och.msgDatabase.MsgToMongoMQ(ctx, conversationID, storageList, lastSeq)
och.toPushTopic(ctx, conversationID, storageList)
och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
och.toPushTopic(ctx, key, conversationID, storageList)
}
}
func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) {
func (och *OnlineHistoryRedisConsumerHandler) toPushTopic(ctx context.Context, key, conversationID string, msgs []*sdkws.MsgData) {
for _, v := range msgs {
och.msgDatabase.MsgToPushMQ(ctx, conversationID, v)
och.msgDatabase.MsgToPushMQ(ctx, key, conversationID, v)
}
}
func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
och.toPushTopic(ctx, conversationID, notStorageList)
func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key, conversationID string, storageList, notStorageList []*sdkws.MsgData) {
och.toPushTopic(ctx, key, conversationID, notStorageList)
if len(storageList) > 0 {
lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
if err != nil && errs.Unwrap(err) != redis.Nil {
@@ -209,8 +211,8 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, con
och.singleMsgSuccessCountMutex.Lock()
och.singleMsgSuccessCount += uint64(len(storageList))
och.singleMsgSuccessCountMutex.Unlock()
och.msgDatabase.MsgToMongoMQ(ctx, conversationID, storageList, lastSeq)
och.toPushTopic(ctx, conversationID, storageList)
och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
och.toPushTopic(ctx, key, conversationID, storageList)
}
}
@@ -253,12 +255,12 @@ func (och *OnlineHistoryRedisConsumerHandler) MessagesDistributionHandle() {
}
}
log.ZDebug(ctx, "generate map list users len", "length", len(aggregationMsgs))
for conversationID, v := range aggregationMsgs {
for uniqueKey, v := range aggregationMsgs {
if len(v) >= 0 {
hashCode := utils.GetHashCode(conversationID)
hashCode := utils.GetHashCode(uniqueKey)
channelID := hashCode % ChannelNum
log.ZDebug(ctx, "generate channelID", "hashCode", hashCode, "channelID", channelID, "conversationID", conversationID)
och.chArrays[channelID] <- Cmd2Value{Cmd: SourceMessages, Value: MsgChannelValue{conversationID: conversationID, ctxMsgList: v, ctx: ctx}}
log.ZDebug(ctx, "generate channelID", "hashCode", hashCode, "channelID", channelID, "uniqueKey", uniqueKey)
och.chArrays[channelID] <- Cmd2Value{Cmd: SourceMessages, Value: MsgChannelValue{uniqueKey: uniqueKey, ctxMsgList: v, ctx: ctx}}
}
}
}
internal/msgtransfer/online_msg_to_mongo_handler.go
+2 -2
View File
@@ -45,11 +45,11 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
log.ZInfo(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.MsgData)
err = mc.msgDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq)
if err != nil {
log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID, "triggerID", msgFromMQ.TriggerID)
log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
}
err = mc.msgDatabase.DeleteMessageFromCache(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData)
if err != nil {
log.ZError(ctx, "remove cache msg from redis err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID, "triggerID", msgFromMQ.TriggerID)
log.ZError(ctx, "remove cache msg from redis err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
}
for _, v := range msgFromMQ.MsgData {
if v.ContentType == constant.DeleteMessageNotification {