msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Improve user subscription and unsubscription

Browse Source
pull/757/head
pluto 2 years ago
parent d9dbd739ab
commit 86e09f0d0e
4 changed files with 103 additions and 44 deletions
Show Stats Download Patch File Download Diff File

20
internal/rpc/user/user.go
Unescape View File

@ -250,18 +250,28 @@ func (s *userServer) GetAllUserID(ctx context.Context, req *pbuser.GetAllUserIDR
// SubscribeOrCancelUsersStatus Subscribe online or cancel online users.
func (s *userServer) SubscribeOrCancelUsersStatus(ctx context.Context, req *pbuser.SubscribeOrCancelUsersStatusReq) (resp *pbuser.SubscribeOrCancelUsersStatusResp, err error) {
err = s.UserDatabase.SubscribeOrCancelUsersStatus(ctx, req.UserID, req.UserIDs, req.Genre)
if req.Genre == constant.SubscriberUser {
err = s.UserDatabase.SubscribeUsersStatus(ctx, req.UserID, req.UserIDs)
var status []*pbuser.OnlineStatus
status, err = s.UserDatabase.GetUserStatus(ctx, req.UserIDs)
if err != nil {
return nil, err
}
//var status map[string][]string
//TODO 获取用户在线列表，返回订阅的用户的在线列表
var statusMap map[string]int32
for _, value := range status {
statusMap[value.UserID] = value.Status
}
return &pbuser.SubscribeOrCancelUsersStatusResp{Status: statusMap}, nil
} else if req.Genre == constant.Unsubscribe {
err = s.UserDatabase.UnsubscribeUsersStatus(ctx, req.UserID, req.UserIDs)
if err != nil {
return nil, err
}
}
return &pbuser.SubscribeOrCancelUsersStatusResp{}, nil
}
func (s *userServer) GetUserStatus(ctx context.Context, req *pbuser.GetUserStatusReq) (resp *pbuser.GetUserStatusResp, err error) {
//TODO 是否加一个参数校验-判断req.userID的数量，每一个获取加一个限制，一次请求限制500？
onlineStatusList, err := s.UserDatabase.GetUserStatus(ctx, req.UserIDs)
if err != nil {
return nil, err

38
pkg/common/db/controller/user.go
Unescape View File

@ -17,7 +17,6 @@ package controller
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"
@ -51,8 +50,10 @@ type UserDatabase interface {
CountTotal(ctx context.Context, before *time.Time) (int64, error)
// CountRangeEverydayTotal Get the user increment in the range
CountRangeEverydayTotal(ctx context.Context, start time.Time, end time.Time) (map[string]int64, error)
//SubscribeOrCancelUsersStatus Subscribe or unsubscribe a user's presence status
SubscribeOrCancelUsersStatus(ctx context.Context, userID string, userIDs []string, genre int32) error
//SubscribeUsersStatus Subscribe a user's presence status
SubscribeUsersStatus(ctx context.Context, userID string, userIDs []string) error
// UnsubscribeUsersStatus unsubscribe a user's presence status
UnsubscribeUsersStatus(ctx context.Context, userID string, userIDs []string) error
// GetAllSubscribeList Get a list of all subscriptions
GetAllSubscribeList(ctx context.Context, userID string) ([]string, error)
// GetSubscribedList Get all subscribed lists
@ -176,29 +177,34 @@ func (u *userDatabase) CountRangeEverydayTotal(ctx context.Context, start time.T
return u.userDB.CountRangeEverydayTotal(ctx, start, end)
}
//SubscribeOrCancelUsersStatus Subscribe or unsubscribe a user's presence status
func (u *userDatabase) SubscribeOrCancelUsersStatus(ctx context.Context, userID string, userIDs []string, genre int32) error {
var err error
if genre == constant.SubscriberUser {
err = u.mongoDB.AddSubscriptionList(ctx, userID, userIDs)
} else if genre == constant.Unsubscribe {
err = u.mongoDB.UnsubscriptionList(ctx, userID, userIDs)
// SubscribeUsersStatus Subscribe or unsubscribe a user's presence status
func (u *userDatabase) SubscribeUsersStatus(ctx context.Context, userID string, userIDs []string) error {
err := u.mongoDB.AddSubscriptionList(ctx, userID, userIDs)
return err
}
// UnsubscribeUsersStatus unsubscribe a user's presence status
func (u *userDatabase) UnsubscribeUsersStatus(ctx context.Context, userID string, userIDs []string) error {
err := u.mongoDB.UnsubscriptionList(ctx, userID, userIDs)
return err
}
// GetAllSubscribeList Get a list of all subscriptions.
func (u *userDatabase) GetAllSubscribeList(ctx context.Context, userID string) ([]string, error) {
//TODO 获取所有订阅
return nil, nil
list, err := u.mongoDB.GetAllSubscribeList(ctx, userID)
if err != nil {
return nil, err
}
return list, nil
}
// GetSubscribedList Get all subscribed lists
func (u *userDatabase) GetSubscribedList(ctx context.Context, userID string) ([]string, error) {
//TODO 获取所有被订阅
return nil, nil
list, err := u.mongoDB.GetSubscribedList(ctx, userID)
if err != nil {
return nil, err
}
return list, nil
}
// GetUserStatus get user status

4
pkg/common/db/table/unrelation/user.go
Unescape View File

@ -39,4 +39,8 @@ type UserModelInterface interface {
UnsubscriptionList(ctx context.Context, userID string, userIDList []string) error
// RemoveSubscribedListFromUser Among the unsubscribed users, delete the user from the subscribed list.
RemoveSubscribedListFromUser(ctx context.Context, userID string, userIDList []string) error
// GetAllSubscribeList Get all users subscribed by this user
GetAllSubscribeList(ctx context.Context, id string) (userIDList []string, err error)
// GetSubscribedList Get the user subscribed by those users
GetSubscribedList(ctx context.Context, id string) (userIDList []string, err error)
}

77
pkg/common/db/unrelation/user.go
Unescape View File

@ -21,7 +21,6 @@ import (
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"log"
)
// prefixes and suffixes.
@ -48,22 +47,35 @@ type UserMongoDriver struct {
// AddSubscriptionList Subscriber's handling of thresholds.
func (u *UserMongoDriver) AddSubscriptionList(ctx context.Context, userID string, userIDList []string) error {
// Check the number of lists in the key.
filter := bson.M{SubscriptionPrefix + userID: bson.M{"$size": 1}}
result, err := u.userCollection.Find(context.Background(), filter)
pipeline := mongo.Pipeline{
{{"$match", bson.D{{"user_id", SubscriptionPrefix + userID}}}},
{{"$project", bson.D{{"count", bson.D{{"$size", "user_id_list"}}}}}},
}
// perform aggregate operations
cursor, err := u.userCollection.Aggregate(ctx, pipeline)
if err != nil {
return err
}
var newUserIDList []string
for result.Next(context.Background()) {
err := result.Decode(&newUserIDList)
defer cursor.Close(ctx)
var cnt struct {
Count int `bson:"count"`
}
// iterate over aggregated results
for cursor.Next(ctx) {
err := cursor.Decode(&cnt)
if err != nil {
log.Fatal(err)
return err
}
}
var newUserIDList []string
// If the threshold is exceeded, pop out the previous MaximumSubscription - len(userIDList) and insert it.
if len(newUserIDList)+len(userIDList) > MaximumSubscription {
if cnt.Count+len(userIDList) > MaximumSubscription {
newUserIDList, err = u.GetAllSubscribeList(ctx, userID)
if err != nil {
return err
}
newUserIDList = newUserIDList[MaximumSubscription-len(userIDList):]
_, err := u.userCollection.UpdateOne(
_, err = u.userCollection.UpdateOne(
ctx,
bson.M{"user_id": SubscriptionPrefix + userID},
bson.M{"$set": bson.M{"user_id_list": newUserIDList}},
@ -71,16 +83,17 @@ func (u *UserMongoDriver) AddSubscriptionList(ctx context.Context, userID string
if err != nil {
return err
}
//for i := 1; i <= MaximumSubscription-len(userIDList); i++ {
// _, err := u.userCollection.UpdateOne(
// ctx,
// bson.M{"user_id": SubscriptionPrefix + userID},
// bson.M{SubscriptionPrefix + userID: bson.M{"$pop": -1}},
// )
// if err != nil {
// return err
// }
//}
// Another way to subscribe to N before pop,Delete after testing
/*for i := 1; i <= MaximumSubscription-len(userIDList); i++ {
_, err := u.userCollection.UpdateOne(
ctx,
bson.M{"user_id": SubscriptionPrefix + userID},
bson.M{SubscriptionPrefix + userID: bson.M{"$pop": -1}},
)
if err != nil {
return err
}
}*/
}
upsert := true
opts := &options.UpdateOptions{
@ -139,3 +152,29 @@ func (u *UserMongoDriver) RemoveSubscribedListFromUser(ctx context.Context, user
)
return utils.Wrap(err, "")
}
// GetAllSubscribeList Get all users subscribed by this user
func (u *UserMongoDriver) GetAllSubscribeList(ctx context.Context, userID string) (userIDList []string, err error) {
var user unrelation.UserModel
cursor := u.userCollection.FindOne(
ctx,
bson.M{"user_id": SubscriptionPrefix + userID})
err = cursor.Decode(&user)
if err != nil {
return nil, err
}
return user.UserIDList, nil
}
// GetSubscribedList Get the user subscribed by those users
func (u *UserMongoDriver) GetSubscribedList(ctx context.Context, userID string) (userIDList []string, err error) {
var user unrelation.UserModel
cursor := u.userCollection.FindOne(
ctx,
bson.M{"user_id": SubscribedPrefix + userID})
err = cursor.Decode(&user)
if err != nil {
return nil, err
}
return user.UserIDList, nil
}

Write Preview
Loading…
Cancel
Save