diff --git a/internal/api/route.go b/internal/api/route.go index 0f46f0f0c..1926b55f1 100644 --- a/internal/api/route.go +++ b/internal/api/route.go @@ -81,6 +81,7 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive userRouterGroup.POST("/get_users_online_token_detail", ParseToken, u.GetUsersOnlineTokenDetail) userRouterGroup.POST("/subscribe_users_status", ParseToken, u.UnSubscriberStatus) userRouterGroup.POST("/unsubscribe_users_status", ParseToken, u.UnSubscriberStatus) + userRouterGroup.POST("/get_users_status", ParseToken, u.GetUserStatus) } // friend routing group diff --git a/internal/api/user.go b/internal/api/user.go index 108ccac69..3eb136b3d 100644 --- a/internal/api/user.go +++ b/internal/api/user.go @@ -194,3 +194,7 @@ func (u *UserApi) SubscriberStatus(c *gin.Context) { func (u *UserApi) UnSubscriberStatus(c *gin.Context) { a2r.Call(user.UserClient.SubscribeOrCancelUsersStatus, u.Client, c) } + +func (u *UserApi) GetUserStatus(c *gin.Context) { + a2r.Call(user.UserClient.GetUserStatus, u.Client, c) +} diff --git a/internal/rpc/user/user.go b/internal/rpc/user/user.go index 576492566..53598db0b 100644 --- a/internal/rpc/user/user.go +++ b/internal/rpc/user/user.go @@ -261,11 +261,18 @@ func (s *userServer) SubscribeOrCancelUsersStatus(ctx context.Context, req *pbus } func (s *userServer) GetUserStatus(ctx context.Context, req *pbuser.GetUserStatusReq) (resp *pbuser.GetUserStatusResp, err error) { - //TODO implement me - panic("implement me") + //TODO 是否加一个参数校验-判断req.userID的数量，每一个获取加一个限制，一次请求限制500？ + onlineStatusList, err := s.UserDatabase.GetUserStatus(ctx, req.UserIDs) + if err != nil { + return nil, err + } + return &pbuser.GetUserStatusResp{StatusList: onlineStatusList}, nil } func (s *userServer) SetUserStatus(ctx context.Context, req *pbuser.SetUserStatusReq) (resp *pbuser.SetUserStatusResp, err error) { - //TODO implement me - panic("implement me") + err = s.UserDatabase.SetUserStatus(ctx, req.StatusList) + if err != nil { + return nil, err + } + return &pbuser.SetUserStatusResp{}, nil } diff --git a/pkg/common/db/cache/user.go b/pkg/common/db/cache/user.go index 455bc9ebe..5c76af22f 100644 --- a/pkg/common/db/cache/user.go +++ b/pkg/common/db/cache/user.go @@ -16,6 +16,9 @@ package cache import ( "context" + "encoding/json" + "github.com/OpenIMSDK/protocol/user" + "strconv" "time" "github.com/dtm-labs/rockscache" @@ -25,9 +28,12 @@ import ( ) const ( - userExpireTime = time.Second * 60 * 60 * 12 - userInfoKey = "USER_INFO:" - userGlobalRecvMsgOptKey = "USER_GLOBAL_RECV_MSG_OPT_KEY:" + userExpireTime = time.Second * 60 * 60 * 12 + userInfoKey = "USER_INFO:" + userGlobalRecvMsgOptKey = "USER_GLOBAL_RECV_MSG_OPT_KEY:" + olineStatusKey = "ONLINE_STATUS:" + userOlineStatusExpireTime = time.Second * 60 * 60 * 24 + statusMod = 500 ) type UserCache interface { @@ -38,10 +44,13 @@ type UserCache interface { DelUsersInfo(userIDs ...string) UserCache GetUserGlobalRecvMsgOpt(ctx context.Context, userID string) (opt int, err error) DelUsersGlobalRecvMsgOpt(userIDs ...string) UserCache + GetUserStatus(ctx context.Context, userIDs []string) ([]*user.OnlineStatus, error) + SetUserStatus(ctx context.Context, list []*user.OnlineStatus) error } type UserCacheRedis struct { metaCache + rdb redis.UniversalClient userDB relationTb.UserModelInterface expireTime time.Duration rcClient *rockscache.Client @@ -54,6 +63,7 @@ func NewUserCacheRedis( ) UserCache { rcClient := rockscache.NewClient(rdb, options) return &UserCacheRedis{ + rdb: rdb, metaCache: NewMetaCacheRedis(rcClient), userDB: userDB, expireTime: userExpireTime, @@ -63,6 +73,7 @@ func NewUserCacheRedis( func (u *UserCacheRedis) NewCache() UserCache { return &UserCacheRedis{ + rdb: u.rdb, metaCache: NewMetaCacheRedis(u.rcClient, u.metaCache.GetPreDelKeys()...), userDB: u.userDB, expireTime: u.expireTime, @@ -145,3 +156,71 @@ func (u *UserCacheRedis) DelUsersGlobalRecvMsgOpt(userIDs ...string) UserCache { cache.AddKeys(keys...) return cache } + +func (u *UserCacheRedis) getOnlineStatusKey(userID string) string { + return olineStatusKey + userID +} + +// GetUserStatus get user status +func (u *UserCacheRedis) GetUserStatus(ctx context.Context, userIDs []string) ([]*user.OnlineStatus, error) { + var res []*user.OnlineStatus + for _, userID := range userIDs { + UserIDNum, err := strconv.Atoi(userID) + if err != nil { + return nil, err + } + var modKey = strconv.Itoa(UserIDNum % statusMod) + var onlineStatus user.OnlineStatus + key := olineStatusKey + modKey + result, err := u.rdb.HGet(ctx, key, userID).Result() + if err != nil { + if err == redis.Nil { + // key or field does not exist + res = append(res, &user.OnlineStatus{ + UserID: userID, + Status: 0, + PlatformID: -1, + }) + continue + } else { + return nil, err + } + } + err = json.Unmarshal([]byte(result), &onlineStatus) + if err != nil { + return nil, err + } + onlineStatus.UserID = userID + res = append(res, &onlineStatus) + } + return res, nil +} + +// SetUserStatus Set the user status and save it in redis +func (u *UserCacheRedis) SetUserStatus(ctx context.Context, list []*user.OnlineStatus) error { + for _, status := range list { + var isNewKey int64 + UserIDNum, err := strconv.Atoi(status.UserID) + if err != nil { + return err + } + var modKey = strconv.Itoa(UserIDNum % statusMod) + key := olineStatusKey + modKey + jsonData, err := json.Marshal(status) + if err != nil { + return err + } + isNewKey, err = u.rdb.Exists(ctx, key).Result() + if err != nil { + return err + } + _, err = u.rdb.HSet(ctx, key, status.UserID, string(jsonData)).Result() + if err != nil { + return err + } + if isNewKey > 0 { + u.rdb.Expire(ctx, key, userOlineStatusExpireTime) + } + } + return nil +} diff --git a/pkg/common/db/controller/user.go b/pkg/common/db/controller/user.go index 83942ce22..5b303ebd7 100644 --- a/pkg/common/db/controller/user.go +++ b/pkg/common/db/controller/user.go @@ -18,6 +18,7 @@ import ( "context" unRelationTb "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/table/unrelation" "github.com/OpenIMSDK/protocol/constant" + "github.com/OpenIMSDK/protocol/user" "time" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/cache" @@ -56,6 +57,10 @@ type UserDatabase interface { GetAllSubscribeList(ctx context.Context, userID string) ([]string, error) // GetSubscribedList Get all subscribed lists GetSubscribedList(ctx context.Context, userID string) ([]string, error) + // GetUserStatus Get the online status of the user + GetUserStatus(ctx context.Context, userIDs []string) ([]*user.OnlineStatus, error) + // SetUserStatus Set the user status and store the user status in redis + SetUserStatus(ctx context.Context, list []*user.OnlineStatus) error } type userDatabase struct { @@ -195,3 +200,14 @@ func (u *userDatabase) GetSubscribedList(ctx context.Context, userID string) ([] //TODO 获取所有被订阅 return nil, nil } + +// GetUserStatus get user status +func (u *userDatabase) GetUserStatus(ctx context.Context, userIDs []string) ([]*user.OnlineStatus, error) { + onlineStatusList, err := u.cache.GetUserStatus(ctx, userIDs) + return onlineStatusList, err +} + +// SetUserStatus Set the user status and save it in redis +func (u *userDatabase) SetUserStatus(ctx context.Context, list []*user.OnlineStatus) error { + return u.cache.SetUserStatus(ctx, list) +}