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

msg update

Browse Source
This commit is contained in:
Gordon 2023-03-22 19:24:38 +08:00
parent 2b0e0eb9e5
commit 19b8bd0ff6
5 changed files with 7 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/msgtransfer/modify_msg_handler.go
View File

@ -41,7 +41,7 @@ func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
for msg := range claim.Messages() { for msg := range claim.Messages() {
log.NewDebug("", "kafka get info to mysql", "ModifyMsgConsumerHandler", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key)) log.NewDebug("", "kafka get info to mysql", "ModifyMsgConsumerHandler", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
if len(msg.Value) != 0 { if len(msg.Value) != 0 {
ctx := mmc.modifyMsgConsumerGroup.GetContextFromMsg(msg, "modify consumer") ctx := mmc.modifyMsgConsumerGroup.GetContextFromMsg(msg)
mmc.ModifyMsg(ctx, msg, string(msg.Key), sess) mmc.ModifyMsg(ctx, msg, string(msg.Key), sess)
} else { } else {
log.Error("", "msg get from kafka but is nil", msg.Key) log.Error("", "msg get from kafka but is nil", msg.Key)

2
internal/msgtransfer/online_msg_to_mongo_handler.go
View File

@ -78,7 +78,7 @@ func (mc *OnlineHistoryMongoConsumerHandler) ConsumeClaim(sess sarama.ConsumerGr
for msg := range claim.Messages() { for msg := range claim.Messages() {
log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key)) log.NewDebug("", "kafka get info to mongo", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
if len(msg.Value) != 0 { if len(msg.Value) != 0 {
ctx := mc.historyConsumerGroup.GetContextFromMsg(msg, "mongoDB consumer") ctx := mc.historyConsumerGroup.GetContextFromMsg(msg)
mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess) mc.handleChatWs2Mongo(ctx, msg, string(msg.Key), sess)
} else { } else {
log.Error("", "mongo msg get from kafka but is nil", msg.Key) log.Error("", "mongo msg get from kafka but is nil", msg.Key)

2
internal/msgtransfer/persistent_msg_handler.go
View File

@ -78,7 +78,7 @@ func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
for msg := range claim.Messages() { for msg := range claim.Messages() {
log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key)) log.NewDebug("", "kafka get info to mysql", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value), "key", string(msg.Key))
if len(msg.Value) != 0 { if len(msg.Value) != 0 {
ctx := pc.persistentConsumerGroup.GetContextFromMsg(msg, "mysql consumer") ctx := pc.persistentConsumerGroup.GetContextFromMsg(msg)
pc.handleChatWs2Mysql(ctx, msg, string(msg.Key), sess) pc.handleChatWs2Mysql(ctx, msg, string(msg.Key), sess)
} else { } else {
log.Error("", "msg get from kafka but is nil", msg.Key) log.Error("", "msg get from kafka but is nil", msg.Key)

2
internal/push/push_handler.go
View File

@ -58,7 +58,7 @@ func (ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil
func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
claim sarama.ConsumerGroupClaim) error { claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() { for msg := range claim.Messages() {
ctx := c.pushConsumerGroup.GetContextFromMsg(msg, "push consumer") ctx := c.pushConsumerGroup.GetContextFromMsg(msg)
c.handleMs2PsChat(ctx, msg.Value) c.handleMs2PsChat(ctx, msg.Value)
sess.MarkMessage(msg, "") sess.MarkMessage(msg, "")
} }

15
pkg/common/kafka/consumer_group.go
View File

@ -8,8 +8,6 @@ package kafka
import ( import (
"context" "context"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
) )
@ -42,16 +40,9 @@ func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []str
} }
} }
func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage, rootFuncName string) context.Context { func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context {
ctx := mcontext.NewCtx(rootFuncName) return GetContextWithMQHeader(cMsg.Headers)
var operationID string
for _, v := range cMsg.Headers {
if string(v.Key) == constant.OperationID {
operationID = string(v.Value)
}
}
mcontext.SetOperationID(ctx, operationID)
return ctx
} }
func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler) { func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler) {