msg_gateway and msg_transfer modules

src/msg_transfer/Makefile

@@ -0,0 +1,26 @@
+.PHONY: all build run gotool install clean help
+
+BINARY_NAME=open_im_msg_transfer
+BIN_DIR=../../bin/
+LAN_FILE=.go
+GO_FILE:=${BINARY_NAME}${LAN_FILE}
+
+all: gotool build
+
+build:
+	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags="-w -s" -o  ${BINARY_NAME}  ${GO_FILE}
+
+run:
+	@go run ./
+
+gotool:
+	go fmt ./
+	go vet ./
+
+install:
+	make build
+	mv ${BINARY_NAME} ${BIN_DIR}
+
+clean:
+	@if [ -f ${BINARY_NAME} ] ; then rm ${BINARY_NAME} ; fi
+

src/msg_transfer/logic/db.go

@@ -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)
+}
+
+func getGroupList(groupID string) ([]string, error) {
+	return im_mysql_model.SelectGroupList(groupID)
+}

src/msg_transfer/logic/history_msg_handler.go

@@ -0,0 +1,178 @@
+package logic
+
+import (
+	"Open_IM/src/common/config"
+	"Open_IM/src/common/constant"
+	kfk "Open_IM/src/common/kafka"
+	"Open_IM/src/common/log"
+	pbMsg "Open_IM/src/proto/chat"
+	pb "Open_IM/src/proto/group"
+	pbPush "Open_IM/src/proto/push"
+	"Open_IM/src/utils"
+	"context"
+	"github.com/Shopify/sarama"
+	"github.com/golang/protobuf/proto"
+	"github.com/skiffer-git/grpc-etcdv3/getcdv3"
+	"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.SendTime = pbData.SendTime
+	pbSaveData.Content = pbData.Content
+	pbSaveData.MsgFrom = pbData.MsgFrom
+	pbSaveData.ContentType = pbData.ContentType
+	pbSaveData.SessionType = pbData.SessionType
+	pbSaveData.MsgID = pbData.MsgID
+	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")
+	//Control whether to push message to sender's other terminal
+	isSenderSync := utils.GetSwitchFromOptions(Options, "senderSync")
+	if pbData.SessionType == constant.SingleChatType {
+		log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist, isSenderSync)
+		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())
+				}
+				pbSaveData.Options = pbData.Options
+				pbSaveData.OfflineInfo = pbData.OfflineInfo
+				sendMessageToPush(&pbSaveData)
+			} 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)
+				}
+			}
+
+		}
+	} else if pbData.SessionType == constant.GroupChatType {
+		log.Info("", "", "msg_transfer chat type = GroupChatType")
+
+		etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
+		client := pb.NewGroupClient(etcdConn)
+		req := &pb.GetGroupInfoReq{
+			GroupID:     pbSaveData.RecvID,
+			Token:       pbData.Token,
+			OperationID: pbSaveData.OperationID,
+		}
+		log.Info("", "", "msg_transfer call group rpc, data = %s", req.String())
+		reply, err := client.GetGroupInfo(context.Background(), req)
+		if err != nil {
+			log.Error("", "", "msg_transfer client.GetGroupInfo fail, err = %s", err.Error())
+			return
+		}
+		for _, v := range reply.GroupMemberList {
+			//Store RecvID is userID+" "+groupID when chatType is Group
+			pbSaveData.RecvID = v.UserID + " " + pbSaveData.RecvID
+			if isHistory {
+				saveUserChat(v.UserID, &pbSaveData)
+			}
+			pbSaveData.Options = pbData.Options
+			pbSaveData.OfflineInfo = pbData.OfflineInfo
+			if v.UserID != pbSaveData.SendID {
+				if utils.IsContain(v.UserID, pbData.ForceList) {
+					pbSaveData.IsEmphasize = true
+				}
+				sendMessageToPush(&pbSaveData)
+			} else {
+				if isSenderSync {
+					pbSaveData.ContentType = constant.SyncSenderMsg
+					sendMessageToPush(&pbSaveData)
+				}
+
+			}
+
+		}
+
+	} else {
+		log.Error("", "", "msg_transfer recv chat err, chat.MsgFrom = %d", pbData.SessionType)
+	}
+
+	log.InfoByKv("msg_transfer handle topic success...", "", "")
+}
+
+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))
+	}
+	return nil
+}
+func sendMessageToPush(message *pbMsg.MsgSvrToPushSvrChatMsg) {
+	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.IsEmphasize = message.IsEmphasize
+	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)
+	defer grpcConn.Close()
+	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())
+
+	}
+}

src/msg_transfer/logic/init.go

@@ -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)
+}

src/msg_transfer/logic/persistent_msg_handler.go

@@ -0,0 +1,63 @@
+/*
+** 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/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"
+)
+
+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 && msgKey == pbData.RecvID {
+		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))
+	}
+	return nil
+}

src/msg_transfer/open_im_msg_transfer.go

@@ -0,0 +1,14 @@
+package main
+
+import (
+	"Open_IM/src/msg_transfer/logic"
+	"sync"
+)
+
+func main() {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	logic.Init()
+	logic.Run()
+	wg.Wait()
+}