blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-25 11:06:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

msg update

Browse Source
This commit is contained in:
away 2021-06-28 19:23:43 +08:00
parent 9b4af25e26
commit 473c956b1b
2 changed files with 16 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/msg_transfer/logic/db.go
View File

@ -12,7 +12,7 @@ func saveUserChat(uid string, pbMsg *pbMsg.MsgSvrToPushSvrChatMsg) error {
return err return err
} }
pbMsg.RecvSeq = seq pbMsg.RecvSeq = seq
return db.DB.SaveUserChat(uid, pbMsg) return db.DB.SaveUserChat(uid, pbMsg.SendTime, pbMsg)
} }
func getGroupList(groupID string) ([]string, error) { func getGroupList(groupID string) ([]string, error) {

53
src/msg_transfer/logic/history_msg_handler.go
View File

@ -6,7 +6,6 @@ import (
kfk "Open_IM/src/common/kafka" kfk "Open_IM/src/common/kafka"
"Open_IM/src/common/log" "Open_IM/src/common/log"
pbMsg "Open_IM/src/proto/chat" pbMsg "Open_IM/src/proto/chat"
pb "Open_IM/src/proto/group"
pbPush "Open_IM/src/proto/push" pbPush "Open_IM/src/proto/push"
"Open_IM/src/utils" "Open_IM/src/utils"
"context" "context"
@ -55,10 +54,8 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string)
isHistory := utils.GetSwitchFromOptions(Options, "history") isHistory := utils.GetSwitchFromOptions(Options, "history")
//Control whether to store history messages (mysql) //Control whether to store history messages (mysql)
isPersist := utils.GetSwitchFromOptions(Options, "persistent") isPersist := utils.GetSwitchFromOptions(Options, "persistent")
//Control whether to push message to sender's other terminal
isSenderSync := utils.GetSwitchFromOptions(Options, "senderSync")
if pbData.SessionType == constant.SingleChatType { if pbData.SessionType == constant.SingleChatType {
log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist, isSenderSync) log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist)
if isHistory { if isHistory {
if msgKey == pbSaveData.RecvID { if msgKey == pbSaveData.RecvID {
err := saveUserChat(pbData.RecvID, &pbSaveData) err := saveUserChat(pbData.RecvID, &pbSaveData)
@ -73,58 +70,30 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string)
if err != nil { if err != nil {
log.ErrorByKv("data insert to mongo err", pbSaveData.OperationID, "data", pbSaveData.String(), "err", err.Error()) log.ErrorByKv("data insert to mongo err", pbSaveData.OperationID, "data", pbSaveData.String(), "err", err.Error())
} }
if isSenderSync { //if isSenderSync {
pbSaveData.ContentType = constant.SyncSenderMsg // pbSaveData.ContentType = constant.SyncSenderMsg
log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String()) // log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String())
sendMessageToPush(&pbSaveData) // sendMessageToPush(&pbSaveData)
} //}
} }
} }
log.InfoByKv("msg_transfer handle topic success...", "", "")
} else if pbData.SessionType == constant.GroupChatType { } else if pbData.SessionType == constant.GroupChatType {
log.Info("", "", "msg_transfer chat type = GroupChatType") log.Info("", "", "msg_transfer chat type = GroupChatType")
uidAndGroupID := strings.Split(pbData.RecvID, " ")
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName) saveUserChat(uidAndGroupID[0], &pbSaveData)
client := pb.NewGroupClient(etcdConn)
req := &pb.GetGroupInfoReq{
GroupID: pbSaveData.RecvID,
Token: pbData.Token,
OperationID: pbSaveData.OperationID,
}
log.Info("", "", "msg_transfer call group rpc, data = %s", req.String())
reply, err := client.GetGroupInfo(context.Background(), req)
if err != nil {
log.Error("", "", "msg_transfer client.GetGroupInfo fail, err = %s", err.Error())
return
}
for _, v := range reply.GroupMemberList {
//Store RecvID is userID+" "+groupID when chatType is Group
pbSaveData.RecvID = v.UserID + " " + pbSaveData.RecvID
if isHistory {
saveUserChat(v.UserID, &pbSaveData)
}
pbSaveData.Options = pbData.Options pbSaveData.Options = pbData.Options
pbSaveData.OfflineInfo = pbData.OfflineInfo pbSaveData.OfflineInfo = pbData.OfflineInfo
if v.UserID != pbSaveData.SendID { if utils.IsContain(uidAndGroupID[0], pbData.ForceList) {
if utils.IsContain(v.UserID, pbData.ForceList) {
pbSaveData.IsEmphasize = true pbSaveData.IsEmphasize = true
} }
sendMessageToPush(&pbSaveData) sendMessageToPush(&pbSaveData)
} else { log.InfoByKv("msg_transfer handle topic success...", "", "")
if isSenderSync {
pbSaveData.ContentType = constant.SyncSenderMsg
sendMessageToPush(&pbSaveData)
}
}
}
} else { } else {
log.Error("", "", "msg_transfer recv chat err, chat.MsgFrom = %d", pbData.SessionType) log.Error("", "", "msg_transfer recv chat err, chat.MsgFrom = %d", pbData.SessionType)
} }
log.InfoByKv("msg_transfer handle topic success...", "", "")
} }
func (HistoryConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil } func (HistoryConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }