blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-11-05 03:42:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

sync option

Browse Source
This commit is contained in:
withchao 2024-06-06 15:28:03 +08:00
parent c5f565ff20
commit a1523f47e7
11 changed files with 72 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
config/redis.yml
View File

@ -1,6 +1,6 @@
address: [ localhost:16379 ] address: [ 172.16.8.48:16379 ]
username: '' username: ''
password: openIM123 password: openIM123
clusterMode: false clusterMode: false
db: 0 db: 0
maxRetry: 10 maxRetry: 10

26
internal/rpc/friend/sync.go
View File

@ -10,30 +10,6 @@ import (
"github.com/openimsdk/protocol/relation" "github.com/openimsdk/protocol/relation"
) )
//func (s *friendServer) SearchFriends(ctx context.Context, req *pbfriend.SearchFriendsReq) (*pbfriend.SearchFriendsResp, error) {
// if err := s.userRpcClient.Access(ctx, req.UserID); err != nil {
// return nil, err
// }
// if req.Keyword == "" {
// total, friends, err := s.friendDatabase.PageOwnerFriends(ctx, req.UserID, req.Pagination)
// if err != nil {
// return nil, err
// }
// return &pbfriend.SearchFriendsResp{
// Total: total,
// Friends: friendsDB2PB(friends),
// }, nil
// }
// total, friends, err := s.friendDatabase.SearchFriend(ctx, req.UserID, req.Keyword, req.Pagination)
// if err != nil {
// return nil, err
// }
// return &pbfriend.SearchFriendsResp{
// Total: total,
// Friends: friendsDB2PB(friends),
// }, nil
//}
func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.GetIncrementalFriendsReq) (*relation.GetIncrementalFriendsResp, error) { func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.GetIncrementalFriendsReq) (*relation.GetIncrementalFriendsResp, error) {
if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil { if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
return nil, err return nil, err
@ -43,10 +19,8 @@ func (s *friendServer) GetIncrementalFriends(ctx context.Context, req *relation.
VersionKey: req.UserID, VersionKey: req.UserID,
VersionID: req.VersionID, VersionID: req.VersionID,
VersionNumber: req.Version, VersionNumber: req.Version,
SyncLimit: s.config.RpcConfig.FriendSyncCount,
Version: s.friendDatabase.FindFriendIncrVersion, Version: s.friendDatabase.FindFriendIncrVersion,
CacheMaxVersion: s.friendDatabase.FindMaxFriendVersionCache, CacheMaxVersion: s.friendDatabase.FindMaxFriendVersionCache,
SortID: s.friendDatabase.FindSortFriendUserIDs,
Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) { Find: func(ctx context.Context, ids []string) ([]*sdkws.FriendInfo, error) {
return s.getFriend(ctx, req.UserID, ids) return s.getFriend(ctx, req.UserID, ids)
}, },

4
internal/rpc/group/sync.go
View File

@ -15,10 +15,8 @@ func (s *groupServer) GetIncrementalGroupMember(ctx context.Context, req *pbgrou
VersionKey: req.GroupID, VersionKey: req.GroupID,
VersionID: req.VersionID, VersionID: req.VersionID,
VersionNumber: req.Version, VersionNumber: req.Version,
SyncLimit: s.config.RpcConfig.GroupSyncCount,
Version: s.db.FindMemberIncrVersion, Version: s.db.FindMemberIncrVersion,
CacheMaxVersion: s.db.FindMaxGroupMemberVersionCache, CacheMaxVersion: s.db.FindMaxGroupMemberVersionCache,
SortID: s.db.FindSortGroupMemberUserIDs,
Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) { Find: func(ctx context.Context, ids []string) ([]*sdkws.GroupMemberFullInfo, error) {
return s.getGroupMembersInfo(ctx, req.GroupID, ids) return s.getGroupMembersInfo(ctx, req.GroupID, ids)
}, },
@ -46,10 +44,8 @@ func (s *groupServer) GetIncrementalJoinGroup(ctx context.Context, req *pbgroup.
VersionKey: req.UserID, VersionKey: req.UserID,
VersionID: req.VersionID, VersionID: req.VersionID,
VersionNumber: req.Version, VersionNumber: req.Version,
SyncLimit: s.config.RpcConfig.GroupSyncCount,
Version: s.db.FindJoinIncrVersion, Version: s.db.FindJoinIncrVersion,
CacheMaxVersion: s.db.FindMaxJoinGroupVersionCache, CacheMaxVersion: s.db.FindMaxJoinGroupVersionCache,
SortID: s.db.FindSortJoinGroupIDs,
Find: s.getGroupsInfo, Find: s.getGroupsInfo,
ID: func(elem *sdkws.GroupInfo) string { return elem.GroupID }, ID: func(elem *sdkws.GroupInfo) string { return elem.GroupID },
Resp: func(version *model.VersionLog, delIDs []string, list []*sdkws.GroupInfo, full bool) *pbgroup.GetIncrementalJoinGroupResp { Resp: func(version *model.VersionLog, delIDs []string, list []*sdkws.GroupInfo, full bool) *pbgroup.GetIncrementalJoinGroupResp {

45
internal/rpc/incrversion/option.go
View File

@ -16,6 +16,8 @@ import (
// return maxSync // return maxSync
//} //}
const syncLimit = 200
const ( const (
tagQuery = iota + 1 tagQuery = iota + 1
tagFull tagFull
@ -23,17 +25,17 @@ const (
) )
type Option[A, B any] struct { type Option[A, B any] struct {
Ctx context.Context Ctx context.Context
VersionKey string VersionKey string
VersionID string VersionID string
VersionNumber uint64 VersionNumber uint64
SyncLimit int //SyncLimit int
CacheMaxVersion func(ctx context.Context, dId string) (*model.VersionLog, error) CacheMaxVersion func(ctx context.Context, dId string) (*model.VersionLog, error)
Version func(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error) Version func(ctx context.Context, dId string, version uint, limit int) (*model.VersionLog, error)
SortID func(ctx context.Context, dId string) ([]string, error) //SortID func(ctx context.Context, dId string) ([]string, error)
Find func(ctx context.Context, ids []string) ([]A, error) Find func(ctx context.Context, ids []string) ([]A, error)
ID func(elem A) string ID func(elem A) string
Resp func(version *model.VersionLog, delIDs []string, list []A, full bool) *B Resp func(version *model.VersionLog, delIDs []string, list []A, full bool) *B
} }
func (o *Option[A, B]) newError(msg string) error { func (o *Option[A, B]) newError(msg string) error {
@ -47,15 +49,15 @@ func (o *Option[A, B]) check() error {
if o.VersionKey == "" { if o.VersionKey == "" {
return o.newError("versionKey is empty") return o.newError("versionKey is empty")
} }
if o.SyncLimit <= 0 { //if o.SyncLimit <= 0 {
return o.newError("invalid synchronization quantity") // return o.newError("invalid synchronization quantity")
} //}
if o.Version == nil { if o.Version == nil {
return o.newError("func version is nil") return o.newError("func version is nil")
} }
if o.SortID == nil { //if o.SortID == nil {
return o.newError("func allID is nil") // return o.newError("func allID is nil")
} //}
if o.Find == nil { if o.Find == nil {
return o.newError("func find is nil") return o.newError("func find is nil")
} }
@ -81,7 +83,7 @@ func (o *Option[A, B]) getVersion(tag *int) (*model.VersionLog, error) {
if o.CacheMaxVersion == nil { if o.CacheMaxVersion == nil {
if o.validVersion() { if o.validVersion() {
*tag = tagQuery *tag = tagQuery
return o.Version(o.Ctx, o.VersionKey, uint(o.VersionNumber), o.SyncLimit) return o.Version(o.Ctx, o.VersionKey, uint(o.VersionNumber), syncLimit)
} }
*tag = tagFull *tag = tagFull
return o.Version(o.Ctx, o.VersionKey, 0, 0) return o.Version(o.Ctx, o.VersionKey, 0, 0)
@ -103,7 +105,7 @@ func (o *Option[A, B]) getVersion(tag *int) (*model.VersionLog, error) {
return cache, nil return cache, nil
} }
*tag = tagQuery *tag = tagQuery
return o.Version(o.Ctx, o.VersionKey, uint(o.VersionNumber), o.SyncLimit) return o.Version(o.Ctx, o.VersionKey, uint(o.VersionNumber), syncLimit)
} }
} }
@ -131,12 +133,11 @@ func (o *Option[A, B]) Build() (*B, error) {
deleteIDs []string deleteIDs []string
changeIDs []string changeIDs []string
) )
//full := o.VersionID != version.ID.Hex() || version.Full()
if full { if full {
changeIDs, err = o.SortID(o.Ctx, o.VersionKey) //changeIDs, err = o.SortID(o.Ctx, o.VersionKey)
if err != nil { //if err != nil {
return nil, err // return nil, err
} //}
} else { } else {
deleteIDs, changeIDs = version.DeleteAndChangeIDs() deleteIDs, changeIDs = version.DeleteAndChangeIDs()
} }

2
pkg/common/storage/cache/friend.go vendored
View File

@ -39,7 +39,7 @@ type FriendCache interface {
//DelSortFriendUserIDs(ownerUserIDs ...string) FriendCache //DelSortFriendUserIDs(ownerUserIDs ...string) FriendCache
FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) //FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error)
//FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*relationtb.VersionLog, error) //FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*relationtb.VersionLog, error)

4
pkg/common/storage/cache/group.go vendored
View File

@ -59,8 +59,8 @@ type GroupCache interface {
GetGroupMemberNum(ctx context.Context, groupID string) (memberNum int64, err error) GetGroupMemberNum(ctx context.Context, groupID string) (memberNum int64, err error)
DelGroupsMemberNum(groupID ...string) GroupCache DelGroupsMemberNum(groupID ...string) GroupCache
FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) //FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error)
FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) //FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error)
DelMaxGroupMemberVersion(groupIDs ...string) GroupCache DelMaxGroupMemberVersion(groupIDs ...string) GroupCache
DelMaxJoinGroupVersion(userIDs ...string) GroupCache DelMaxJoinGroupVersion(userIDs ...string) GroupCache

20
pkg/common/storage/cache/redis/friend.go vendored
View File

@ -193,16 +193,16 @@ func (f *FriendCacheRedis) DelMaxFriendVersion(ownerUserIDs ...string) cache.Fri
return newFriendCache return newFriendCache
} }
func (f *FriendCacheRedis) FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) { //func (f *FriendCacheRedis) FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) {
userIDs, err := f.GetFriendIDs(ctx, ownerUserID) // userIDs, err := f.GetFriendIDs(ctx, ownerUserID)
if err != nil { // if err != nil {
return nil, err // return nil, err
} // }
if len(userIDs) > f.syncCount { // if len(userIDs) > f.syncCount {
userIDs = userIDs[:f.syncCount] // userIDs = userIDs[:f.syncCount]
} // }
return userIDs, nil // return userIDs, nil
} //}
func (f *FriendCacheRedis) FindMaxFriendVersion(ctx context.Context, ownerUserID string) (*model.VersionLog, error) { func (f *FriendCacheRedis) FindMaxFriendVersion(ctx context.Context, ownerUserID string) (*model.VersionLog, error) {
return getCache(ctx, f.rcClient, f.getFriendMaxVersionKey(ownerUserID), f.expireTime, func(ctx context.Context) (*model.VersionLog, error) { return getCache(ctx, f.rcClient, f.getFriendMaxVersionKey(ownerUserID), f.expireTime, func(ctx context.Context) (*model.VersionLog, error) {

42
pkg/common/storage/cache/redis/group.go vendored
View File

@ -405,27 +405,27 @@ func (g *GroupCacheRedis) FindGroupMemberUser(ctx context.Context, groupIDs []st
}) })
} }
func (g *GroupCacheRedis) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) { //func (g *GroupCacheRedis) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) {
userIDs, err := g.GetGroupMemberIDs(ctx, groupID) // userIDs, err := g.GetGroupMemberIDs(ctx, groupID)
if err != nil { // if err != nil {
return nil, err // return nil, err
} // }
if len(userIDs) > g.syncCount { // if len(userIDs) > g.syncCount {
userIDs = userIDs[:g.syncCount] // userIDs = userIDs[:g.syncCount]
} // }
return userIDs, nil // return userIDs, nil
} //}
//
func (g *GroupCacheRedis) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) { //func (g *GroupCacheRedis) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) {
groupIDs, err := g.GetJoinedGroupIDs(ctx, userID) // groupIDs, err := g.GetJoinedGroupIDs(ctx, userID)
if err != nil { // if err != nil {
return nil, err // return nil, err
} // }
if len(groupIDs) > g.syncCount { // if len(groupIDs) > g.syncCount {
groupIDs = groupIDs[:g.syncCount] // groupIDs = groupIDs[:g.syncCount]
} // }
return groupIDs, nil // return groupIDs, nil
} //}
func (g *GroupCacheRedis) DelMaxGroupMemberVersion(groupIDs ...string) cache.GroupCache { func (g *GroupCacheRedis) DelMaxGroupMemberVersion(groupIDs ...string) cache.GroupCache {
keys := make([]string, 0, len(groupIDs)) keys := make([]string, 0, len(groupIDs))

8
pkg/common/storage/controller/friend.go
View File

@ -78,7 +78,7 @@ type FriendDatabase interface {
// UpdateFriends updates fields for friends // UpdateFriends updates fields for friends
UpdateFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, val map[string]any) (err error) UpdateFriends(ctx context.Context, ownerUserID string, friendUserIDs []string, val map[string]any) (err error)
FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) //FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error)
FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*model.VersionLog, error) FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*model.VersionLog, error)
@ -360,9 +360,9 @@ func (f *friendDatabase) UpdateFriends(ctx context.Context, ownerUserID string,
}) })
} }
func (f *friendDatabase) FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) { //func (f *friendDatabase) FindSortFriendUserIDs(ctx context.Context, ownerUserID string) ([]string, error) {
return f.cache.FindSortFriendUserIDs(ctx, ownerUserID) // return f.cache.FindSortFriendUserIDs(ctx, ownerUserID)
} //}
func (f *friendDatabase) FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*model.VersionLog, error) { func (f *friendDatabase) FindFriendIncrVersion(ctx context.Context, ownerUserID string, version uint, limit int) (*model.VersionLog, error) {
return f.friend.FindIncrVersion(ctx, ownerUserID, version, limit) return f.friend.FindIncrVersion(ctx, ownerUserID, version, limit)

19
pkg/common/storage/controller/group.go
View File

@ -110,9 +110,8 @@ type GroupDatabase interface {
FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error) FindMemberIncrVersion(ctx context.Context, groupID string, version uint, limit int) (*model.VersionLog, error)
FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error) FindJoinIncrVersion(ctx context.Context, userID string, version uint, limit int) (*model.VersionLog, error)
FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) //FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error)
//FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error)
FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error)
FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error) FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error)
FindMaxJoinGroupVersionCache(ctx context.Context, userID string) (*model.VersionLog, error) FindMaxJoinGroupVersionCache(ctx context.Context, userID string) (*model.VersionLog, error)
@ -497,13 +496,13 @@ func (g *groupDatabase) FindJoinIncrVersion(ctx context.Context, userID string,
return g.groupMemberDB.FindJoinIncrVersion(ctx, userID, version, limit) return g.groupMemberDB.FindJoinIncrVersion(ctx, userID, version, limit)
} }
func (g *groupDatabase) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) { //func (g *groupDatabase) FindSortGroupMemberUserIDs(ctx context.Context, groupID string) ([]string, error) {
return g.cache.FindSortGroupMemberUserIDs(ctx, groupID) // return g.cache.FindSortGroupMemberUserIDs(ctx, groupID)
} //}
//
func (g *groupDatabase) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) { //func (g *groupDatabase) FindSortJoinGroupIDs(ctx context.Context, userID string) ([]string, error) {
return g.cache.FindSortJoinGroupIDs(ctx, userID) // return g.cache.FindSortJoinGroupIDs(ctx, userID)
} //}
func (g *groupDatabase) FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error) { func (g *groupDatabase) FindMaxGroupMemberVersionCache(ctx context.Context, groupID string) (*model.VersionLog, error) {
return g.cache.FindMaxGroupMemberVersion(ctx, groupID) return g.cache.FindMaxGroupMemberVersion(ctx, groupID)

2
pkg/common/storage/model/friend.go
View File

@ -22,8 +22,6 @@ import (
type Friend struct { type Friend struct {
OwnerUserID string `bson:"owner_user_id"` OwnerUserID string `bson:"owner_user_id"`
FriendUserID string `bson:"friend_user_id"` FriendUserID string `bson:"friend_user_id"`
FriendNickname string `bson:"friend_nickname"`
FriendFaceURL string `bson:"friend_face_url"`
Remark string `bson:"remark"` Remark string `bson:"remark"`
CreateTime time.Time `bson:"create_time"` CreateTime time.Time `bson:"create_time"`
AddSource int32 `bson:"add_source"` AddSource int32 `bson:"add_source"`