batch push

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 {
+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
-	//ws.getUserConn(pushToUserID, platformID)
-	//msgData.Seq
-	//msgList := r.GetSingleMsgForPush(operationID, msgData, pushToUserID, platformID)
+	}
+
+	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
 }