add cmd/open_im_msg_transfer

2026-05-11 20:45:57 +08:00 · 2021-10-11 21:40:31 +08:00
parent 5fb4ea2aa3
commit b4195bd81c
6 changed files with 1 additions and 1 deletions
@@ -0,0 +1,20 @@
+package logic
+
+import (
+	"Open_IM/src/common/db"
+	"Open_IM/src/common/db/mysql_model/im_mysql_model"
+	pbMsg "Open_IM/src/proto/chat"
+)
+
+func saveUserChat(uid string, pbMsg *pbMsg.MsgSvrToPushSvrChatMsg) error {
+	seq, err := db.DB.IncrUserSeq(uid)
+	if err != nil {
+		return err
+	}
+	pbMsg.RecvSeq = seq
+	return db.DB.SaveUserChat(uid, pbMsg.SendTime, pbMsg)
+}
+
+func getGroupList(groupID string) ([]string, error) {
+	return im_mysql_model.SelectGroupList(groupID)
+}
@@ -0,0 +1,156 @@
+package logic
+
+import (
+	"Open_IM/src/common/config"
+	"Open_IM/src/common/constant"
+	kfk "Open_IM/src/common/kafka"
+	"Open_IM/src/common/log"
+	"Open_IM/src/grpc-etcdv3/getcdv3"
+	pbMsg "Open_IM/src/proto/chat"
+	pbPush "Open_IM/src/proto/push"
+	"Open_IM/src/utils"
+	"context"
+	"github.com/Shopify/sarama"
+	"github.com/golang/protobuf/proto"
+	"strings"
+)
+
+type fcb func(msg []byte, msgKey string)
+
+type HistoryConsumerHandler struct {
+	msgHandle            map[string]fcb
+	historyConsumerGroup *kfk.MConsumerGroup
+}
+
+func (mc *HistoryConsumerHandler) Init() {
+	mc.msgHandle = make(map[string]fcb)
+	mc.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = mc.handleChatWs2Mongo
+	mc.historyConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V0_10_2_0,
+		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
+		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMongo)
+
+}
+
+func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string) {
+	log.InfoByKv("chat come mongo!!!", "", "chat", string(msg))
+	pbData := pbMsg.WSToMsgSvrChatMsg{}
+	err := proto.Unmarshal(msg, &pbData)
+	if err != nil {
+		log.ErrorByKv("msg_transfer Unmarshal chat err", "", "chat", string(msg), "err", err.Error())
+		return
+	}
+	pbSaveData := pbMsg.MsgSvrToPushSvrChatMsg{}
+	pbSaveData.SendID = pbData.SendID
+	pbSaveData.SenderNickName = pbData.SenderNickName
+	pbSaveData.SenderFaceURL = pbData.SenderFaceURL
+	pbSaveData.ClientMsgID = pbData.ClientMsgID
+	pbSaveData.SendTime = pbData.SendTime
+	pbSaveData.Content = pbData.Content
+	pbSaveData.MsgFrom = pbData.MsgFrom
+	pbSaveData.ContentType = pbData.ContentType
+	pbSaveData.SessionType = pbData.SessionType
+	pbSaveData.MsgID = pbData.MsgID
+	pbSaveData.OperationID = pbData.OperationID
+	pbSaveData.RecvID = pbData.RecvID
+	pbSaveData.PlatformID = pbData.PlatformID
+	Options := utils.JsonStringToMap(pbData.Options)
+	//Control whether to store offline messages (mongo)
+	isHistory := utils.GetSwitchFromOptions(Options, "history")
+	//Control whether to store history messages (mysql)
+	isPersist := utils.GetSwitchFromOptions(Options, "persistent")
+	if pbData.SessionType == constant.SingleChatType {
+		log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist)
+		if isHistory {
+			if msgKey == pbSaveData.RecvID {
+				err := saveUserChat(pbData.RecvID, &pbSaveData)
+				if err != nil {
+					log.ErrorByKv("data insert to mongo err", pbSaveData.OperationID, "data", pbSaveData.String(), "err", err.Error())
+				}
+			} else if msgKey == pbSaveData.SendID {
+				err := saveUserChat(pbData.SendID, &pbSaveData)
+				if err != nil {
+					log.ErrorByKv("data insert to mongo err", pbSaveData.OperationID, "data", pbSaveData.String(), "err", err.Error())
+				}
+				//if isSenderSync {
+				//	pbSaveData.ContentType = constant.SyncSenderMsg
+				//	log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String())
+				//	sendMessageToPush(&pbSaveData)
+				//}
+			}
+
+		}
+		if msgKey == pbSaveData.RecvID {
+			pbSaveData.Options = pbData.Options
+			pbSaveData.OfflineInfo = pbData.OfflineInfo
+			sendMessageToPush(&pbSaveData)
+		}
+
+		log.InfoByKv("msg_transfer handle topic success...", "", "")
+	} else if pbData.SessionType == constant.GroupChatType {
+		log.Info("", "", "msg_transfer chat type = GroupChatType")
+		if isHistory {
+			uidAndGroupID := strings.Split(pbData.RecvID, " ")
+			saveUserChat(uidAndGroupID[0], &pbSaveData)
+		}
+		pbSaveData.Options = pbData.Options
+		pbSaveData.OfflineInfo = pbData.OfflineInfo
+		sendMessageToPush(&pbSaveData)
+		log.InfoByKv("msg_transfer handle topic success...", "", "")
+	} else {
+		log.Error("", "", "msg_transfer recv chat err, chat.MsgFrom = %d", pbData.SessionType)
+	}
+
+}
+
+func (HistoryConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (HistoryConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (mc *HistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
+	claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		log.InfoByKv("kafka get info to mongo", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "chat", string(msg.Value))
+		mc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}
+func sendMessageToPush(message *pbMsg.MsgSvrToPushSvrChatMsg) {
+	log.InfoByKv("msg_transfer send message to push", message.OperationID, "message", message.String())
+	msg := pbPush.PushMsgReq{}
+	msg.OperationID = message.OperationID
+	msg.PlatformID = message.PlatformID
+	msg.Content = message.Content
+	msg.ContentType = message.ContentType
+	msg.SessionType = message.SessionType
+	msg.RecvID = message.RecvID
+	msg.SendID = message.SendID
+	msg.SenderNickName = message.SenderNickName
+	msg.SenderFaceURL = message.SenderFaceURL
+	msg.ClientMsgID = message.ClientMsgID
+	msg.MsgFrom = message.MsgFrom
+	msg.Options = message.Options
+	msg.RecvSeq = message.RecvSeq
+	msg.SendTime = message.SendTime
+	msg.MsgID = message.MsgID
+	msg.OfflineInfo = message.OfflineInfo
+	grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImPushName)
+	if grpcConn == nil {
+		log.ErrorByKv("rpc dial failed", msg.OperationID, "push data", msg.String())
+		pid, offset, err := producer.SendMessage(message)
+		if err != nil {
+			log.ErrorByKv("kafka send failed", msg.OperationID, "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
+		}
+		return
+	}
+	msgClient := pbPush.NewPushMsgServiceClient(grpcConn)
+	_, err := msgClient.PushMsg(context.Background(), &msg)
+	if err != nil {
+		log.ErrorByKv("rpc send failed", msg.OperationID, "push data", msg.String(), "err", err.Error())
+		pid, offset, err := producer.SendMessage(message)
+		if err != nil {
+			log.ErrorByKv("kafka send failed", msg.OperationID, "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
+		}
+	} else {
+		log.InfoByKv("rpc send success", msg.OperationID, "push data", msg.String())
+
+	}
+}
@@ -0,0 +1,25 @@
+package logic
+
+import (
+	"Open_IM/src/common/config"
+	"Open_IM/src/common/kafka"
+	"Open_IM/src/common/log"
+)
+
+var (
+	persistentCH PersistentConsumerHandler
+	historyCH    HistoryConsumerHandler
+	producer     *kafka.Producer
+)
+
+func Init() {
+	log.NewPrivateLog(config.Config.ModuleName.MsgTransferName)
+	persistentCH.Init()
+	historyCH.Init()
+	producer = kafka.NewKafkaProducer(config.Config.Kafka.Ms2pschat.Addr, config.Config.Kafka.Ms2pschat.Topic)
+}
+func Run() {
+	//register mysqlConsumerHandler to
+	go persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(&persistentCH)
+	go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH)
+}
@@ -0,0 +1,75 @@
+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/5/11 15:37).
+ */
+package logic
+
+import (
+	"Open_IM/src/common/config"
+	"Open_IM/src/common/constant"
+	"Open_IM/src/common/db/mysql_model/im_mysql_msg_model"
+	kfk "Open_IM/src/common/kafka"
+	"Open_IM/src/common/log"
+	pbMsg "Open_IM/src/proto/chat"
+	"Open_IM/src/utils"
+	"github.com/Shopify/sarama"
+	"github.com/golang/protobuf/proto"
+	"strings"
+)
+
+type PersistentConsumerHandler struct {
+	msgHandle               map[string]fcb
+	persistentConsumerGroup *kfk.MConsumerGroup
+}
+
+func (pc *PersistentConsumerHandler) Init() {
+	pc.msgHandle = make(map[string]fcb)
+	pc.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = pc.handleChatWs2Mysql
+	pc.persistentConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V0_10_2_0,
+		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
+		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql)
+
+}
+func (pc *PersistentConsumerHandler) handleChatWs2Mysql(msg []byte, msgKey string) {
+	log.InfoByKv("chat come here mysql!!!", "", "chat", string(msg))
+	pbData := pbMsg.WSToMsgSvrChatMsg{}
+	err := proto.Unmarshal(msg, &pbData)
+	if err != nil {
+		log.ErrorByKv("msg_transfer Unmarshal chat err", "", "chat", string(msg), "err", err.Error())
+		return
+	}
+	Options := utils.JsonStringToMap(pbData.Options)
+	//Control whether to store history messages (mysql)
+	isPersist := utils.GetSwitchFromOptions(Options, "persistent")
+	//Only process receiver data
+	if isPersist {
+		if msgKey == pbData.RecvID && pbData.SessionType == constant.SingleChatType {
+			log.InfoByKv("msg_transfer chat persisting", pbData.OperationID)
+			if err = im_mysql_msg_model.InsertMessageToChatLog(pbData); err != nil {
+				log.ErrorByKv("Message insert failed", pbData.OperationID, "err", err.Error(), "chat", pbData.String())
+				return
+			}
+		} else if pbData.SessionType == constant.GroupChatType && msgKey == "0" {
+			pbData.RecvID = strings.Split(pbData.RecvID, " ")[1]
+			log.InfoByKv("msg_transfer chat persisting", pbData.OperationID)
+			if err = im_mysql_msg_model.InsertMessageToChatLog(pbData); err != nil {
+				log.ErrorByKv("Message insert failed", pbData.OperationID, "err", err.Error(), "chat", pbData.String())
+				return
+			}
+		}
+
+	}
+}
+func (PersistentConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
+func (PersistentConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
+func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
+	claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		log.InfoByKv("kafka get info to mysql", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "chat", string(msg.Value))
+		pc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
+		sess.MarkMessage(msg, "")
+	}
+	return nil
+}