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

feat: msg rpc local cache

Browse Source
This commit is contained in:
withchao
2024-01-17 16:37:26 +08:00
parent 025eb44113
commit c16f17582a
19 changed files with 242 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/rpc/msg/as_read.go
+3 -3
View File
@@ -43,7 +43,7 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m
if err != nil {
return nil, err
}
conversations, err := m.Conversation.GetConversations(ctx, req.UserID, conversationIDs)
conversations, err := m.ConversationLocalCache.GetConversations(ctx, req.UserID, conversationIDs)
if err != nil {
return nil, err
}
@@ -106,7 +106,7 @@ func (m *msgServer) MarkMsgsAsRead(
if hasReadSeq > maxSeq {
return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq")
}
conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
if err != nil {
return
}
@@ -135,7 +135,7 @@ func (m *msgServer) MarkConversationAsRead(
ctx context.Context,
req *msg.MarkConversationAsReadReq,
) (resp *msg.MarkConversationAsReadResp, err error) {
conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
if err != nil {
return nil, err
}
internal/rpc/msg/revoke.go
+2 -7
View File
@@ -47,7 +47,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
if err := authverify.CheckAccessV3(ctx, req.UserID); err != nil {
return nil, err
}
user, err := m.User.GetUserInfo(ctx, req.UserID)
user, err := m.UserLocalCache.GetUserInfo(ctx, req.UserID)
if err != nil {
return nil, err
}
@@ -73,12 +73,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
}
role = user.AppMangerLevel
case constant.SuperGroupChatType:
members, err := m.Group.GetGroupMemberInfoMap(
ctx,
msgs[0].GroupID,
utils.Distinct([]string{req.UserID, msgs[0].SendID}),
true,
)
members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID}))
if err != nil {
return nil, err
}
internal/rpc/msg/send.go
+1 -1
View File
@@ -98,7 +98,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
}
tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList)
if tagAll {
memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
if err != nil {
log.ZWarn(ctx, "GetGroupMemberIDs", err)
return
internal/rpc/msg/server.go
+2 -4
View File
@@ -36,9 +36,8 @@ type (
msgServer struct {
RegisterCenter discoveryregistry.SvcDiscoveryRegistry
MsgDatabase controller.CommonMsgDatabase
Group *rpcclient.GroupRpcClient
User *rpcclient.UserRpcClient
Conversation *rpcclient.ConversationRpcClient
UserLocalCache *rpccache.UserLocalCache
FriendLocalCache *rpccache.FriendLocalCache
GroupLocalCache *rpccache.GroupLocalCache
ConversationLocalCache *rpccache.ConversationLocalCache
@@ -84,10 +83,9 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
s := &msgServer{
Conversation: &conversationClient,
User: &userRpcClient,
Group: &groupRpcClient,
MsgDatabase: msgDatabase,
RegisterCenter: client,
UserLocalCache: rpccache.NewUserLocalCache(userRpcClient, rdb),
GroupLocalCache: rpccache.NewGroupLocalCache(groupRpcClient, rdb),
ConversationLocalCache: rpccache.NewConversationLocalCache(conversationClient, rdb),
FriendLocalCache: rpccache.NewFriendLocalCache(friendRpcClient, rdb),
internal/rpc/msg/statistics.go
+2 -2
View File
@@ -41,7 +41,7 @@ func (m *msgServer) GetActiveUser(ctx context.Context, req *msg.GetActiveUserReq
var pbUsers []*msg.ActiveUser
if len(users) > 0 {
userIDs := utils.Slice(users, func(e *unrelation.UserCount) string { return e.UserID })
userMap, err := m.User.GetUsersInfoMap(ctx, userIDs)
userMap, err := m.UserLocalCache.GetUsersInfoMap(ctx, userIDs)
if err != nil {
return nil, err
}
@@ -83,7 +83,7 @@ func (m *msgServer) GetActiveGroup(ctx context.Context, req *msg.GetActiveGroupR
var pbgroups []*msg.ActiveGroup
if len(groups) > 0 {
groupIDs := utils.Slice(groups, func(e *unrelation.GroupCount) string { return e.GroupID })
resp, err := m.Group.GetGroupInfos(ctx, groupIDs, false)
resp, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
if err != nil {
return nil, err
}
internal/rpc/msg/sync_msg.go
+4 -4
View File
@@ -37,7 +37,7 @@ func (m *msgServer) PullMessageBySeqs(
resp.NotificationMsgs = make(map[string]*sdkws.PullMsgs)
for _, seq := range req.SeqRanges {
if !msgprocessor.IsNotification(seq.ConversationID) {
conversation, err := m.Conversation.GetConversation(ctx, req.UserID, seq.ConversationID)
conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, seq.ConversationID)
if err != nil {
log.ZError(ctx, "GetConversation error", err, "conversationID", seq.ConversationID)
continue
@@ -140,7 +140,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
}
}
if len(sendIDs) != 0 {
sendInfos, err := m.User.GetUsersInfo(ctx, sendIDs)
sendInfos, err := m.UserLocalCache.GetUsersInfo(ctx, sendIDs)
if err != nil {
return nil, err
}
@@ -149,7 +149,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
}
}
if len(recvIDs) != 0 {
recvInfos, err := m.User.GetUsersInfo(ctx, recvIDs)
recvInfos, err := m.UserLocalCache.GetUsersInfo(ctx, recvIDs)
if err != nil {
return nil, err
}
@@ -158,7 +158,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
}
}
if len(groupIDs) != 0 {
groupInfos, err := m.Group.GetGroupInfos(ctx, groupIDs, true)
groupInfos, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
if err != nil {
return nil, err
}
internal/rpc/msg/verify.go
+4 -4
View File
@@ -78,7 +78,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
}
return nil
case constant.SuperGroupChatType:
groupInfo, err := m.Group.GetGroupInfoCache(ctx, data.MsgData.GroupID)
groupInfo, err := m.GroupLocalCache.GetGroupInfo(ctx, data.MsgData.GroupID)
if err != nil {
return err
}
@@ -104,9 +104,9 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
return errs.ErrNotInGroupYet.Wrap()
}
groupMemberInfo, err := m.Group.GetGroupMemberCache(ctx, data.MsgData.GroupID, data.MsgData.SendID)
groupMemberInfo, err := m.GroupLocalCache.GetGroupMember(ctx, data.MsgData.GroupID, data.MsgData.SendID)
if err != nil {
if err == errs.ErrRecordNotFound {
if errs.ErrRecordNotFound.Is(err) {
return errs.ErrNotInGroupYet.Wrap(err.Error())
}
return err
@@ -188,7 +188,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
pb *msg.SendMsgReq,
) (bool, error) {
defer log.ZDebug(ctx, "modifyMessageByUserMessageReceiveOpt return")
opt, err := m.User.GetUserGlobalMsgRecvOpt(ctx, userID) // todo local cache
opt, err := m.UserLocalCache.GetUserGlobalMsgRecvOpt(ctx, userID)
if err != nil {
return false, err
}