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

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

Browse Source
This commit is contained in:
wangchuxiao
2023-06-07 15:06:55 +08:00
parent 74ea708893 b8e469fb2e
commit 6360980b66
13 changed files with 1439 additions and 1272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/msggateway/client.go
+16 -16
View File
@@ -45,10 +45,10 @@ type PongHandler func(string) error
type Client struct {
w *sync.Mutex
conn LongConn
platformID int
isCompress bool
userID string
isBackground bool
PlatformID int `json:"platformID"`
IsCompress bool `json:"isCompress"`
UserID string `json:"userID"`
IsBackground bool `json:"isBackground"`
ctx *UserConnContext
longConnServer LongConnServer
closed bool
@@ -59,21 +59,21 @@ func newClient(ctx *UserConnContext, conn LongConn, isCompress bool) *Client {
return &Client{
w: new(sync.Mutex),
conn: conn,
platformID: utils.StringToInt(ctx.GetPlatformID()),
isCompress: isCompress,
userID: ctx.GetUserID(),
PlatformID: utils.StringToInt(ctx.GetPlatformID()),
IsCompress: isCompress,
UserID: ctx.GetUserID(),
ctx: ctx,
}
}
func (c *Client) ResetClient(ctx *UserConnContext, conn LongConn, isCompress bool, longConnServer LongConnServer) {
c.w = new(sync.Mutex)
c.conn = conn
c.platformID = utils.StringToInt(ctx.GetPlatformID())
c.isCompress = isCompress
c.userID = ctx.GetUserID()
c.PlatformID = utils.StringToInt(ctx.GetPlatformID())
c.IsCompress = isCompress
c.UserID = ctx.GetUserID()
c.ctx = ctx
c.longConnServer = longConnServer
c.isBackground = false
c.IsBackground = false
c.closed = false
c.closedErr = nil
}
@@ -126,7 +126,7 @@ func (c *Client) readMessage() {
}
func (c *Client) handleMessage(message []byte) error {
if c.isCompress {
if c.IsCompress {
var decompressErr error
message, decompressErr = c.longConnServer.DeCompress(message)
if decompressErr != nil {
@@ -141,10 +141,10 @@ func (c *Client) handleMessage(message []byte) error {
if err := c.longConnServer.Validate(binaryReq); err != nil {
return utils.Wrap(err, "")
}
if binaryReq.SendID != c.userID {
if binaryReq.SendID != c.UserID {
return utils.Wrap(errors.New("exception conn userID not same to req userID"), binaryReq.String())
}
ctx := mcontext.WithMustInfoCtx([]string{binaryReq.OperationID, binaryReq.SendID, constant.PlatformIDToName(c.platformID), c.ctx.GetConnID()})
ctx := mcontext.WithMustInfoCtx([]string{binaryReq.OperationID, binaryReq.SendID, constant.PlatformIDToName(c.PlatformID), c.ctx.GetConnID()})
log.ZDebug(ctx, "gateway req message", "req", binaryReq.String())
var messageErr error
var resp []byte
@@ -173,7 +173,7 @@ func (c *Client) setAppBackgroundStatus(ctx context.Context, req Req) ([]byte, e
if messageErr != nil {
return nil, messageErr
}
c.isBackground = isBackground
c.IsBackground = isBackground
//todo callback
return resp, nil
@@ -240,7 +240,7 @@ func (c *Client) writeBinaryMsg(resp Resp) error {
return utils.Wrap(err, "")
}
_ = c.conn.SetWriteDeadline(writeWait)
if c.isCompress {
if c.IsCompress {
var compressErr error
resultBuf, compressErr = c.longConnServer.Compress(encodedBuf)
if compressErr != nil {
internal/msggateway/hub_server.go
+5 -5
View File
@@ -60,10 +60,10 @@ func (s *Server) GetUsersOnlineStatus(ctx context.Context, req *msggateway.GetUs
for _, client := range clients {
if client != nil {
ps := new(msggateway.GetUsersOnlineStatusResp_SuccessDetail)
ps.Platform = constant.PlatformIDToName(client.platformID)
ps.Platform = constant.PlatformIDToName(client.PlatformID)
ps.Status = constant.OnlineStatus
ps.ConnID = client.ctx.GetConnID()
ps.IsBackground = client.isBackground
ps.IsBackground = client.IsBackground
temp.Status = constant.OnlineStatus
temp.DetailPlatformStatus = append(temp.DetailPlatformStatus, ps)
}
@@ -98,15 +98,15 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msgga
if client != nil {
temp := &msggateway.SingleMsgToUserPlatform{
RecvID: v,
RecvPlatFormID: int32(client.platformID),
RecvPlatFormID: int32(client.PlatformID),
}
if !client.isBackground {
if !client.IsBackground {
err := client.PushMessage(ctx, req.MsgData)
if err != nil {
temp.ResultCode = -2
resp = append(resp, temp)
} else {
if utils.IsContainInt(client.platformID, s.pushTerminal) {
if utils.IsContainInt(client.PlatformID, s.pushTerminal) {
tempT.OnlinePush = true
prome.Inc(prome.MsgOnlinePushSuccessCounter)
resp = append(resp, temp)
internal/msggateway/n_ws_server.go
+8 -8
View File
@@ -123,22 +123,22 @@ func (ws *WsServer) registerClient(client *Client) {
clientOK bool
cli []*Client
)
cli, userOK, clientOK = ws.clients.Get(client.userID, client.platformID)
cli, userOK, clientOK = ws.clients.Get(client.UserID, client.PlatformID)
if !userOK {
log.ZDebug(client.ctx, "user not exist", "userID", client.userID, "platformID", client.platformID)
ws.clients.Set(client.userID, client)
log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID)
ws.clients.Set(client.UserID, client)
atomic.AddInt64(&ws.onlineUserNum, 1)
atomic.AddInt64(&ws.onlineUserConnNum, 1)
} else {
log.ZDebug(client.ctx, "user exist", "userID", client.userID, "platformID", client.platformID)
log.ZDebug(client.ctx, "user exist", "userID", client.UserID, "platformID", client.PlatformID)
if clientOK { //已经有同平台的连接存在
ws.clients.Set(client.userID, client)
ws.clients.Set(client.UserID, client)
ws.multiTerminalLoginChecker(cli)
log.ZInfo(client.ctx, "repeat login", "userID", client.userID, "platformID", client.platformID, "old remote addr", getRemoteAdders(cli))
log.ZInfo(client.ctx, "repeat login", "userID", client.UserID, "platformID", client.PlatformID, "old remote addr", getRemoteAdders(cli))
atomic.AddInt64(&ws.onlineUserConnNum, 1)
} else {
ws.clients.Set(client.userID, client)
ws.clients.Set(client.UserID, client)
atomic.AddInt64(&ws.onlineUserConnNum, 1)
}
}
@@ -161,7 +161,7 @@ func (ws *WsServer) multiTerminalLoginChecker(client []*Client) {
}
func (ws *WsServer) unregisterClient(client *Client) {
defer ws.clientPool.Put(client)
isDeleteUser := ws.clients.delete(client.userID, client.ctx.GetRemoteAddr())
isDeleteUser := ws.clients.delete(client.UserID, client.ctx.GetRemoteAddr())
if isDeleteUser {
atomic.AddInt64(&ws.onlineUserNum, -1)
}
internal/msggateway/user_map.go
+1 -1
View File
@@ -25,7 +25,7 @@ func (u *UserMap) Get(key string, platformID int) ([]*Client, bool, bool) {
if userExisted {
var clients []*Client
for _, client := range allClients.([]*Client) {
if client.platformID == platformID {
if client.PlatformID == platformID {
clients = append(clients, client)
}
}
internal/rpc/group/group.go
+26 -14
View File
@@ -26,7 +26,6 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/tokenverify"
"github.com/OpenIMSDK/Open-IM-Server/pkg/discoveryregistry"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
pbConversation "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/conversation"
pbGroup "github.com/OpenIMSDK/Open-IM-Server/pkg/proto/group"
"github.com/OpenIMSDK/Open-IM-Server/pkg/proto/sdkws"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
@@ -926,21 +925,34 @@ func (s *groupServer) SetGroupInfo(ctx context.Context, req *pbGroup.SetGroupInf
} else {
tips.OpUser = s.groupMemberDB2PB(opMember, 0)
}
s.Notification.GroupInfoSetNotification(ctx, tips)
var num int
if req.GroupInfoForSet.Notification != "" {
args := &pbConversation.ModifyConversationFieldReq{
Conversation: &pbConversation.Conversation{
OwnerUserID: mcontext.GetOpUserID(ctx),
ConversationID: utils.GetConversationIDBySessionType(constant.GroupChatType, group.GroupID),
ConversationType: constant.SuperGroupChatType,
GroupID: group.GroupID,
},
FieldType: constant.FieldGroupAtType,
UserIDList: userIDs,
}
if err := s.conversationRpcClient.ModifyConversationField(ctx, args); err != nil {
log.ZWarn(ctx, "modifyConversationField failed", err, "args", args)
num++
s.Notification.GroupInfoSetAnnouncementNotification(ctx, &sdkws.GroupInfoSetAnnouncementTips{Group: tips.Group, OpUser: tips.OpUser})
//args := &pbConversation.ModifyConversationFieldReq{
// Conversation: &pbConversation.Conversation{
// OwnerUserID: mcontext.GetOpUserID(ctx),
// ConversationID: utils.GetConversationIDBySessionType(constant.GroupChatType, group.GroupID),
// ConversationType: constant.SuperGroupChatType,
// GroupID: group.GroupID,
// },
// FieldType: constant.FieldGroupAtType,
// UserIDList: userIDs,
//}
//if err := s.conversationRpcClient.ModifyConversationField(ctx, args); err != nil {
// log.ZWarn(ctx, "modifyConversationField failed", err, "args", args)
//}
}
switch len(data) - num {
case 0:
case 1:
if req.GroupInfoForSet.GroupName == "" {
s.Notification.GroupInfoSetNotification(ctx, tips)
} else {
s.Notification.GroupInfoSetNameNotification(ctx, &sdkws.GroupInfoSetNameTips{Group: tips.Group, OpUser: tips.OpUser})
}
default:
s.Notification.GroupInfoSetNotification(ctx, tips)
}
return resp, nil
}