diff --git a/internal/msg_gateway/gate/batch_push.go b/internal/msg_gateway/gate/batch_push.go index 089c91e4c..075de0586 100644 --- a/internal/msg_gateway/gate/batch_push.go +++ b/internal/msg_gateway/gate/batch_push.go @@ -2,10 +2,12 @@ package gate import ( "Open_IM/pkg/common/config" + "Open_IM/pkg/common/db" "Open_IM/pkg/common/log" "Open_IM/pkg/grpc-etcdv3/getcdv3" pbChat "Open_IM/pkg/proto/chat" sdk_ws "Open_IM/pkg/proto/sdk_ws" + "Open_IM/pkg/utils" "context" "strings" ) @@ -13,24 +15,62 @@ import ( var MaxPullMsgNum = 100 func (r *RPCServer) GenPullSeqList(currentSeq uint32, operationID string, userID string) ([]uint32, error) { - return nil, nil + maxSeq, err := db.DB.GetUserMaxSeq(userID) + if err != nil { + log.Error(operationID, "GetUserMaxSeq failed ", userID, err.Error()) + return nil, utils.Wrap(err, "") + } + + var seqList []uint32 + num := 0 + for i := currentSeq + 1; i < uint32(maxSeq); i++ { + seqList = append(seqList, i) + num++ + if num == MaxPullMsgNum { + break + } + } + log.Info(operationID, "GenPullSeqList ", seqList, "current seq", currentSeq) + return seqList, nil } -func (r *RPCServer) GetMergeSingleMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserID string, platformID int) []*sdk_ws.MsgData { - return nil - //ws.getUserConn(pushToUserID, platformID) - //msgData.Seq - //msgList := r.GetSingleMsgForPush(operationID, msgData, pushToUserID, platformID) +func (r *RPCServer) GetSingleUserMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserID string, platformID int) []*sdk_ws.MsgData { + userConn := ws.getUserConn(pushToUserID, platformID) + if userConn == nil { + return []*sdk_ws.MsgData{msgData} + } + if msgData.Seq <= userConn.PushedMaxSeq { + return nil + } + + msgList := r.GetSingleUserMsg(operationID, msgData.Seq, pushToUserID) + if msgList == nil { + userConn.PushedMaxSeq = msgData.Seq + return []*sdk_ws.MsgData{msgData} + } + msgList = append(msgList, msgData) + + for _, v := range msgList { + if v.Seq > userConn.PushedMaxSeq { + userConn.PushedMaxSeq = v.Seq + } + } + return msgList } -func (r *RPCServer) GetSingleMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserID string, platformID string) []*sdk_ws.MsgData { - seqList, err := r.GenPullSeqList(msgData.Seq, operationID, pushToUserID) + +func (r *RPCServer) GetSingleUserMsg(operationID string, currentMsgSeq uint32, userID string) []*sdk_ws.MsgData { + seqList, err := r.GenPullSeqList(currentMsgSeq, operationID, userID) if err != nil { - log.Error(operationID, "GenPullSeqList failed ", err.Error(), msgData.Seq, pushToUserID) + log.Error(operationID, "GenPullSeqList failed ", err.Error(), currentMsgSeq, userID) + return nil + } + if len(seqList) == 0 { + log.Error(operationID, "GenPullSeqList len == 0 ", currentMsgSeq, userID) return nil } rpcReq := sdk_ws.PullMessageBySeqListReq{} rpcReq.SeqList = seqList - rpcReq.UserID = pushToUserID + rpcReq.UserID = userID rpcReq.OperationID = operationID grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName) msgClient := pbChat.NewChatClient(grpcConn) @@ -45,10 +85,10 @@ func (r *RPCServer) GetSingleMsgForPush(operationID string, msgData *sdk_ws.MsgD return reply.List } -func (r *RPCServer) GetBatchMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserIDList []string, platformID string) map[string][]*sdk_ws.MsgData { - return nil -} - -func (r *RPCServer) GetMaxSeq(userID string) (uint32, error) { - return 0, nil +func (r *RPCServer) GetBatchUserMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserIDList []string, platformID int) map[string][]*sdk_ws.MsgData { + user2PushMsg := make(map[string][]*sdk_ws.MsgData, 0) + for _, v := range pushToUserIDList { + user2PushMsg[v] = r.GetSingleUserMsgForPush(operationID, msgData, v, platformID) + } + return user2PushMsg }