IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-13 05:25:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: in standalone mode, the user online status is wrong

Browse Source
This commit is contained in:
withchao
2025-05-16 17:26:52 +08:00
parent 1144076168
commit cd2573534d
10 changed files with 127 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rpccache/online.go
+74 -18
View File
@@ -9,6 +9,7 @@ import (
"sync/atomic"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
"github.com/openimsdk/protocol/constant"
"github.com/openimsdk/protocol/user"
@@ -23,9 +24,25 @@ import (
"github.com/redis/go-redis/v9"
)
func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (*OnlineCache, error) {
const (
Begin uint32 = iota
DoOnlineStatusOver
DoSubscribeOver
)
type OnlineCache interface {
GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error)
GetUserOnline(ctx context.Context, userID string) (bool, error)
GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error)
WaitCache()
}
func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.UniversalClient, fullUserCache bool, fn func(ctx context.Context, userID string, platformIDs []int32)) (OnlineCache, error) {
if config.Standalone() {
return disableOnlineCache{client: client}, nil
}
l := &sync.Mutex{}
x := &OnlineCache{
x := &defaultOnlineCache{
client: client,
group: group,
fullUserCache: fullUserCache,
@@ -60,13 +77,7 @@ func NewOnlineCache(client *rpcli.UserClient, group *GroupLocalCache, rdb redis.
return x, nil
}
const (
Begin uint32 = iota
DoOnlineStatusOver
DoSubscribeOver
)
type OnlineCache struct {
type defaultOnlineCache struct {
client *rpcli.UserClient
group *GroupLocalCache
@@ -82,7 +93,7 @@ type OnlineCache struct {
CurrentPhase atomic.Uint32
}
func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
func (o *defaultOnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
log.ZDebug(ctx, "init users online status begin")
var (
@@ -135,7 +146,7 @@ func (o *OnlineCache) initUsersOnlineStatus(ctx context.Context) (err error) {
return nil
}
func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
func (o *defaultOnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient, fn func(ctx context.Context, userID string, platformIDs []int32)) {
o.Lock.Lock()
ch := rdb.Subscribe(ctx, cachekey.OnlineChannel).Channel()
for o.CurrentPhase.Load() < DoOnlineStatusOver {
@@ -186,7 +197,7 @@ func (o *OnlineCache) doSubscribe(ctx context.Context, rdb redis.UniversalClient
}
}
func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
func (o *defaultOnlineCache) getUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
platformIDs, err := o.lruCache.Get(userID, func() ([]int32, error) {
return o.client.GetUserOnlinePlatform(ctx, userID)
})
@@ -198,7 +209,7 @@ func (o *OnlineCache) getUserOnlinePlatform(ctx context.Context, userID string)
return platformIDs, nil
}
func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
func (o *defaultOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
if err != nil {
return nil, err
@@ -208,7 +219,7 @@ func (o *OnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string)
return platformIDs, nil
}
func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
func (o *defaultOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
platformIDs, err := o.getUserOnlinePlatform(ctx, userID)
if err != nil {
return false, err
@@ -216,7 +227,7 @@ func (o *OnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, e
return len(platformIDs) > 0, nil
}
func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
func (o *defaultOnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []string) (map[string][]int32, error) {
if len(userIDs) == 0 {
return nil, nil
}
@@ -240,7 +251,7 @@ func (o *OnlineCache) getUserOnlinePlatformBatch(ctx context.Context, userIDs []
return platformIDsMap, nil
}
func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
func (o *defaultOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
t := time.Now()
var (
@@ -276,7 +287,7 @@ func (o *OnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]s
return onlineUserIDs, offlineUserIDs, nil
}
func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
func (o *defaultOnlineCache) setUserOnline(userID string, platformIDs []int32) {
switch o.fullUserCache {
case true:
o.mapCache.Store(userID, platformIDs)
@@ -285,6 +296,51 @@ func (o *OnlineCache) setUserOnline(userID string, platformIDs []int32) {
}
}
func (o *OnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
func (o *defaultOnlineCache) setHasUserOnline(userID string, platformIDs []int32) bool {
return o.lruCache.SetHas(userID, platformIDs)
}
func (o *defaultOnlineCache) WaitCache() {
o.Lock.Lock()
for o.CurrentPhase.Load() < DoSubscribeOver {
o.Cond.Wait()
}
o.Lock.Unlock()
}
type disableOnlineCache struct {
client *rpcli.UserClient
}
func (o disableOnlineCache) GetUserOnlinePlatform(ctx context.Context, userID string) ([]int32, error) {
return o.client.GetUserOnlinePlatform(ctx, userID)
}
func (o disableOnlineCache) GetUserOnline(ctx context.Context, userID string) (bool, error) {
onlinePlatform, err := o.client.GetUserOnlinePlatform(ctx, userID)
if err != nil {
return false, err
}
return len(onlinePlatform) > 0, err
}
func (o disableOnlineCache) GetUsersOnline(ctx context.Context, userIDs []string) ([]string, []string, error) {
var (
onlineUserIDs = make([]string, 0, len(userIDs))
offlineUserIDs = make([]string, 0, len(userIDs))
)
for _, userID := range userIDs {
online, err := o.GetUserOnline(ctx, userID)
if err != nil {
return nil, nil, err
}
if online {
onlineUserIDs = append(onlineUserIDs, userID)
} else {
offlineUserIDs = append(offlineUserIDs, userID)
}
}
return onlineUserIDs, offlineUserIDs, nil
}
func (o disableOnlineCache) WaitCache() {}