From 9eb0dd5e704c1abd46bf77ab083b9a407f62b18f Mon Sep 17 00:00:00 2001
From: skiffer-git <44203734@qq.com>
Date: Thu, 9 Jun 2022 12:49:16 +0800
Subject: [PATCH] batch push

---
 internal/msg_gateway/gate/batch_push.go | 39 +++++++++++++++++++++++++
 internal/msg_gateway/gate/rpc_server.go |  5 ----
 internal/msg_gateway/gate/ws_server.go  |  2 +-
 internal/rpc/friend/firend.go           |  2 +-
 4 files changed, 41 insertions(+), 7 deletions(-)
 create mode 100644 internal/msg_gateway/gate/batch_push.go

diff --git a/internal/msg_gateway/gate/batch_push.go b/internal/msg_gateway/gate/batch_push.go
new file mode 100644
index 000000000..c373fa996
--- /dev/null
+++ b/internal/msg_gateway/gate/batch_push.go
@@ -0,0 +1,39 @@
+package gate
+
+import (
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/grpc-etcdv3/getcdv3"
+	pbChat "Open_IM/pkg/proto/chat"
+	sdk_ws "Open_IM/pkg/proto/sdk_ws"
+	"strings"
+)
+
+var MaxPullMsgNum = 100
+
+func (r *RPCServer) GenPullSeqList(currentSeq uint32, operationID string, userID string) ([]uint32, error) {
+
+}
+
+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)
+	if err != nil {
+		return nil
+	}
+	rpcReq := sdk_ws.PullMessageBySeqListReq{}
+	rpcReq.SeqList = seqList
+	rpcReq.UserID = pushToUserID
+	rpcReq.OperationID = operationID
+	grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
+	msgClient := pbChat.NewChatClient(grpcConn)
+	reply, err := msgClient.PullMessageBySeqList(context.Background(), &rpcReq)
+
+	return nil
+}
+
+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
+}
diff --git a/internal/msg_gateway/gate/rpc_server.go b/internal/msg_gateway/gate/rpc_server.go
index 30cca624b..cc16f1226 100644
--- a/internal/msg_gateway/gate/rpc_server.go
+++ b/internal/msg_gateway/gate/rpc_server.go
@@ -7,7 +7,6 @@ import (
 	"Open_IM/pkg/common/token_verify"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	pbRelay "Open_IM/pkg/proto/relay"
-	sdk_ws "Open_IM/pkg/proto/sdk_ws"
 	"Open_IM/pkg/utils"
 	"bytes"
 	"context"
@@ -144,10 +143,6 @@ func (r *RPCServer) GetUsersOnlineStatus(_ context.Context, req *pbRelay.GetUser
 	return &resp, nil
 }
 
-func (r *RPCServer) GetBatchMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserIDList []string, platformID string) map[string][]*sdk_ws.MsgData {
-	return nil
-}
-
 func (r *RPCServer) OnlineBatchPushOneMsg(_ context.Context, req *pbRelay.OnlineBatchPushOneMsgReq) (*pbRelay.OnlineBatchPushOneMsgResp, error) {
 	log.NewInfo(req.OperationID, "BatchPushMsgToUser is arriving", req.String())
 	var singleUserResult []*pbRelay.SingelMsgToUserResultList
diff --git a/internal/msg_gateway/gate/ws_server.go b/internal/msg_gateway/gate/ws_server.go
index 20bd3d484..690e57cd3 100644
--- a/internal/msg_gateway/gate/ws_server.go
+++ b/internal/msg_gateway/gate/ws_server.go
@@ -61,7 +61,7 @@ func (ws *WServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 			//Connection mapping relationship,
 			//userID+" "+platformID->conn
 			//Initialize a lock for each user
-			newConn := &UserConn{conn, new(sync.Mutex)}
+			newConn := &UserConn{conn, new(sync.Mutex), 0}
 			userCount++
 			ws.addUserConn(query["sendID"][0], utils.StringToInt(query["platformID"][0]), newConn, query["token"][0])
 			go ws.readMsg(newConn)
diff --git a/internal/rpc/friend/firend.go b/internal/rpc/friend/firend.go
index 313b2becf..4e3172b44 100644
--- a/internal/rpc/friend/firend.go
+++ b/internal/rpc/friend/firend.go
@@ -63,7 +63,7 @@ func (s *friendServer) Run() {
 	defer srv.GracefulStop()
 	//User friend related services register to etcd
 	pbFriend.RegisterFriendServer(srv, s)
-	rpcRegisterIP := ""
+	rpcRegisterIP := config.Config.RpcRegisterIP
 	if config.Config.RpcRegisterIP == "" {
 		rpcRegisterIP, err = utils.GetLocalIP()
 		if err != nil {