From 375f63a447dcbbdafbeade31b5a53b4b780fd24d Mon Sep 17 00:00:00 2001 From: chao <48119764+withchao@users.noreply.github.com> Date: Thu, 1 Aug 2024 15:11:17 +0800 Subject: [PATCH] fix: online status renewal (#2468) * fix: GroupApplicationAcceptedNotification * fix: GroupApplicationAcceptedNotification * fix: NotificationUserInfoUpdate * cicd: robot automated Change * fix: component * fix: getConversationInfo * feat: cron task * feat: cron task * feat: cron task * feat: cron task * feat: cron task * fix: minio config url recognition error * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * new mongo * friend incr sync * friend incr sync * friend incr sync * friend incr sync * friend incr sync * mage * optimization version log * optimization version log * sync * sync * sync * group sync * sync option * sync option * refactor: replace `friend` package with `realtion`. * refactor: update lastest commit to relation. * sync option * sync option * sync option * sync * sync * go.mod * seq * update: go mod * refactor: change incremental to full * feat: get full friend user ids * feat: api and config * seq * group version * merge * seq * seq * seq * fix: sort by id avoid unstable sort friends. * group * group * group * fix: sort by id avoid unstable sort friends. * fix: sort by id avoid unstable sort friends. * fix: sort by id avoid unstable sort friends. * user version * seq * seq * seq user * user online * implement minio expire delete. * user online * config * fix * fix * implement minio expire delete logic. * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * online cache * feat: implement scheduled delete outdated object in minio. * update gomake version * update gomake version * implement FindExpires pagination. * remove unnesseary incr. * fix uncorrect args call. * online push * online push * online push * resolving conflicts * resolving conflicts * test * api prommetrics * api prommetrics * api prommetrics * api prommetrics * api prommetrics * rpc prommetrics * rpc prommetrics * online status * online status * online status * online status * sub * conversation version incremental * merge seq * merge online * merge online * merge online * merge seq * GetOwnerConversation * fix: change incremental syncer router name. * rockscache batch get * rockscache seq batch get * fix: GetMsgDocModelByIndex bug * update go.mod * update go.mod * merge * feat: prometheus * feat: prometheus * group member sort * sub * sub * fix: seq conversion bug * fix: redis pipe exec * sort version * sort version * sort version * remove old version online subscription * remove old version online subscription * version log index * version log index * batch push * batch push * seq void filling * fix: batchGetMaxSeq * fix: batchGetMaxSeq * cache db error log * 111 * fix bug * fix: ImportFriends * add online cache * add some logs * add some logs * fix: onlineUserIDs * add logs * test * test * test * test --------- Co-authored-by: withchao Co-authored-by: Monet Lee Co-authored-by: OpenIM-Gordon <46924906+FGadvancer@users.noreply.github.com> Co-authored-by: icey-yu <1186114839@qq.com> --- internal/msggateway/online.go | 9 +++++++-- internal/msggateway/user_map.go | 11 ++++++++--- internal/push/push_handler.go | 2 +- 3 files changed, 16 insertions(+), 6 deletions(-) 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 }