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

modify msg

Browse Source
This commit is contained in:
wangchuxiao
2023-02-27 12:29:40 +08:00
parent f384bb2aff
commit 59d1090a6d
5 changed files with 88 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/msg_transfer/logic/modify_msg_handler.go
+36
View File
@@ -83,11 +83,47 @@ func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msg
LatestUpdateTime: v.LatestUpdateTime,
}
}
// modify old msg
if msgDataToMQ.MsgData.SessionType == constant.SuperGroupChatType && utils.GetSwitchFromOptions(msgDataToMQ.MsgData.Options, constant.IsHistory) {
if msgDataToMQ.MsgData.Seq == 0 {
log.NewError(msgDataToMQ.OperationID, "seq==0, error msg", msgDataToMQ.MsgData)
continue
}
msg, err := db.DB.GetMsgBySeqIndex(notification.SourceID, notification.Seq, msgDataToMQ.OperationID)
if (msg != nil && msg.Seq != notification.Seq) || err != nil {
if err != nil {
log.NewError(msgDataToMQ.OperationID, "GetMsgBySeqIndex failed", notification, err.Error())
}
msgs, indexes, err := db.DB.GetSuperGroupMsgBySeqListMongo(notification.SourceID, []uint32{notification.Seq}, msgDataToMQ.OperationID)
if err != nil {
log.NewError(msgDataToMQ.OperationID, "GetSuperGroupMsgBySeqListMongo failed", notification, err.Error())
continue
}
var index int
if len(msgs) < 1 || len(indexes) < 1 {
log.NewError(msgDataToMQ.OperationID, "GetSuperGroupMsgBySeqListMongo failed", notification, "len<1", msgs, indexes)
continue
} else {
msg = msgs[0]
index = indexes[msg.Seq]
}
if err := db.DB.ReplaceMsgByIndex(notification.SourceID, msg, index); err != nil {
log.NewError(msgDataToMQ.OperationID, "ReplaceMsgByIndex failed", notification.SourceID, *msg)
}
} else {
msg.IsReact = true
if err = db.DB.ReplaceMsgBySeq(notification.SourceID, msg, msgDataToMQ.OperationID); err != nil {
log.NewError(msgDataToMQ.OperationID, "ReplaceMsgBySeq failed", notification.SourceID, *msg)
}
}
}
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 {
internal/rpc/msg/pull_message.go
+1 -1
View File
@@ -85,7 +85,7 @@ func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *open_im_sdk.Pull
} else {
log.Debug(in.OperationID, "get message from redis is nil", failedSeqList)
}
msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(k, failedSeqList, in.OperationID)
msgList, _, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(k, failedSeqList, in.OperationID)
if err1 != nil {
promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err1.Error())
internal/rpc/msg/query_msg.go
+1 -1
View File
@@ -21,7 +21,7 @@ func (rpc *rpcChat) GetSuperGroupMsg(context context.Context, req *msg.GetSuperG
} else {
log.Debug(req.OperationID, "get message from redis is nil", failedSeqList)
}
msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(req.GroupID, failedSeqList, req.OperationID)
msgList, _, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(req.GroupID, failedSeqList, req.OperationID)
if err1 != nil {
promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
log.Error(req.OperationID, "GetSuperGroupMsg data error", req.String(), err.Error())