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

Merge branch 'superGroup' of github.com:OpenIMSDK/Open-IM-Server into superGroup

Browse Source
This commit is contained in:
wangchuxiao
2022-06-13 10:38:15 +08:00
parent dbc055efb4 af612e4b92
commit 7d6701780c
6 changed files with 36 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/demo/register/onboarding_process.go
-1
View File
@@ -14,7 +14,6 @@ import (
commonPb "Open_IM/pkg/proto/sdk_ws"
"Open_IM/pkg/utils"
"context"
"encoding/json"
"errors"
"fmt"
"github.com/golang/protobuf/proto"
internal/msg_gateway/gate/batch_push.go
+7
View File
@@ -44,17 +44,23 @@ func (r *RPCServer) GetSingleUserMsgForPushPlatforms(operationID string, msgData
}
func (r *RPCServer) GetSingleUserMsgForPush(operationID string, msgData *sdk_ws.MsgData, pushToUserID string, platformID int) []*sdk_ws.MsgData {
msgData.MsgDataList = nil
return []*sdk_ws.MsgData{msgData}
userConn := ws.getUserConn(pushToUserID, platformID)
if userConn == nil {
log.Debug(operationID, "userConn == nil")
return []*sdk_ws.MsgData{msgData}
}
if msgData.Seq <= userConn.PushedMaxSeq {
log.Debug(operationID, "msgData.Seq <= userConn.PushedMaxSeq", msgData.Seq, userConn.PushedMaxSeq)
return nil
}
msgList := r.GetSingleUserMsg(operationID, msgData.Seq, pushToUserID)
if msgList == nil {
log.Debug(operationID, "GetSingleUserMsg msgList == nil", msgData.Seq, userConn.PushedMaxSeq)
userConn.PushedMaxSeq = msgData.Seq
return []*sdk_ws.MsgData{msgData}
}
@@ -65,6 +71,7 @@ func (r *RPCServer) GetSingleUserMsgForPush(operationID string, msgData *sdk_ws.
userConn.PushedMaxSeq = v.Seq
}
}
log.Debug(operationID, "GetSingleUserMsg msgList len ", len(msgList), userConn.PushedMaxSeq)
return msgList
}
internal/msg_gateway/gate/rpc_server.go
+20 -6
View File
@@ -205,20 +205,28 @@ func (r *RPCServer) OnlineBatchPushOneMsg(_ context.Context, req *pbRelay.Online
for k, _ := range userConnMap {
platformList = append(platformList, k)
}
log.Debug(req.OperationID, "GetSingleUserMsgForPushPlatforms begin", req.MsgData.Seq, v, platformList)
log.Debug(req.OperationID, "GetSingleUserMsgForPushPlatforms begin", req.MsgData.Seq, v, platformList, req.MsgData.String())
needPushMapList := r.GetSingleUserMsgForPushPlatforms(req.OperationID, req.MsgData, v, platformList)
log.Debug(req.OperationID, "GetSingleUserMsgForPushPlatforms ", req.MsgData.Seq, v, platformList, len(needPushMapList))
log.Debug(req.OperationID, "GetSingleUserMsgForPushPlatforms end", req.MsgData.Seq, v, platformList, len(needPushMapList))
for platform, list := range needPushMapList {
if list != nil {
log.Debug(req.OperationID, "GetSingleUserMsgForPushPlatforms ", "userID: ", v, "platform: ", platform, "push msg num:", len(list))
log.Debug(req.OperationID, "needPushMapList ", "userID: ", v, "platform: ", platform, "push msg num:", len(list))
for _, v := range list {
log.Debug(req.OperationID, "req.MsgData.MsgDataList begin", "len: ", len(req.MsgData.MsgDataList), v.String())
req.MsgData.MsgDataList = append(req.MsgData.MsgDataList, v)
log.Debug(req.OperationID, "req.MsgData.MsgDataList end", "len: ", len(req.MsgData.MsgDataList))
}
log.Debug(req.OperationID, "r.encodeWsData no string")
log.Debug(req.OperationID, "r.encodeWsData data0 list ", req.MsgData.MsgDataList[0].String())
log.Debug(req.OperationID, "r.encodeWsData ", req.MsgData.String())
replyBytes, err := r.encodeWsData(req.MsgData, req.OperationID)
if err != nil {
log.Error(req.OperationID, "encodeWsData failed ", req.MsgData.String())
continue
}
log.Debug(req.OperationID, "encodeWsData", "len: ", replyBytes.Len())
resultCode := sendMsgBatchToUser(userConnMap[platform], replyBytes.Bytes(), req, platform, v)
if resultCode == 0 && utils.IsContainInt(platform, r.pushTerminal) {
tempT.OnlinePush = true
@@ -250,7 +258,13 @@ func (r *RPCServer) OnlineBatchPushOneMsg(_ context.Context, req *pbRelay.Online
}, nil
}
func (r *RPCServer) encodeWsData(wsData *sdk_ws.MsgData, operationID string) (bytes.Buffer, error) {
msgBytes, _ := proto.Marshal(wsData)
log.Debug(operationID, "encodeWsData begin", wsData.String())
msgBytes, err := proto.Marshal(wsData)
if err != nil {
log.NewError(operationID, "Marshal", err.Error())
return bytes.Buffer{}, utils.Wrap(err, "")
}
log.Debug(operationID, "encodeWsData begin", wsData.String())
mReply := Resp{
ReqIdentifier: constant.WSPushMsg,
OperationID: operationID,
@@ -258,10 +272,10 @@ func (r *RPCServer) encodeWsData(wsData *sdk_ws.MsgData, operationID string) (by
}
var replyBytes bytes.Buffer
enc := gob.NewEncoder(&replyBytes)
err := enc.Encode(mReply)
err = enc.Encode(mReply)
if err != nil {
log.NewError(operationID, "data encode err", err.Error())
return bytes.Buffer{}, err
return bytes.Buffer{}, utils.Wrap(err, "")
}
return replyBytes, nil
}
internal/push/logic/push_to_client.go
+1 -1
View File
@@ -47,7 +47,7 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
grpcCons = getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
}
//Online push message
log.Debug("test", pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String())
log.Debug(pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String())
for _, v := range grpcCons {
msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v)
reply, err := msgClient.OnlineBatchPushOneMsg(context.Background(), &pbRelay.OnlineBatchPushOneMsgReq{OperationID: pushMsg.OperationID, MsgData: pushMsg.MsgData, PushToUserIDList: []string{pushMsg.PushToUserID}})