fix: get userid function call rpc and message gateway add status change callback (#699)

* fix: to start im or chat, ZooKeeper must be started first.

* fix: msg gateway start output err info

Signed-off-by: Gordon <1432970085@qq.com>

* fix: msg gateway start output err info

Signed-off-by: Gordon <1432970085@qq.com>

* chore: package path changes

Signed-off-by: withchao <993506633@qq.com>

* fix: go mod update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* chore: package path changes

Signed-off-by: withchao <993506633@qq.com>

* chore: package path changes

Signed-off-by: withchao <993506633@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: token update

Signed-off-by: Gordon <1432970085@qq.com>

* fix: get all userID

Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com>

* fix: msggateway add online status call

Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com>

---------

Signed-off-by: Gordon <1432970085@qq.com>
Signed-off-by: withchao <993506633@qq.com>
Signed-off-by: Gordon <46924906+FGadvancer@users.noreply.github.com>
Co-authored-by: withchao <993506633@qq.com>

go.mod

@@ -37,7 +37,7 @@ require (
 require github.com/google/uuid v1.3.0
 
 require (
-	github.com/OpenIMSDK/protocol v0.0.2
+	github.com/OpenIMSDK/protocol v0.0.3
 	github.com/OpenIMSDK/tools v0.0.5
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.7+incompatible
 	github.com/go-redis/redis v6.15.9+incompatible

go.sum

@@ -16,8 +16,8 @@ cloud.google.com/go/storage v1.28.1/go.mod h1:Qnisd4CqDdo6BGs2AD5LLnEsmSQ80wQ5og
 firebase.google.com/go v3.13.0+incompatible h1:3TdYC3DDi6aHn20qoRkxwGqNgdjtblwVAyRLQwGn/+4=
 firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIwjt8toICdV5Wh9ptHs=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/OpenIMSDK/protocol v0.0.2 h1:O53/WiqLCHF9aWPLI32GPF82hn7suM8PkhrtL89Klrw=
+github.com/OpenIMSDK/protocol v0.0.3 h1:CFQtmnyW+1dYKVFaVaHcJ6oYuMiMdNfU2gC1xz3K/9I=
-github.com/OpenIMSDK/protocol v0.0.2/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
+github.com/OpenIMSDK/protocol v0.0.3/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
 github.com/OpenIMSDK/tools v0.0.5 h1:yBVHJ3EpIDcp8VFKPjuGr6MQvFa3t4JByZ+vmeC06/Q=
 github.com/OpenIMSDK/tools v0.0.5/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
 github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=

internal/api/user.go

@@ -51,7 +51,7 @@ func (u *UserApi) GetUsersPublicInfo(c *gin.Context) {
 }
 
 func (u *UserApi) GetAllUsersID(c *gin.Context) {
-	a2r.Call(user.UserClient.GetDesignateUsers, u.Client, c)
+	a2r.Call(user.UserClient.GetAllUserID, u.Client, c)
 }
 
 func (u *UserApi) AccountCheck(c *gin.Context) {

internal/msggateway/n_ws_server.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"github.com/OpenIMSDK/Open-IM-Server/pkg/authverify"
+	"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
 	"net/http"
 	"strconv"
 	"sync"
@@ -74,6 +75,7 @@ type WsServer struct {
 	hubServer         *Server
 	validate          *validator.Validate
 	cache             cache.MsgModel
+	userClient        *rpcclient.UserRpcClient
 	Compressor
 	Encoder
 	MessageHandler
@@ -86,6 +88,28 @@ type kickHandler struct {
 
 func (ws *WsServer) SetDiscoveryRegistry(client discoveryregistry.SvcDiscoveryRegistry) {
 	ws.MessageHandler = NewGrpcHandler(ws.validate, client)
+	u := rpcclient.NewUserRpcClient(client)
+	ws.userClient = &u
+}
+func (ws *WsServer) SetUserOnlineStatus(ctx context.Context, client *Client, status int32) {
+	err := ws.userClient.SetUserStatus(ctx, client.UserID, status, client.PlatformID)
+	if err != nil {
+		log.ZWarn(ctx, "SetUserStatus err", err)
+	}
+	switch status {
+	case constant.Online:
+		err := CallbackUserOnline(ctx, client.UserID, client.PlatformID, client.IsBackground, client.ctx.GetConnID())
+		if err != nil {
+			log.ZWarn(ctx, "CallbackUserOnline err", err)
+		}
+	case constant.Offline:
+		err := CallbackUserOffline(ctx, client.UserID, client.PlatformID, client.ctx.GetConnID())
+		if err != nil {
+			log.ZWarn(ctx, "CallbackUserOffline err", err)
+		}
+
+	}
+
 }
 
 func (ws *WsServer) SetCacheHandler(cache cache.MsgModel) {
@@ -186,6 +210,7 @@ func (ws *WsServer) registerClient(client *Client) {
 			atomic.AddInt64(&ws.onlineUserConnNum, 1)
 		}
 	}
+	ws.SetUserOnlineStatus(client.ctx, client, constant.Online)
 	log.ZInfo(
 		client.ctx,
 		"user online",
@@ -292,14 +317,8 @@ func (ws *WsServer) unregisterClient(client *Client) {
 		atomic.AddInt64(&ws.onlineUserNum, -1)
 	}
 	atomic.AddInt64(&ws.onlineUserConnNum, -1)
-	log.ZInfo(
+	ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
-		client.ctx,
+	log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num", ws.onlineUserNum, "online user conn Num",
-		"user offline",
-		"close reason",
-		client.closedErr,
-		"online user Num",
-		ws.onlineUserNum,
-		"online user conn Num",
 		ws.onlineUserConnNum,
 	)
 }

pkg/rpcclient/user.go

@@ -154,3 +154,7 @@ func (u *UserRpcClient) GetAllUserIDs(ctx context.Context, pageNumber, showNumbe
 	}
 	return resp.UserIDs, nil
 }
+func (u *UserRpcClient) SetUserStatus(ctx context.Context, userID string, status int32, platformID int) error {
+	_, err := u.Client.SetUserStatus(ctx, &user.SetUserStatusReq{StatusList: []*user.OnlineStatus{{UserID: userID, Status: status, PlatformID: int32(platformID)}}})
+	return err
+}