blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-10-09 10:46:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

not storage message push fix bug

Browse Source
This commit is contained in:
Gordon 2022-07-25 12:53:49 +08:00
parent f3ed48e0ac
commit a71b42c335
1 changed files with 33 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
internal/msg_transfer/logic/online_history_msg_handler.go
View File

@ -97,6 +97,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
// return // return
//} //}
log.Debug(triggerID, "msg storage length", len(storageMsgList), "push length", len(notStoragePushMsgList)) log.Debug(triggerID, "msg storage length", len(storageMsgList), "push length", len(notStoragePushMsgList))
if len(storageMsgList) > 0 {
err, lastSeq := saveUserChatList(msgChannelValue.aggregationID, storageMsgList, triggerID) err, lastSeq := saveUserChatList(msgChannelValue.aggregationID, storageMsgList, triggerID)
if err != nil { if err != nil {
singleMsgFailedCount += uint64(len(storageMsgList)) singleMsgFailedCount += uint64(len(storageMsgList))
@ -117,17 +118,30 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
}(notStoragePushMsgList, storageMsgList) }(notStoragePushMsgList, storageMsgList)
} }
} else {
go func(push []*pbMsg.MsgDataToMQ) {
for _, x := range push {
sendMessageToPush(x, msgChannelValue.aggregationID)
}
}(notStoragePushMsgList)
} }
} }
} }
}
} }
func (och *OnlineHistoryRedisConsumerHandler) SendMessageToMongoCH(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ, lastSeq uint64) { 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) pid, offset, err := producerToMongo.SendMessage(&pbMsg.MsgDataToMongoByMQ{LastSeq: lastSeq, AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
if err != nil { if err != nil {
log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID) log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
} else { } else {
// log.NewWarn(m.OperationID, "sendMsgToKafka client msgID ", m.MsgData.ClientMsgID) // log.NewWarn(m.OperationID, "sendMsgToKafka client msgID ", m.MsgData.ClientMsgID)
} }
}
//hashCode := getHashCode(aggregationID) //hashCode := getHashCode(aggregationID)
//channelID := hashCode % ChannelNum //channelID := hashCode % ChannelNum
//log.Debug(triggerID, "generate channelID", hashCode, channelID, aggregationID) //log.Debug(triggerID, "generate channelID", hashCode, channelID, aggregationID)