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

PullMessageBySeqs

Browse Source
This commit is contained in:
withchao 2023-05-31 11:49:47 +08:00
parent 634247a5c7
commit 765a33db1f
2 changed files with 27 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
internal/rpc/msg/sync_msg.go
View File

@ -15,7 +15,7 @@ func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessag
resp.NotificationMsgs = make(map[string]*sdkws.PullMsgs) resp.NotificationMsgs = make(map[string]*sdkws.PullMsgs)
for _, seq := range req.SeqRanges { for _, seq := range req.SeqRanges {
if !utils.IsNotification(seq.ConversationID) { if !utils.IsNotification(seq.ConversationID) {
minSeq, msgs, err := m.MsgDatabase.GetMsgBySeqsRange(ctx, req.UserID, seq.ConversationID, seq.Begin, seq.End, seq.Num) minSeq, maxSeq, msgs, err := m.MsgDatabase.GetMsgBySeqsRange(ctx, req.UserID, seq.ConversationID, seq.Begin, seq.End, seq.Num)
if err != nil { if err != nil {
log.ZWarn(ctx, "GetMsgBySeqsRange error", err, "conversationID", seq.ConversationID, "seq", seq) log.ZWarn(ctx, "GetMsgBySeqsRange error", err, "conversationID", seq.ConversationID, "seq", seq)
continue continue
@ -23,11 +23,6 @@ func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessag
var isEnd bool var isEnd bool
switch req.Order { switch req.Order {
case sdkws.PullOrder_PullOrderAsc: case sdkws.PullOrder_PullOrderAsc:
maxSeq, err := m.MsgDatabase.GetMaxSeq(ctx, seq.ConversationID)
if err != nil {
log.ZError(ctx, "GetMaxSeq error", err, "conversationID", seq.ConversationID)
continue
}
isEnd = maxSeq <= seq.End isEnd = maxSeq <= seq.End
case sdkws.PullOrder_PullOrderDesc: case sdkws.PullOrder_PullOrderDesc:
isEnd = seq.Begin <= minSeq isEnd = seq.Begin <= minSeq
@ -38,16 +33,11 @@ func (m *msgServer) PullMessageBySeqs(ctx context.Context, req *sdkws.PullMessag
for i := seq.Begin; i <= seq.End; i++ { for i := seq.Begin; i <= seq.End; i++ {
seqs = append(seqs, i) seqs = append(seqs, i)
} }
minSeq, notificationMsgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, req.UserID, seq.ConversationID, seqs) minSeq, maxSeq, notificationMsgs, err := m.MsgDatabase.GetMsgBySeqs(ctx, req.UserID, seq.ConversationID, seqs)
if err != nil { if err != nil {
log.ZWarn(ctx, "GetMsgBySeqs error", err, "conversationID", seq.ConversationID, "seq", seq) log.ZWarn(ctx, "GetMsgBySeqs error", err, "conversationID", seq.ConversationID, "seq", seq)
continue continue
} }
maxSeq, err := m.MsgDatabase.GetMaxSeq(ctx, seq.ConversationID)
if err != nil {
log.ZError(ctx, "GetMaxSeq error", err, "conversationID", seq.ConversationID)
continue
}
var isEnd bool var isEnd bool
switch req.Order { switch req.Order {
case sdkws.PullOrder_PullOrderAsc: case sdkws.PullOrder_PullOrderAsc:

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

@ -41,9 +41,9 @@ type CommonMsgDatabase interface {
BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNewConversation bool, err error)
// 通过seqList获取mongo中写扩散消息 // 通过seqList获取mongo中写扩散消息
GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num int64) (minSeq int64, seqMsg []*sdkws.MsgData, err error) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
// 通过seqList获取大群在 mongo里面的消息 // 通过seqList获取大群在 mongo里面的消息
GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, seqMsg []*sdkws.MsgData, err error) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (minSeq int64, maxSeq int64, seqMsg []*sdkws.MsgData, err error)
// 删除会话消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache) // 删除会话消息重置最小seq， remainTime为消息保留的时间单位秒,超时消息删除， 传0删除所有消息(此方法不删除redis cache)
DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error
// 用户根据seq删除消息 // 用户根据seq删除消息
@ -448,26 +448,32 @@ func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID strin
return seqMsgs, nil return seqMsgs, nil
} }
func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num int64) (int64, []*sdkws.MsgData, error) { func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID string, conversationID string, begin, end, num int64) (int64, int64, []*sdkws.MsgData, error) {
userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID) userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
if err != nil && errs.Unwrap(err) != redis.Nil { if err != nil && errs.Unwrap(err) != redis.Nil {
return 0, nil, err return 0, 0, nil, err
} }
minSeq, err := db.cache.GetMinSeq(ctx, conversationID) minSeq, err := db.cache.GetMinSeq(ctx, conversationID)
if err != nil && errs.Unwrap(err) != redis.Nil { if err != nil && errs.Unwrap(err) != redis.Nil {
return 0, nil, err return 0, 0, nil, err
} }
if userMinSeq < minSeq { if userMinSeq < minSeq {
minSeq = userMinSeq minSeq = userMinSeq
} }
if minSeq > end { if minSeq > end {
log.ZInfo(ctx, "minSeq > end", "minSeq", minSeq, "end", end) log.ZInfo(ctx, "minSeq > end", "minSeq", minSeq, "end", end)
return 0, nil, nil return 0, 0, nil, nil
}
maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
if err != nil && errs.Unwrap(err) != redis.Nil {
return 0, 0, nil, err
} }
if begin < minSeq { if begin < minSeq {
begin = minSeq begin = minSeq
} }
if end > maxSeq {
end = maxSeq
}
var seqs []int64 var seqs []int64
for i := end; i > end-num; i-- { for i := end; i > end-num; i-- {
if i >= begin { if i >= begin {
@ -492,29 +498,33 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
mongoMsgs, err := db.getMsgBySeqsRange(ctx, userID, conversationID, failedSeqs, begin, end) mongoMsgs, err := db.getMsgBySeqsRange(ctx, userID, conversationID, failedSeqs, begin, end)
if err != nil { if err != nil {
prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs)) prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs))
return 0, nil, err return 0, 0, nil, err
} }
prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs)) prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs))
successMsgs = append(successMsgs, mongoMsgs...) successMsgs = append(successMsgs, mongoMsgs...)
} }
return minSeq, successMsgs, nil return minSeq, maxSeq, successMsgs, nil
} }
func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, []*sdkws.MsgData, error) { func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID) userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
if err != nil && errs.Unwrap(err) != redis.Nil { if err != nil && errs.Unwrap(err) != redis.Nil {
return 0, nil, err return 0, 0, nil, err
} }
minSeq, err := db.cache.GetMinSeq(ctx, conversationID) minSeq, err := db.cache.GetMinSeq(ctx, conversationID)
if err != nil && errs.Unwrap(err) != redis.Nil { if err != nil && errs.Unwrap(err) != redis.Nil {
return 0, nil, err return 0, 0, nil, err
}
maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
if err != nil && errs.Unwrap(err) != redis.Nil {
return 0, 0, nil, err
} }
if userMinSeq < minSeq { if userMinSeq < minSeq {
minSeq = userMinSeq minSeq = userMinSeq
} }
var newSeqs []int64 var newSeqs []int64
for _, seq := range seqs { for _, seq := range seqs {
if seq >= minSeq { if seq >= minSeq && seq <= maxSeq {
newSeqs = append(newSeqs, seq) newSeqs = append(newSeqs, seq)
} }
} }
@ -531,12 +541,12 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs) mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
if err != nil { if err != nil {
prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs)) prome.Add(prome.MsgPullFromMongoFailedCounter, len(failedSeqs))
return 0, nil, err return 0, 0, nil, err
} }
prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs)) prome.Add(prome.MsgPullFromMongoSuccessCounter, len(mongoMsgs))
successMsgs = append(successMsgs, mongoMsgs...) successMsgs = append(successMsgs, mongoMsgs...)
} }
return minSeq, successMsgs, nil return minSeq, maxSeq, successMsgs, nil
} }
func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error { func (db *commonMsgDatabase) DeleteConversationMsgsAndSetMinSeq(ctx context.Context, conversationID string, remainTime int64) error {