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

msg send pull

Browse Source
This commit is contained in:
withchao 2023-02-13 16:36:05 +08:00
parent 6b47c471f7
commit 884d28b9a8
2 changed files with 103 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

184
internal/rpc/msg/send_pull.go
View File

@ -8,7 +8,6 @@ import (
"Open_IM/pkg/proto/sdkws" "Open_IM/pkg/proto/sdkws"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
go_redis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"strings" "strings"
"sync" "sync"
@ -262,98 +261,109 @@ func (m *msgServer) SendMsg(ctx context.Context, req *msg.SendMsgReq) (resp *msg
} }
} }
func (m *msgServer) GetMaxAndMinSeq(_ context.Context, in *sdkws.GetMaxAndMinSeqReq) (*sdkws.GetMaxAndMinSeqResp, error) { func (m *msgServer) GetMaxAndMinSeq(ctx context.Context, req *sdkws.GetMaxAndMinSeqReq) (*sdkws.GetMaxAndMinSeqResp, error) {
log.NewInfo(in.OperationID, "rpc getMaxAndMinSeq is arriving", in.String())
resp := new(sdkws.GetMaxAndMinSeqResp) resp := new(sdkws.GetMaxAndMinSeqResp)
m := make(map[string]*sdkws.MaxAndMinSeq) m2 := make(map[string]*sdkws.MaxAndMinSeq)
var maxSeq, minSeq uint64 maxSeq, err := m.MsgInterface.GetUserMaxSeq(ctx, req.UserID)
var err1, err2 error
maxSeq, err1 = commonDB.DB.GetUserMaxSeq(in.UserID)
minSeq, err2 = commonDB.DB.GetUserMinSeq(in.UserID)
if (err1 != nil && err1 != go_redis.Nil) || (err2 != nil && err2 != go_redis.Nil) {
log.NewError(in.OperationID, "getMaxSeq from redis error", in.String())
if err1 != nil {
log.NewError(in.OperationID, utils.GetSelfFuncName(), err1.Error())
}
if err2 != nil {
log.NewError(in.OperationID, utils.GetSelfFuncName(), err2.Error())
}
resp.ErrCode = 200
resp.ErrMsg = "redis get err"
return resp, nil
}
resp.MaxSeq = uint32(maxSeq)
resp.MinSeq = uint32(minSeq)
for _, groupID := range in.GroupIDList {
x := new(sdkws.MaxAndMinSeq)
maxSeq, _ := commonDB.DB.GetGroupMaxSeq(groupID)
minSeq, _ := commonDB.DB.GetGroupUserMinSeq(groupID, in.UserID)
x.MaxSeq = uint32(maxSeq)
x.MinSeq = uint32(minSeq)
m[groupID] = x
}
resp.GroupMaxAndMinSeq = m
return resp, nil
}
func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *sdkws.PullMessageBySeqListReq) (*sdkws.PullMessageBySeqListResp, error) {
log.NewInfo(in.OperationID, "rpc PullMessageBySeqList is arriving", in.String())
resp := new(sdkws.PullMessageBySeqListResp)
m := make(map[string]*sdkws.MsgDataList)
redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(in.UserID, in.SeqList, in.OperationID)
if err != nil { if err != nil {
if err != go_redis.Nil { return nil, err
promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
log.Error(in.OperationID, "get message from redis exception", err.Error(), failedSeqList)
} else {
log.Debug(in.OperationID, "get message from redis is nil", failedSeqList)
}
msgList, err1 := commonDB.DB.GetMsgBySeqListMongo2(in.UserID, failedSeqList, in.OperationID)
if err1 != nil {
promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err1.Error())
resp.ErrCode = 201
resp.ErrMsg = err1.Error()
return resp, nil
} else {
promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
redisMsgList = append(redisMsgList, msgList...)
resp.List = redisMsgList
}
} else {
promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
resp.List = redisMsgList
} }
minSeq, err := m.MsgInterface.GetUserMinSeq(ctx, req.UserID)
for k, v := range in.GroupSeqList { if err != nil {
x := new(sdkws.MsgDataList) return nil, err
redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(k, v.SeqList, in.OperationID) }
if err != nil { resp.MaxSeq = maxSeq
if err != go_redis.Nil { resp.MinSeq = minSeq
promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList)) if len(req.GroupIDList) > 0 {
log.Error(in.OperationID, "get message from redis exception", err.Error(), failedSeqList) resp.GroupMaxAndMinSeq = make(map[string]*sdkws.MaxAndMinSeq)
} else { for _, groupID := range req.GroupIDList {
log.Debug(in.OperationID, "get message from redis is nil", failedSeqList) maxSeq, err := m.MsgInterface.GetGroupMaxSeq(ctx, groupID)
if err != nil {
return nil, err
} }
msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(k, failedSeqList, in.OperationID) minSeq, err := m.MsgInterface.GetGroupMinSeq(ctx, groupID)
if err1 != nil { if err != nil {
promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList)) return nil, err
log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err1.Error()) }
resp.ErrCode = 201 m2[groupID] = &sdkws.MaxAndMinSeq{
resp.ErrMsg = err1.Error() MaxSeq: maxSeq,
return resp, nil MinSeq: minSeq,
} else {
promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
redisMsgList = append(redisMsgList, msgList...)
x.MsgDataList = redisMsgList
m[k] = x
} }
} else {
promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
x.MsgDataList = redisMsgList
m[k] = x
} }
} }
resp.GroupMsgDataList = m return resp, nil
}
func (m *msgServer) PullMessageBySeqList(ctx context.Context, req *sdkws.PullMessageBySeqListReq) (*sdkws.PullMessageBySeqListResp, error) {
resp := &sdkws.PullMessageBySeqListResp{GroupMsgDataList: make(map[string]*sdkws.MsgDataList)}
msgs, err := m.MsgInterface.GetMessageListBySeq(ctx, req.UserID, req.SeqList)
if err != nil {
return nil, err
}
resp.List = msgs
for userID, list := range req.GroupSeqList {
msgs, err := m.MsgInterface.GetMessageListBySeq(ctx, userID, list.SeqList)
if err != nil {
return nil, err
}
resp.GroupMsgDataList[userID] = &sdkws.MsgDataList{
MsgDataList: msgs,
}
}
//redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(req.UserID, req.SeqList, req.OperationID)
//if err != nil {
// if err != go_redis.Nil {
// promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
// log.Error(req.OperationID, "get message from redis exception", err.Error(), failedSeqList)
// } else {
// log.Debug(req.OperationID, "get message from redis is nil", failedSeqList)
// }
// msgList, err1 := commonDB.DB.GetMsgBySeqListMongo2(req.UserID, failedSeqList, req.OperationID)
// if err1 != nil {
// promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
// log.Error(req.OperationID, "PullMessageBySeqList data error", req.String(), err1.Error())
// resp.ErrCode = 201
// resp.ErrMsg = err1.Error()
// return resp, nil
// } else {
// promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
// redisMsgList = append(redisMsgList, msgList...)
// resp.List = redisMsgList
// }
//} else {
// promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
// resp.List = redisMsgList
//}
//
//for k, v := range req.GroupSeqList {
// x := new(sdkws.MsgDataList)
// redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(k, v.SeqList, req.OperationID)
// if err != nil {
// if err != go_redis.Nil {
// promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
// log.Error(req.OperationID, "get message from redis exception", err.Error(), failedSeqList)
// } else {
// log.Debug(req.OperationID, "get message from redis is nil", failedSeqList)
// }
// msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(k, failedSeqList, req.OperationID)
// if err1 != nil {
// promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
// log.Error(req.OperationID, "PullMessageBySeqList data error", req.String(), err1.Error())
// resp.ErrCode = 201
// resp.ErrMsg = err1.Error()
// return resp, nil
// } else {
// promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
// redisMsgList = append(redisMsgList, msgList...)
// x.MsgDataList = redisMsgList
// m[k] = x
// }
// } else {
// promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
// x.MsgDataList = redisMsgList
// m[k] = x
// }
//}
return resp, nil return resp, nil
} }

6
pkg/common/db/controller/msg.go
View File

@ -3,6 +3,7 @@ package controller
import ( import (
"Open_IM/pkg/proto/msg" "Open_IM/pkg/proto/msg"
pbMsg "Open_IM/pkg/proto/msg" pbMsg "Open_IM/pkg/proto/msg"
"Open_IM/pkg/proto/sdkws"
"context" "context"
) )
@ -25,8 +26,13 @@ type MsgInterface interface {
// delete // delete
DelMsgFromCache(ctx context.Context, userID string, seqs []uint32) error DelMsgFromCache(ctx context.Context, userID string, seqs []uint32) error
GetGroupMaxSeq(ctx context.Context, groupID string) (uint32, error) GetGroupMaxSeq(ctx context.Context, groupID string) (uint32, error)
GetGroupMinSeq(ctx context.Context, groupID string) (uint32, error)
SetGroupUserMinSeq(ctx context.Context, groupID string, seq uint32) error SetGroupUserMinSeq(ctx context.Context, groupID string, seq uint32) error
DelUserAllSeq(ctx context.Context, userID string) error // redis and mongodb DelUserAllSeq(ctx context.Context, userID string) error // redis and mongodb
GetUserMaxSeq(ctx context.Context, userID string) (uint32, error)
GetUserMinSeq(ctx context.Context, userID string) (uint32, error)
GetMessageListBySeq(ctx context.Context, userID string, seqs []uint32) ([]*sdkws.MsgData, error)
} }
type MsgDatabaseInterface interface { type MsgDatabaseInterface interface {