diff --git a/internal/msggateway/online.go b/internal/msggateway/online.go index b50608f93..27b4544aa 100644 --- a/internal/msggateway/online.go +++ b/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, "num", len(users), "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) diff --git a/internal/msggateway/user_map.go b/internal/msggateway/user_map.go index bd1f19728..36cab4ed7 100644 --- a/internal/msggateway/user_map.go +++ b/internal/msggateway/user_map.go @@ -1,6 +1,10 @@ package msggateway import ( + "context" + "fmt" + "github.com/openimsdk/protocol/constant" + "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/utils/datautil" "sync" "time" @@ -117,6 +121,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] @@ -162,12 +167,12 @@ func (u *userMap) DeleteClients(userID string, clients []*Client) (isDeleteUser return true } -func (u *userMap) GetAllUserStatus(deadline time.Time, nowtime time.Time) []UserState { +func (u *userMap) GetAllUserStatus(deadline time.Time, nowtime time.Time) (result []UserState) { u.lock.RLock() defer u.lock.RUnlock() - result := make([]UserState, 0, len(u.data)) + result = make([]UserState, 0, len(u.data)) for userID, userPlatform := range u.data { - if userPlatform.Time.Before(deadline) { + if deadline.Before(userPlatform.Time) { continue } userPlatform.Time = nowtime diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index c3b27372b..249622a59 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -202,7 +202,7 @@ func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws. var result []*msggateway.SingleMsgToUserResults if len(onlineUserIDs) > 0 { var err error - result, err = c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs) + result, err = c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, onlineUserIDs) if err != nil { return nil, err }