add logs

internal/msggateway/online.go

@@ -4,13 +4,16 @@ import (
 	"context"
 	"crypto/md5"
 	"encoding/binary"
+	"fmt"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
 	pbuser "github.com/openimsdk/protocol/user"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/utils/datautil"
 	"math/rand"
+	"os"
 	"strconv"
+	"sync/atomic"
 	"time"
 )
 
@@ -78,8 +81,10 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
 		}
 	}
 
-	opIdCtx := mcontext.SetOperationID(context.Background(), "r"+strconv.FormatUint(rNum, 10))
+	var count atomic.Int64
+	operationIDPrefix := fmt.Sprintf("p_%d_", os.Getpid())
 	doRequest := func(req *pbuser.SetUserOnlineStatusReq) {
+		opIdCtx := mcontext.SetOperationID(context.Background(), operationIDPrefix+strconv.FormatInt(count.Add(1), 10))
 		ctx, cancel := context.WithTimeout(opIdCtx, time.Second*5)
 		defer cancel()
 		if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil {
@@ -102,7 +107,7 @@ func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
 		case now := <-renewalTicker.C:
 			deadline := now.Add(-cachekey.OnlineExpire / 3)
 			users := ws.clients.GetAllUserStatus(deadline, now)
-			log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "num", len(users))
+			log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "users", users)
 			pushUserState(users...)
 		case state := <-ws.clients.UserState():
 			log.ZDebug(context.Background(), "OnlineCache user online change", "userID", state.UserID, "online", state.Online, "offline", state.Offline)

internal/msggateway/user_map.go

@@ -1,7 +1,13 @@
 package msggateway
 
 import (
+	"bytes"
+	"context"
+	"fmt"
+	"github.com/openimsdk/protocol/constant"
+	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/utils/datautil"
+	"strconv"
 	"sync"
 	"time"
 )
@@ -27,6 +33,23 @@ type UserPlatform struct {
 	Clients []*Client
 }
 
+func (u *UserPlatform) String() string {
+	buf := bytes.NewBuffer(nil)
+	buf.WriteString("UserPlatform{Time: ")
+	buf.WriteString(u.Time.Format(time.DateTime))
+	buf.WriteString(", Clients<")
+	buf.WriteString(strconv.Itoa(len(u.Clients)))
+	buf.WriteString(">: [")
+	for i, client := range u.Clients {
+		if i > 0 {
+			buf.WriteString(", ")
+		}
+		buf.WriteString(fmt.Sprintf("%p %d %s", u.Clients[i], client.PlatformID, constant.PlatformIDToName(client.PlatformID)))
+	}
+	buf.WriteString("]}")
+	return buf.String()
+}
+
 func (u *UserPlatform) PlatformIDs() []int32 {
 	if len(u.Clients) == 0 {
 		return nil
@@ -117,6 +140,7 @@ func (u *userMap) Get(userID string, platformID int) ([]*Client, bool, bool) {
 }
 
 func (u *userMap) Set(userID string, client *Client) {
+	log.ZDebug(context.Background(), "userMap Set", "userID", userID, "platformID", client.PlatformID, "platform", constant.PlatformIDToName(client.PlatformID), "pointer", fmt.Sprintf("%p", client))
 	u.lock.Lock()
 	defer u.lock.Unlock()
 	result, ok := u.data[userID]
@@ -135,6 +159,9 @@ func (u *userMap) DeleteClients(userID string, clients []*Client) (isDeleteUser
 	if len(clients) == 0 {
 		return false
 	}
+	for i, client := range clients {
+		log.ZDebug(context.Background(), "userMap DeleteClients", "userID", userID, "platformID", client.PlatformID, "platform", constant.PlatformIDToName(client.PlatformID), "count", fmt.Sprintf("%p %d/%d", clients[i], i+1, len(clients)))
+	}
 	u.lock.Lock()
 	defer u.lock.Unlock()
 	result, ok := u.data[userID]
@@ -167,6 +194,7 @@ func (u *userMap) GetAllUserStatus(deadline time.Time, nowtime time.Time) []User
 	defer u.lock.RUnlock()
 	result := make([]UserState, 0, len(u.data))
 	for userID, userPlatform := range u.data {
+		log.ZDebug(context.Background(), "userMap GetAllUserStatus", "userID", userID, "platforms", userPlatform.String())
 		if userPlatform.Time.Before(deadline) {
 			continue
 		}