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

pipeline

Browse Source
This commit is contained in:
wangchuxiao 2023-05-11 15:30:06 +08:00
parent 89d5f7bd45
commit 351c5bacfa
2 changed files with 27 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
pkg/common/db/cache/msg.go vendored
View File

@ -3,7 +3,6 @@ package cache
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"strconv" "strconv"
"time" "time"
@ -290,7 +289,7 @@ func (c *msgCache) GetMessagesBySeq(ctx context.Context, conversationID string,
func (c *msgCache) SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) { func (c *msgCache) SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) {
pipe := c.rdb.Pipeline() pipe := c.rdb.Pipeline()
var failedMsgs []sdkws.MsgData var failedMsgs []*sdkws.MsgData
for _, msg := range msgs { for _, msg := range msgs {
key := c.getMessageCacheKey(conversationID, msg.Seq) key := c.getMessageCacheKey(conversationID, msg.Seq)
s, err := utils.Pb2String(msg) s, err := utils.Pb2String(msg)
@ -299,14 +298,12 @@ func (c *msgCache) SetMessageToCache(ctx context.Context, conversationID string,
} }
err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err() err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err()
if err != nil { if err != nil {
return 0, errs.Wrap(err) failedMsgs = append(failedMsgs, msg)
log.ZWarn(ctx, "set msg 2 cache failed", err, "msg", failedMsgs)
} }
} }
if len(failedMsgs) != 0 {
return len(failedMsgs), fmt.Errorf("set msg to msgCache failed, failed lists: %v, %s", failedMsgs, conversationID)
}
_, err := pipe.Exec(ctx) _, err := pipe.Exec(ctx)
return 0, err return len(failedMsgs), err
} }
func (c *msgCache) DeleteMessageFromCache(ctx context.Context, userID string, msgList []*sdkws.MsgData) error { func (c *msgCache) DeleteMessageFromCache(ctx context.Context, userID string, msgList []*sdkws.MsgData) error {

27
pkg/common/db/unrelation/msg.go
View File

@ -137,12 +137,31 @@ func (m *MsgMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.MsgDocMode
func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, beginSeq, endSeq int64) (msgs []*sdkws.MsgData, seqs []int64, err error) { func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, beginSeq, endSeq int64) (msgs []*sdkws.MsgData, seqs []int64, err error) {
beginIndex := m.msg.GetMsgIndex(beginSeq) beginIndex := m.msg.GetMsgIndex(beginSeq)
num := endSeq - beginSeq + 1 num := endSeq - beginSeq + 1
result, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": docID, "msgs": bson.M{"$slice": []int64{beginIndex, num}}})
if err != nil { pipeline := bson.A{
return nil, nil, err bson.M{
"$match": bson.M{"doc_id": docID},
},
bson.M{
"$project": bson.M{
"doc_id": 1,
"msgs": bson.M{
"$slice": []interface{}{"$msgs", beginIndex, num},
},
},
},
} }
cursor, err := m.MsgCollection.Aggregate(ctx, pipeline)
if err != nil {
return nil, nil, errs.Wrap(err)
}
// result, err := m.MsgCollection.Find(ctx, bson.M{"doc_id": docID, "msgs": bson.M{"$slice": []int64{beginIndex, num}}})
// if err != nil {
// return nil, nil, err
// }
var msgInfos []table.MsgInfoModel var msgInfos []table.MsgInfoModel
if err := result.Decode(&msgInfos); err != nil { if err := cursor.All(ctx, &msgInfos); err != nil {
return nil, nil, err return nil, nil, err
} }
if len(msgInfos) < 1 { if len(msgInfos) < 1 {