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

mysql storage

Browse Source
This commit is contained in:
wangchuxiao 2023-05-19 16:20:20 +08:00
parent d0aa0a8772
commit 37dd079129
1 changed files with 2 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/msgtransfer/persistent_msg_handler.go
View File

@ -39,15 +39,14 @@ func NewPersistentConsumerHandler(database controller.ChatLogDatabase) *Persiste
func (pc *PersistentConsumerHandler) handleChatWs2Mysql(ctx context.Context, cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) { func (pc *PersistentConsumerHandler) handleChatWs2Mysql(ctx context.Context, cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
msg := cMsg.Value msg := cMsg.Value
operationID := mcontext.GetOperationID(ctx) operationID := mcontext.GetOperationID(ctx)
log.NewInfo("msg come here mysql!!!", "", "msg", string(msg), msgKey)
var tag bool var tag bool
msgFromMQ := pbMsg.MsgDataToMQ{} msgFromMQ := pbMsg.MsgDataToMQ{}
err := proto.Unmarshal(msg, &msgFromMQ) err := proto.Unmarshal(msg, &msgFromMQ)
if err != nil { if err != nil {
log.NewError(operationID, "msg_transfer Unmarshal msg err", "msg", string(msg), "err", err.Error()) log.ZError(ctx, "msg_transfer Unmarshal msg err", err)
return return
} }
log.Debug(operationID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String()) log.ZDebug(ctx, "handleChatWs2Mysql", "msg", msgFromMQ.MsgData)
//Control whether to store history messages (mysql) //Control whether to store history messages (mysql)
isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent) isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
//Only process receiver data //Only process receiver data