mongo operation

2026-05-03 08:35:59 +08:00 · 2022-12-12 19:22:50 +08:00
parent 63d75bb68b
commit 0f1208dba9
11 changed files with 404 additions and 227 deletions
@@ -22,7 +22,9 @@ var (
 	persistentCH          PersistentConsumerHandler
 	historyCH             OnlineHistoryRedisConsumerHandler
 	historyMongoCH        OnlineHistoryMongoConsumerHandler
+	modifyCH              ModifyMsgConsumerHandler
 	producer              *kafka.Producer
+	producerToModify      *kafka.Producer
 	producerToMongo       *kafka.Producer
 	cmdCh                 chan Cmd2Value
 	onlineTopicStatus     int
@@ -43,11 +45,13 @@ func Init() {
 	persistentCH.Init()   // ws2mschat save mysql
 	historyCH.Init(cmdCh) //
 	historyMongoCH.Init()
+	modifyCH.Init()
 	onlineTopicStatus = OnlineTopicVacancy
 	//offlineHistoryCH.Init(cmdCh)
 	statistics.NewStatistics(&singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
 	statistics.NewStatistics(&groupMsgCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d second groupMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
 	producer = kafka.NewKafkaProducer(config.Config.Kafka.Ms2pschat.Addr, config.Config.Kafka.Ms2pschat.Topic)
+	producerToModify = kafka.NewKafkaProducer(config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.MsgToModify.Topic)
 	producerToMongo = kafka.NewKafkaProducer(config.Config.Kafka.MsgToMongo.Addr, config.Config.Kafka.MsgToMongo.Topic)
 }
 func Run(promethuesPort int) {
@@ -59,6 +63,7 @@ func Run(promethuesPort int) {
 	}
 	go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH)
 	go historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyMongoCH)
+	go modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(&modifyCH)
 	//go offlineHistoryCH.historyConsumerGroup.RegisterHandleAndConsumer(&offlineHistoryCH)
 	go func() {
 		err := promePkg.StartPromeSrv(promethuesPort)
@@ -0,0 +1,117 @@
+package logic
+
+import (
+	"Open_IM/pkg/base_info"
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/db"
+	kfk "Open_IM/pkg/common/kafka"
+	"Open_IM/pkg/common/log"
+	pbMsg "Open_IM/pkg/proto/msg"
+	server_api_params "Open_IM/pkg/proto/sdk_ws"
+	"Open_IM/pkg/utils"
+	"encoding/json"
+	"github.com/Shopify/sarama"
+
+	"github.com/golang/protobuf/proto"
+)
+
+type ModifyMsgConsumerHandler struct {
+	msgHandle              map[string]fcb
+	modifyMsgConsumerGroup *kfk.MConsumerGroup
+}
+
+func (mmc *ModifyMsgConsumerHandler) Init() {
+	mmc.msgHandle = make(map[string]fcb)
+	mmc.msgHandle[config.Config.Kafka.MsgToModify.Topic] = mmc.ModifyMsg
+	mmc.modifyMsgConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
+		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.MsgToModify.Topic},
+		config.Config.Kafka.MsgToModify.Addr, config.Config.Kafka.ConsumerGroupID.MsgToModify)
+}
+
+func (ModifyMsgConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (ModifyMsgConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (mmc *ModifyMsgConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
+	claim sarama.ConsumerGroupClaim) error {
+	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))
+		if len(msg.Value) != 0 {
+			mmc.msgHandle[msg.Topic](msg, string(msg.Key), sess)
+		} else {
+			log.Error("", "msg get from kafka but is nil", msg.Key)
+		}
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}
+
+func (mmc *ModifyMsgConsumerHandler) ModifyMsg(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
+	log.NewInfo("msg come here ModifyMsg!!!", "", "msg", string(cMsg.Value), msgKey)
+	msgFromMQ := pbMsg.MsgDataToModifyByMQ{}
+	err := proto.Unmarshal(cMsg.Value, &msgFromMQ)
+	if err != nil {
+		log.NewError(msgFromMQ.TriggerID, "msg_transfer Unmarshal msg err", "msg", string(cMsg.Value), "err", err.Error())
+		return
+	}
+	log.Debug(msgFromMQ.TriggerID, "proto.Unmarshal MsgDataToMQ", msgFromMQ.String())
+	for _, msgDataToMQ := range msgFromMQ.MessageList {
+		if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageModifier {
+			notification := &base_info.ReactionMessageModifierNotification{}
+			if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
+				continue
+			}
+			if notification.IsExternalExtensions {
+				log.NewInfo(msgDataToMQ.OperationID, "msg:", notification, "this is external extensions")
+				continue
+			}
+			if !notification.IsReact {
+				// first time to modify
+				var reactionExtensionList = make(map[string]db.KeyValue)
+				extendMsg := db.ExtendMsg{
+					ReactionExtensionList: reactionExtensionList,
+					ClientMsgID:           notification.ClientMsgID,
+					MsgFirstModifyTime:    notification.MsgFirstModifyTime,
+				}
+				for _, v := range notification.SuccessReactionExtensionList {
+					reactionExtensionList[v.TypeKey] = db.KeyValue{
+						TypeKey:          v.TypeKey,
+						Value:            v.Value,
+						LatestUpdateTime: v.LatestUpdateTime,
+					}
+				}
+
+				if err := db.DB.InsertExtendMsg(notification.SourceID, notification.SessionType, &extendMsg); err != nil {
+					log.NewError(msgDataToMQ.OperationID, "MsgFirstModify InsertExtendMsg failed", notification.SourceID, notification.SessionType, extendMsg, err.Error())
+					continue
+				}
+			} else {
+				var reactionExtensionList = make(map[string]*server_api_params.KeyValue)
+				for _, v := range notification.SuccessReactionExtensionList {
+					reactionExtensionList[v.TypeKey] = &server_api_params.KeyValue{
+						TypeKey:          v.TypeKey,
+						Value:            v.Value,
+						LatestUpdateTime: v.LatestUpdateTime,
+					}
+				}
+				// is already modify
+				if err := db.DB.InsertOrUpdateReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, reactionExtensionList); err != nil {
+					log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
+				}
+			}
+		} else if msgDataToMQ.MsgData.ContentType == constant.ReactionMessageDeleter {
+			notification := &base_info.ReactionMessageDeleteNotification{}
+			if err := json.Unmarshal(msgDataToMQ.MsgData.Content, notification); err != nil {
+				continue
+			}
+			if err := db.DB.DeleteReactionExtendMsgSet(notification.SourceID, notification.SessionType, notification.ClientMsgID, notification.MsgFirstModifyTime, notification.SuccessReactionExtensionList); err != nil {
+				log.NewError(msgDataToMQ.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
+			}
+		}
+	}
+
+}
+
+func UnMarshallSetReactionMsgContent(content []byte) (notification *base_info.ReactionMessageModifierNotification, err error) {
+
+	return notification, nil
+}
@@ -72,6 +72,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 				storageMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
 				notStoragePushMsgList := make([]*pbMsg.MsgDataToMQ, 0, 80)
 				log.Debug(triggerID, "msg arrived channel", "channel id", channelID, msgList, msgChannelValue.aggregationID, len(msgList))
+				var modifyMsgList []*pbMsg.MsgDataToMQ
 				for _, v := range msgList {
 					log.Debug(triggerID, "msg come to storage center", v.String())
 					isHistory := utils.GetSwitchFromOptions(v.MsgData.Options, constant.IsHistory)
@@ -85,8 +86,13 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 						}
 					}

+					if v.MsgData.ContentType == constant.ReactionMessageModifier || v.MsgData.ContentType == constant.ReactionMessageDeleter {
+						modifyMsgList = append(modifyMsgList, v)
+					}
+				}
+				if len(modifyMsgList) > 0 {
+					sendMessageToModifyMQ(msgChannelValue.aggregationID, triggerID, modifyMsgList)
 				}
-
 				//switch msgChannelValue.msg.MsgData.SessionType {
 				//case constant.SingleChatType:
 				//case constant.GroupChatType:
@@ -552,6 +558,17 @@ func sendMessageToPushMQ(message *pbMsg.MsgDataToMQ, pushToUserID string) {
 	return
 }

+func sendMessageToModifyMQ(aggregationID string, triggerID string, messages []*pbMsg.MsgDataToMQ) {
+	if len(messages) > 0 {
+		pid, offset, err := producerToModify.SendMessage(&pbMsg.MsgDataToModifyByMQ{AggregationID: aggregationID, MessageList: messages, TriggerID: triggerID}, aggregationID, triggerID)
+		if err != nil {
+			log.Error(triggerID, "kafka send failed", "send data", len(messages), "pid", pid, "offset", offset, "err", err.Error(), "key", aggregationID)
+		} else {
+			//	log.NewWarn(m.OperationID, "sendMsgToKafka   client msgID ", m.MsgData.ClientMsgID)
+		}
+	}
+}
+
 // String hashes a string to a unique hashcode.
 //
 // crc32 returns a uint32, but for our use we need
@@ -61,50 +61,6 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(cMsg *sarama.Con
 			if unexistSeqList, err := db.DB.DelMsgBySeqList(DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID); err != nil {
 				log.NewError(v.OperationID, utils.GetSelfFuncName(), "DelMsgBySeqList args: ", DeleteMessageTips.UserID, DeleteMessageTips.SeqList, v.OperationID, err.Error(), unexistSeqList)
 			}
-		} else if v.MsgData.ContentType == constant.ReactionMessageModifier {
-			var req pbMsg.ModifyMessageReactionExtensionsReq
-			if req.IsExternalExtensions {
-				log.NewInfo(req.OperationID, "msg:", req.String(), "this is external extensions")
-				continue
-			}
-			if !req.IsReact {
-				// first time to modify
-				var reactionExtensionList = make(map[string]db.KeyValue)
-				for k, v := range req.ReactionExtensionList {
-					reactionExtensionList[k] = db.KeyValue{
-						TypeKey:          v.TypeKey,
-						Value:            v.Value,
-						LatestUpdateTime: v.LatestUpdateTime,
-					}
-				}
-				extendMsg := db.ExtendMsg{
-					ReactionExtensionList: reactionExtensionList,
-					ClientMsgID:           req.ClientMsgID,
-					MsgFirstModifyTime:    req.MsgFirstModifyTime,
-				}
-				if req.AttachedInfo != nil {
-					extendMsg.AttachedInfo = req.AttachedInfo.Value
-				}
-				if req.Ex != nil {
-					extendMsg.Ex = req.Ex.Value
-				}
-				if err := db.DB.InsertExtendMsg(req.SourceID, req.SessionType, &extendMsg); err != nil {
-					log.NewError(req.OperationID, "MsgFirstModify InsertExtendMsg failed", req.SourceID, req.SessionType, extendMsg, err.Error())
-					continue
-				}
-			} else {
-				// is already modify
-				if err := db.DB.InsertOrUpdateReactionExtendMsgSet(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime, req.ReactionExtensionList); err != nil {
-					log.NewError(req.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
-				}
-			}
-		} else if v.MsgData.ContentType == constant.ReactionMessageDeleter {
-			var req pbMsg.OperateMessageListReactionExtensionsReq
-			for _, v := range req.MessageReactionKeyList {
-				if err := db.DB.DeleteReactionExtendMsgSet(req.SourceID, req.SessionType, v.ClientMsgID, v.MsgFirstModifyTime, v.ReactionExtensionList); err != nil {
-					log.NewError(req.OperationID, "InsertOrUpdateReactionExtendMsgSet failed")
-				}
-			}
 		}
 	}
 }