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

msg

Browse Source
This commit is contained in:
wangchuxiao 2023-05-15 11:44:51 +08:00
parent 54a5e06f37
commit dbec5f2191
4 changed files with 16 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
internal/push/push_to_client.go
View File

@ -71,7 +71,6 @@ func (p *Pusher) MsgToUser(ctx context.Context, userID string, msg *sdkws.MsgDat
return err return err
} }
isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush) isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
//log.NewInfo(operationID, "push_result", wsResults, "sendData", msg, "isOfflinePush", isOfflinePush)
log.ZDebug(ctx, "push_result", "ws push result", wsResults, "sendData", msg, "isOfflinePush", isOfflinePush, "push_to_userID", userID) log.ZDebug(ctx, "push_result", "ws push result", wsResults, "sendData", msg, "isOfflinePush", isOfflinePush, "push_to_userID", userID)
p.successCount++ p.successCount++
if isOfflinePush && userID != msg.SendID { if isOfflinePush && userID != msg.SendID {

20
pkg/common/db/controller/common_msg.go
View File

@ -396,23 +396,23 @@ func (db *commonMsgDatabase) refetchDelSeqsMsgs(ctx context.Context, conversatio
func (db *commonMsgDatabase) findMsgBySeq(ctx context.Context, docID string, seqs []int64) (seqMsgs []*sdkws.MsgData, unExistSeqs []int64, err error) { func (db *commonMsgDatabase) findMsgBySeq(ctx context.Context, docID string, seqs []int64) (seqMsgs []*sdkws.MsgData, unExistSeqs []int64, err error) {
beginSeq, endSeq := db.msg.GetSeqsBeginEnd(seqs) beginSeq, endSeq := db.msg.GetSeqsBeginEnd(seqs)
log.ZDebug(ctx, "findMsgBySeq", "docID", docID, "seqs", seqs, "beginSeq", beginSeq, "endSeq", endSeq) msgs, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, docID, beginSeq, endSeq)
msgs, _, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, docID, beginSeq, endSeq)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
for _, seq := range seqs { log.ZDebug(ctx, "findMsgBySeq", "docID", docID, "seqs", seqs, "beginSeq", beginSeq, "endSeq", endSeq, "len(msgs)", len(msgs))
for i, msg := range msgs { seqMsgs = append(seqMsgs, msgs...)
for i, seq := range seqs {
for _, msg := range msgs {
if seq == msg.Seq { if seq == msg.Seq {
seqMsgs = append(seqMsgs, msg) break
continue
}
if i == len(msgs)-1 {
unExistSeqs = append(unExistSeqs, seq)
} }
} }
if i == len(seqs)-1 {
unExistSeqs = append(unExistSeqs, seq)
}
} }
msgs, _, unExistSeqs, err = db.GetMsgAndIndexBySeqsInOneDoc(ctx, docID, seqs) msgs, _, unExistSeqs, err = db.GetMsgAndIndexBySeqsInOneDoc(ctx, docID, unExistSeqs)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }

2
pkg/common/db/table/unrelation/msg.go
View File

@ -31,7 +31,7 @@ type MsgDocModelInterface interface {
Create(ctx context.Context, model *MsgDocModel) error Create(ctx context.Context, model *MsgDocModel) error
UpdateMsgStatusByIndexInOneDoc(ctx context.Context, docID string, msg *sdkws.MsgData, seqIndex int, status int32) error UpdateMsgStatusByIndexInOneDoc(ctx context.Context, docID string, msg *sdkws.MsgData, seqIndex int, status int32) error
FindOneByDocID(ctx context.Context, docID string) (*MsgDocModel, error) FindOneByDocID(ctx context.Context, docID string) (*MsgDocModel, error)
GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, beginSeq, endSeq int64) ([]*sdkws.MsgData, []int64, error) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID string, beginSeq, endSeq int64) ([]*sdkws.MsgData, error)
GetNewestMsg(ctx context.Context, conversationID string) (*MsgInfoModel, error) GetNewestMsg(ctx context.Context, conversationID string) (*MsgInfoModel, error)
GetOldestMsg(ctx context.Context, conversationID string) (*MsgInfoModel, error) GetOldestMsg(ctx context.Context, conversationID string) (*MsgInfoModel, error)
Delete(ctx context.Context, docIDs []string) error Delete(ctx context.Context, docIDs []string) error

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

@ -134,7 +134,7 @@ func (m *MsgMongoDriver) UpdateOneDoc(ctx context.Context, msg *table.MsgDocMode
return err return err
} }
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, err error) {
beginIndex := m.msg.GetMsgIndex(beginSeq) beginIndex := m.msg.GetMsgIndex(beginSeq)
num := endSeq - beginSeq + 1 num := endSeq - beginSeq + 1
pipeline := bson.A{ pipeline := bson.A{
@ -151,7 +151,7 @@ func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID strin
} }
cursor, err := m.MsgCollection.Aggregate(ctx, pipeline) cursor, err := m.MsgCollection.Aggregate(ctx, pipeline)
if err != nil { if err != nil {
return nil, nil, errs.Wrap(err) return nil, errs.Wrap(err)
} }
defer cursor.Close(ctx) defer cursor.Close(ctx)
var doc table.MsgDocModel var doc table.MsgDocModel
@ -159,28 +159,24 @@ func (m *MsgMongoDriver) GetMsgBySeqIndexIn1Doc(ctx context.Context, docID strin
for cursor.Next(ctx) { for cursor.Next(ctx) {
err := cursor.Decode(&doc) err := cursor.Decode(&doc)
if err != nil { if err != nil {
return nil, nil, err return nil, err
} }
if i == 0 { if i == 0 {
break break
} }
} }
if len(doc.Msg) < 1 {
return nil, nil, errs.ErrRecordNotFound.Wrap("mongo GetMsgBySeqIndex failed, len is 0")
}
log.ZDebug(ctx, "msgInfos", "num", len(doc.Msg)) log.ZDebug(ctx, "msgInfos", "num", len(doc.Msg))
for _, v := range doc.Msg { for _, v := range doc.Msg {
var msg sdkws.MsgData var msg sdkws.MsgData
if err := proto.Unmarshal(v.Msg, &msg); err != nil { if err := proto.Unmarshal(v.Msg, &msg); err != nil {
return nil, nil, err return nil, err
} }
if msg.Seq >= beginSeq && msg.Seq <= endSeq { if msg.Seq >= beginSeq && msg.Seq <= endSeq {
log.ZDebug(ctx, "find msg", "msg", &msg) log.ZDebug(ctx, "find msg", "msg", &msg)
msgs = append(msgs, &msg) msgs = append(msgs, &msg)
seqs = append(seqs, msg.Seq)
} else { } else {
log.ZWarn(ctx, "this msg is at wrong position", nil, "msg", &msg) log.ZWarn(ctx, "this msg is at wrong position", nil, "msg", &msg)
} }
} }
return msgs, seqs, nil return msgs, nil
} }