msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/errcode' into errcode

Browse Source
test-errcode
withchao 1 year ago
parent 95cbf28b99 1e075a2468
commit e103e9ad3b
5 changed files with 35 additions and 30 deletions
Show Stats Download Patch File Download Diff File

5
cmd/api/main.go
Unescape View File

@ -38,6 +38,7 @@ func run(port int) error {
if err != nil {
return err
}
fmt.Println("api start init discov client")
var client discoveryregistry.SvcDiscoveryRegistry
client, err = openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
openKeeper.WithFreq(time.Hour), openKeeper.WithUserNameAndPassword(config.Config.Zookeeper.UserName,
@ -45,14 +46,18 @@ func run(port int) error {
if err != nil {
return err
}
fmt.Println("api init discov client success")
buf := bytes.NewBuffer(nil)
if err := yaml.NewEncoder(buf).Encode(config.Config); err != nil {
return err
}
fmt.Println("api register public config to discov")
if err := client.RegisterConf2Registry(constant.OpenIMCommonConfigKey, buf.Bytes()); err != nil {
return err
}
fmt.Println("api register public config to discov success")
router := api.NewGinRouter(client, rdb)
fmt.Println("api init router success")
var address string
if config.Config.Api.ListenIP != "" {
address = net.JoinHostPort(config.Config.Api.ListenIP, strconv.Itoa(port))

32
internal/msggateway/client.go
Unescape 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 {

10
internal/msggateway/hub_server.go
Unescape 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)

16
internal/msggateway/n_ws_server.go
Unescape 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)
}

2
internal/msggateway/user_map.go
Unescape 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)
}
}

Write Preview
Loading…
Cancel
Save