diff --git a/internal/msggateway/ws_server.go b/internal/msggateway/ws_server.go index 304cb3b5c..c59f9fe11 100644 --- a/internal/msggateway/ws_server.go +++ b/internal/msggateway/ws_server.go @@ -49,7 +49,7 @@ type WsServer struct { unregisterChan chan *Client kickHandlerChan chan *kickHandler clients UserMap - online *rpccache.OnlineCache + online rpccache.OnlineCache subscription *Subscription clientPool sync.Pool onlineUserNum atomic.Int64 diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index ae14d02a1..a895bb9c4 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -48,15 +48,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(ctx context.Cont log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String()) err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq) if err != nil { - log.ZError( - ctx, - "single data insert to mongo err", - err, - "msg", - msgFromMQ.MsgData, - "conversationID", - msgFromMQ.ConversationID, - ) + log.ZError(ctx, "single data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) prommetrics.MsgInsertMongoFailedCounter.Inc() } else { prommetrics.MsgInsertMongoSuccessCounter.Inc() diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index 418c4c7f2..7b1efe3bf 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -33,7 +33,7 @@ type ConsumerHandler struct { offlinePusher offlinepush.OfflinePusher onlinePusher OnlinePusher pushDatabase controller.PushDatabase - onlineCache *rpccache.OnlineCache + onlineCache rpccache.OnlineCache groupLocalCache *rpccache.GroupLocalCache conversationLocalCache *rpccache.ConversationLocalCache webhookClient *webhook.Client @@ -120,11 +120,7 @@ func (c *ConsumerHandler) HandleMs2PsChat(ctx context.Context, msg []byte) { } func (c *ConsumerHandler) WaitCache() { - c.onlineCache.Lock.Lock() - for c.onlineCache.CurrentPhase.Load() < rpccache.DoSubscribeOver { - c.onlineCache.Cond.Wait() - } - c.onlineCache.Lock.Unlock() + c.onlineCache.WaitCache() } // Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType. diff --git a/pkg/rpccache/online.go b/pkg/rpccache/online.go index 959b054a2..9b8c03101 100644 --- a/pkg/rpccache/online.go +++ b/pkg/rpccache/online.go @@ -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() {}