Merge remote-tracking branch 'upstream/main'

# Conflicts:
#	internal/tools/msg.go

cmd/openim-api/main.go

@@ -26,6 +26,7 @@ import (
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"github.com/OpenIMSDK/tools/log"
+
 	"github.com/openimsdk/open-im-server/v3/internal/api"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"

go.mod

@@ -38,7 +38,7 @@ require github.com/google/uuid v1.3.1
 require (
 	github.com/IBM/sarama v1.41.3
 	github.com/OpenIMSDK/protocol v0.0.30
-	github.com/OpenIMSDK/tools v0.0.15
+	github.com/OpenIMSDK/tools v0.0.16
 	github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
 	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/go-sql-driver/mysql v1.7.1

go.sum

@@ -20,8 +20,8 @@ github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c=
 github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ=
 github.com/OpenIMSDK/protocol v0.0.30 h1:MiHO6PyQMR9ojBHNnSFxCHLmsoE2xZqaiYj975JiZnM=
 github.com/OpenIMSDK/protocol v0.0.30/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
-github.com/OpenIMSDK/tools v0.0.15 h1:FF3m0TQUG56pJC15a11jmBG6Y1EjXarEW4JV3CBF/Jc=
+github.com/OpenIMSDK/tools v0.0.16 h1:te/GIq2imCMsrRPgU9OObYKbzZ3rT08Lih/o+3QFIz0=
-github.com/OpenIMSDK/tools v0.0.15/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
+github.com/OpenIMSDK/tools v0.0.16/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
 github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
 github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 h1:uSoVVbwJiQipAclBbw+8quDsfcvFjOpI5iCf4p/cqCs=
 github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7/go.mod h1:6zEj6s6u/ghQa61ZWa/C2Aw3RkjiTBOix7dkqa1VLIs=

internal/msggateway/n_ws_server.go

@@ -74,8 +74,8 @@ type WsServer struct {
 	kickHandlerChan   chan *kickHandler
 	clients           *UserMap
 	clientPool        sync.Pool
-	onlineUserNum     int64
+	onlineUserNum     atomic.Int64
-	onlineUserConnNum int64
+	onlineUserConnNum atomic.Int64
 	handshakeTimeout  time.Duration
 	hubServer         *Server
 	validate          *validator.Validate
@@ -220,8 +220,8 @@ func (ws *WsServer) registerClient(client *Client) {
 	if !userOK {
 		ws.clients.Set(client.UserID, client)
 		log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID)
-		atomic.AddInt64(&ws.onlineUserNum, 1)
+		ws.onlineUserNum.Add(1)
-		atomic.AddInt64(&ws.onlineUserConnNum, 1)
+		ws.onlineUserConnNum.Add(1)
 	} else {
 		i := &kickHandler{
 			clientOK:   clientOK,
@@ -234,22 +234,35 @@ func (ws *WsServer) registerClient(client *Client) {
 			ws.clients.Set(client.UserID, client)
 			// 已经有同平台的连接存在
 			log.ZInfo(client.ctx, "repeat login", "userID", client.UserID, "platformID", client.PlatformID, "old remote addr", getRemoteAdders(oldClients))
-			atomic.AddInt64(&ws.onlineUserConnNum, 1)
+			ws.onlineUserConnNum.Add(1)
 		} else {
 			ws.clients.Set(client.UserID, client)
+			ws.onlineUserConnNum.Add(1)
+		}
+	}
 
-			atomic.AddInt64(&ws.onlineUserConnNum, 1)
+	wg := sync.WaitGroup{}
-		}
+	wg.Add(1)
-	}
+	go func() {
-	ws.sendUserOnlineInfoToOtherNode(client.ctx, client)
+		defer wg.Done()
+		_ = ws.sendUserOnlineInfoToOtherNode(client.ctx, client)
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
 		ws.SetUserOnlineStatus(client.ctx, client, constant.Online)
+	}()
+
+	wg.Wait()
+
 	log.ZInfo(
 		client.ctx,
 		"user online",
 		"online user Num",
-		ws.onlineUserNum,
+		ws.onlineUserNum.Load(),
 		"online user conn Num",
-		ws.onlineUserConnNum,
+		ws.onlineUserConnNum.Load(),
 	)
 }
 
@@ -282,7 +295,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		if clientOK {
 			isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients)
 			if isDeleteUser {
-				atomic.AddInt64(&ws.onlineUserNum, -1)
+				ws.onlineUserNum.Add(-1)
 			}
 			for _, c := range oldClients {
 				err := c.KickOnlineMessage()
@@ -350,18 +363,18 @@ func (ws *WsServer) unregisterClient(client *Client) {
 	defer ws.clientPool.Put(client)
 	isDeleteUser := ws.clients.delete(client.UserID, client.ctx.GetRemoteAddr())
 	if isDeleteUser {
-		atomic.AddInt64(&ws.onlineUserNum, -1)
+		ws.onlineUserNum.Add(-1)
 	}
-	atomic.AddInt64(&ws.onlineUserConnNum, -1)
+	ws.onlineUserConnNum.Add(-1)
 	ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
 	log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num", ws.onlineUserNum, "online user conn Num",
-		ws.onlineUserConnNum,
+		ws.onlineUserConnNum.Load(),
 	)
 }
 
 func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 	connContext := newContext(w, r)
-	if ws.onlineUserConnNum >= ws.wsMaxConnNum {
+	if ws.onlineUserConnNum.Load() >= ws.wsMaxConnNum {
 		httpError(connContext, errs.ErrConnOverMaxNumLimit)
 		return
 	}

internal/msgtransfer/init.go

@@ -16,10 +16,12 @@ package msgtransfer
 
 import (
 	"fmt"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
+	"sync"
+
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	"sync"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
 
 	"github.com/OpenIMSDK/tools/mw"
 

internal/tools/msg.go

@@ -17,6 +17,12 @@ package tools
 import (
 	"context"
 	"fmt"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
+	"github.com/redis/go-redis/v9"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"math"
+
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/mcontext"

pkg/common/cmd/api.go

@@ -16,9 +16,11 @@ package cmd
 
 import (
 	"fmt"
+
 	"github.com/OpenIMSDK/protocol/constant"
-	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/spf13/cobra"
+
+	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
 type ApiCmd struct {

pkg/common/cmd/rpc.go

@@ -16,11 +16,13 @@ package cmd
 
 import (
 	"errors"
+
 	"github.com/OpenIMSDK/protocol/constant"
-	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/spf13/cobra"
 	"google.golang.org/grpc"
 
+	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"

pkg/common/config/parse.go

@@ -21,8 +21,9 @@ import (
 	"path/filepath"
 
 	"github.com/OpenIMSDK/protocol/constant"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"gopkg.in/yaml.v3"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 )
 
 //go:embed version

pkg/common/db/relation/mysql_init.go

@@ -22,10 +22,11 @@ import (
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/mw/specialerror"
 	mysqldriver "github.com/go-sql-driver/mysql"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"gorm.io/driver/mysql"
 	"gorm.io/gorm"
 	"gorm.io/gorm/logger"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
 const (

pkg/common/discovery_register/k8s_discovery_register.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
+
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	openkeeper "github.com/OpenIMSDK/tools/discoveryregistry/zookeeper"
 	"github.com/OpenIMSDK/tools/log"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"google.golang.org/grpc"
-	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
 func NewDiscoveryRegister(envType string) (discoveryregistry.SvcDiscoveryRegistry, error) {