|
|
@ -73,14 +73,14 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(ctx context.Cont
|
|
|
|
} else {
|
|
|
|
} else {
|
|
|
|
prommetrics.MsgInsertMongoSuccessCounter.Inc()
|
|
|
|
prommetrics.MsgInsertMongoSuccessCounter.Inc()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
var seqs []int64
|
|
|
|
//var seqs []int64
|
|
|
|
for _, msg := range msgFromMQ.MsgData {
|
|
|
|
//for _, msg := range msgFromMQ.MsgData {
|
|
|
|
seqs = append(seqs, msg.Seq)
|
|
|
|
// seqs = append(seqs, msg.Seq)
|
|
|
|
}
|
|
|
|
//}
|
|
|
|
if err := mc.msgTransferDatabase.DeleteMessagesFromCache(ctx, msgFromMQ.ConversationID, seqs); err != nil {
|
|
|
|
//if err := mc.msgTransferDatabase.DeleteMessagesFromCache(ctx, msgFromMQ.ConversationID, seqs); err != nil {
|
|
|
|
log.ZError(ctx, "remove cache msg from redis err", err, "msg",
|
|
|
|
// log.ZError(ctx, "remove cache msg from redis err", err, "msg",
|
|
|
|
msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
|
|
|
|
// msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID)
|
|
|
|
}
|
|
|
|
//}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
|
|
|
func (*OnlineHistoryMongoConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
|
|
|