Merge remote-tracking branch 'origin/main'

README.md

@@ -2,7 +2,7 @@
 Open-IM-Server is open source instant messaging Server.Backend in Go.
 ![avatar](https://github.com/OpenIMSDK/Open-IM-Server/blob/main/docs/Open-IM.png)
 
-[![LICENSE](https://img.shields.io/github/license/pingcap/tidb.svg)](https://github.com/OpenIMSDK/Open-IM-Server/blob/main/LICENSE)
+[![LICENSE](https://img.shields.io/badge/license-Apache--2.0-green)](https://github.com/OpenIMSDK/Open-IM-Server/blob/main/LICENSE)
 [![Language](https://img.shields.io/badge/Language-Go-blue.svg)](https://golang.org/)
 
 ## Open-IM-Server: Open source  Instant Messaging Server

src/common/constant/constant.go

+package constant
+
+const (
+
+	//group admin
+	GroupAdmin = 1
+	//feiend related
+	BlackListFlag = 1
+	NotFriendFlag = 0
+	FriendFlag    = 1
+
+	//Websocket Protocol
+	WSGetNewestSeq = 1001
+	WSPullMsg      = 1002
+	WSSendMsg      = 1003
+	WSPushMsg      = 2001
+
+	///ContentType
+	//UserRelated
+	Text    = 101
+	Picture = 102
+	Voice   = 103
+	Video   = 104
+	File    = 105
+
+	SyncSenderMsg = 108
+	//SysRelated
+	AddFriendTip      = 201
+	AgreeAddFriendTip = 202
+	KickOnlineTip     = 203
+
+	//MsgFrom
+	UserMsgType = 100
+	SysMsgType  = 200
+
+	//SessionType
+	SingleChatType = 1
+	GroupChatType  = 2
+)
+
+var ContentType2PushContent = map[int64]string{
+	Picture: "[picture]",
+	Voice:   "[voice]",
+	Video:   "[video]",
+	File:    "[file]",
+}

src/common/db/mysql_model/im_mysql_msg_model/chat_log_model.go

+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/3/4 11:18).
+ */
+package im_mysql_msg_model
+
+import (
+	"Open_IM/src/common/db"
+	pbMsg "Open_IM/src/proto/chat"
+	"Open_IM/src/utils"
+	"database/sql"
+	"time"
+)
+
+// ChatLog Chat information table structure
+type ChatLog struct {
+	MsgId            string         `gorm:"primary_key"`               // Chat history primary key ID
+	SendID           string         `gorm:"column:send_id"`            // Send ID
+	RecvID           string         `gorm:"column:recv_id"`            //Receive ID
+	SendTime         time.Time      `gorm:"column:send_time"`          // Send time
+	SessionType      int32          `gorm:"column:session_type"`       // Session type
+	ContentType      int32          `gorm:"column:content_type"`       // Message content type
+	MsgFrom          int32          `gorm:"column:msg_from"`           // Source, user, system
+	Content          string         `gorm:"column:content"`            // Chat content
+	SenderPlatformID int32          `gorm:"column:sender_platform_id"` //The sender's platform ID
+	Remark           sql.NullString `gorm:"column:remark"`             // remark
+}
+
+func InsertMessageToChatLog(msgData pbMsg.WSToMsgSvrChatMsg) error {
+	dbConn, err := db.DB.MysqlDB.DefaultGormDB()
+	if err != nil {
+		return err
+	}
+	chatLog := ChatLog{
+		MsgId:            msgData.MsgID,
+		SendID:           msgData.SendID,
+		RecvID:           msgData.RecvID,
+		SendTime:         utils.UnixSecondToTime(msgData.SendTime),
+		SessionType:      msgData.SessionType,
+		ContentType:      msgData.ContentType,
+		MsgFrom:          msgData.MsgFrom,
+		Content:          msgData.Content,
+		SenderPlatformID: msgData.PlatformID,
+	}
+	return dbConn.Table("chat_log").Create(chatLog).Error
+}

src/common/db/mysql_model/im_mysql_msg_model/hash_code.go

+package im_mysql_msg_model
+
+import (
+	"Open_IM/src/common/config"
+	"Open_IM/src/common/db"
+	"hash/crc32"
+	"strconv"
+)
+
+func getHashMsgDBAddr(userID string) string {
+	hCode := crc32.ChecksumIEEE([]byte(userID))
+	return config.Config.Mysql.DBAddress[hCode%uint32(len(config.Config.Mysql.DBAddress))]
+}
+
+func getHashMsgTableIndex(userID string) int {
+	hCode := crc32.ChecksumIEEE([]byte(userID))
+	return int(hCode % uint32(config.Config.Mysql.DBMsgTableNum))
+}
+
+func QueryUserMsgID(userID string) ([]string, error) {
+	dbAddress, dbTableIndex := getHashMsgDBAddr(userID), getHashMsgTableIndex(userID)
+	dbTableName := "receive" + strconv.Itoa(dbTableIndex)
+
+	dbConn, _ := db.DB.MysqlDB.GormDB(dbAddress, config.Config.Mysql.DBTableName)
+
+	var msgID string
+	var msgIDList []string
+	rows, _ := dbConn.Raw("select msg_id from ? where user_id = ?", dbTableName, userID).Rows()
+	defer rows.Close()
+	for rows.Next() {
+		rows.Scan(&msgID)
+		msgIDList = append(msgIDList, msgID)
+	}
+
+	return msgIDList, nil
+}

src/common/db/mysql_model/im_mysql_msg_model/receive_model.go

+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/3/4 11:18).
+ */
+package im_mysql_msg_model
+
+import (
+	"time"
+)
+
+// Receive Inbox table structure
+type Receive struct {
+	UserId     string `gorm:"primary_key"` // 收件箱主键ID
+	Seq        int64  `gorm:"primary_key"` // 收件箱主键ID
+	MsgId      string
+	CreateTime *time.Time
+}
+
+//func InsertMessageToReceive(seq int64, userid, msgid string) error {
+//	conn := db.NewDbConnection()
+//	receive := Receive{
+//		UID: userid,
+//		Seq:    seq,
+//		MsgId:  msgid,
+//	}
+//	err := conn.Table("receive").Create(&receive).Error
+//	return err
+//}
+//func GetBiggestSeqFromReceive(userid string) (seq int64, err error) {
+//	//得到数据库的连接（并非真连接，调用时才连接，由gorm自动维护数据库连接池）
+//	conn := db.NewDbConnection()
+//	err = conn.Raw("select max(seq) from receive where user_id = ?", userid).Row().Scan(&seq)
+//	return seq, err
+//}

src/common/kafka/consumer.go

+package kafka
+
+import (
+	"github.com/Shopify/sarama"
+	"sync"
+)
+
+type Consumer struct {
+	addr          []string
+	WG            sync.WaitGroup
+	Topic         string
+	PartitionList []int32
+	Consumer      sarama.Consumer
+}
+
+func NewKafkaConsumer(addr []string, topic string) *Consumer {
+	p := Consumer{}
+	p.Topic = topic
+	p.addr = addr
+
+	consumer, err := sarama.NewConsumer(p.addr, nil)
+	if err != nil {
+		panic(err)
+		return nil
+	}
+	p.Consumer = consumer
+
+	partitionList, err := consumer.Partitions(p.Topic)
+	if err != nil {
+		panic(err)
+		return nil
+	}
+	p.PartitionList = partitionList
+
+	return &p
+}

src/common/kafka/consumer_group.go

+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/5/11 9:36).
+ */
+package kafka
+
+import (
+	"context"
+	"github.com/Shopify/sarama"
+)
+
+type MConsumerGroup struct {
+	sarama.ConsumerGroup
+	groupID string
+	topics  []string
+}
+
+type MConsumerGroupConfig struct {
+	KafkaVersion   sarama.KafkaVersion
+	OffsetsInitial int64
+	IsReturnErr    bool
+}
+
+func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addr []string, groupID string) *MConsumerGroup {
+	config := sarama.NewConfig()
+	config.Version = consumerConfig.KafkaVersion
+	config.Consumer.Offsets.Initial = consumerConfig.OffsetsInitial
+	config.Consumer.Return.Errors = consumerConfig.IsReturnErr
+	client, err := sarama.NewClient(addr, config)
+	if err != nil {
+		panic(err)
+	}
+	consumerGroup, err := sarama.NewConsumerGroupFromClient(groupID, client)
+	if err != nil {
+		panic(err)
+	}
+	return &MConsumerGroup{
+		consumerGroup,
+		groupID,
+		topics,
+	}
+}
+func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler) {
+	ctx := context.Background()
+	for {
+		err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler)
+		if err != nil {
+			panic(err)
+		}
+	}
+}

src/common/kafka/producer.go

+package kafka
+
+import (
+	log2 "Open_IM/src/common/log"
+	"github.com/Shopify/sarama"
+	"github.com/golang/protobuf/proto"
+)
+
+type Producer struct {
+	topic    string
+	addr     []string
+	config   *sarama.Config
+	producer sarama.SyncProducer
+}
+
+func NewKafkaProducer(addr []string, topic string) *Producer {
+	p := Producer{}
+	p.config = sarama.NewConfig()                             //Instantiate a sarama Config
+	p.config.Producer.Return.Successes = true                 //Whether to enable the successes channel to be notified after the message is sent successfully
+	p.config.Producer.RequiredAcks = sarama.WaitForAll        //Set producer Message Reply level 0 1 all
+	p.config.Producer.Partitioner = sarama.NewHashPartitioner //Set the hash-key automatic hash partition. When sending a message, you must specify the key value of the message. If there is no key, the partition will be selected randomly
+
+	p.addr = addr
+	p.topic = topic
+
+	producer, err := sarama.NewSyncProducer(p.addr, p.config) //初始化客户端
+	if err != nil {
+		panic(err)
+		return nil
+	}
+	p.producer = producer
+	return &p
+}
+
+func (p *Producer) SendMessage(m proto.Message, key ...string) (int32, int64, error) {
+	kMsg := &sarama.ProducerMessage{}
+	kMsg.Topic = p.topic
+	if len(key) == 1 {
+		kMsg.Key = sarama.StringEncoder(key[0])
+	}
+	bMsg, err := proto.Marshal(m)
+	if err != nil {
+		log2.Error("", "", "proto marshal err = %s", err.Error())
+		return -1, -1, err
+	}
+	kMsg.Value = sarama.ByteEncoder(bMsg)
+
+	return p.producer.SendMessage(kMsg)
+}

src/common/log/es_hk.go

+/*
+** description("Hook to send logs to elasticsearch").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/3/26 17:05).
+ */
+package log
+
+import (
+	"Open_IM/src/common/config"
+	"context"
+	"fmt"
+	elasticV7 "github.com/olivere/elastic/v7"
+	"github.com/sirupsen/logrus"
+	"log"
+	"os"
+	"strings"
+	"time"
+)
+
+//esHook custom es hook
+type esHook struct {
+	moduleName string
+	client     *elasticV7.Client
+}
+
+//newEsHook initialization
+func newEsHook(moduleName string) *esHook {
+	//https://github.com/sohlich/elogrus
+	//client, err := elastic.NewClient(elastic.SetURL("http://localhost:9200"))
+	//if err != nil {
+	//	log.Panic(err)
+	//}
+	//hook, err := elogrus.NewAsyncElasticHook(client, "localhost", logrus.DebugLevel, "mylog")
+	//if err != nil {
+	//	log.Panic(err)
+	//}
+	es, err := elasticV7.NewClient(
+		elasticV7.SetURL(config.Config.Log.ElasticSearchAddr...),
+		elasticV7.SetBasicAuth(config.Config.Log.ElasticSearchUser, config.Config.Log.ElasticSearchPassword),
+		elasticV7.SetSniff(false),
+		elasticV7.SetHealthcheckInterval(60*time.Second),
+		elasticV7.SetErrorLog(log.New(os.Stderr, "ES:", log.LstdFlags)),
+	)
+
+	if err != nil {
+		log.Fatal("failed to create Elastic V7 Client: ", err)
+	}
+
+	//info, code, err := es.Ping(logConfig.ElasticSearch.EsAddr[0]).Do(context.Background())
+	//if err != nil {
+	//	panic(err)
+	//}
+	//fmt.Printf("Elasticsearch returned with code %d and version %s\n", code, info.Version.Number)
+	//
+	//esversion, err := es.ElasticsearchVersion(logConfig.ElasticSearch.EsAddr[0])
+	//if err != nil {
+	//	panic(err)
+	//}
+	//fmt.Printf("Elasticsearch version %s\n", esversion)
+	return &esHook{client: es, moduleName: moduleName}
+}
+
+//Fire log hook interface method
+func (hook *esHook) Fire(entry *logrus.Entry) error {
+	doc := newEsLog(entry)
+	go hook.sendEs(doc)
+	return nil
+}
+
+//Levels log hook interface method, the log affected by this hook
+func (hook *esHook) Levels() []logrus.Level {
+	return logrus.AllLevels
+}
+
+//sendEs Asynchronously send logs to es
+func (hook *esHook) sendEs(doc appLogDocModel) {
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Println("send entry to es failed: ", r)
+		}
+	}()
+	_, err := hook.client.Index().Index(hook.moduleName).Type(doc.indexName()).BodyJson(doc).Do(context.Background())
+	if err != nil {
+		log.Println(err)
+	}
+
+}
+
+//appLogDocModel es model
+type appLogDocModel map[string]interface{}
+
+func newEsLog(e *logrus.Entry) appLogDocModel {
+	ins := make(map[string]interface{})
+	ins["level"] = strings.ToUpper(e.Level.String())
+	ins["time"] = e.Time.Format("2006-01-02 15:04:05")
+	for kk, vv := range e.Data {
+		ins[kk] = vv
+	}
+	ins["tipInfo"] = e.Message
+
+	return ins
+}
+
+// indexName es index name time division
+func (m *appLogDocModel) indexName() string {
+	return time.Now().Format("2006-01-02")
+}

src/common/log/file_line_hk.go

+/*
+** description("Get the hook of the calling file name and line number").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/3/16 11:26).
+ */
+package log
+
+import (
+	"fmt"
+	"github.com/sirupsen/logrus"
+	"runtime"
+	"strings"
+)
+
+type fileHook struct{}
+
+func newFileHook() *fileHook {
+	return &fileHook{}
+}
+
+func (f *fileHook) Levels() []logrus.Level {
+	return logrus.AllLevels
+}
+
+func (f *fileHook) Fire(entry *logrus.Entry) error {
+	entry.Data["FilePath"] = findCaller(5)
+	return nil
+}
+
+func findCaller(skip int) string {
+	file := ""
+	line := 0
+	for i := 0; i < 10; i++ {
+		file, line = getCaller(skip + i)
+		if !strings.HasPrefix(file, "log") {
+			break
+		}
+	}
+	return fmt.Sprintf("%s:%d", file, line)
+}
+
+func getCaller(skip int) (string, int) {
+	_, file, line, ok := runtime.Caller(skip)
+	if !ok {
+		return "", 0
+	}
+
+	n := 0
+	for i := len(file) - 1; i > 0; i-- {
+		if file[i] == '/' {
+			n++
+			if n >= 2 {
+				file = file[i+1:]
+				break
+			}
+		}
+	}
+	return file, line
+}

src/common/log/logrus.go

+package log
+
+import (
+	"Open_IM/src/common/config"
+	"fmt"
+	nested "github.com/antonfisher/nested-logrus-formatter"
+	rotatelogs "github.com/lestrrat-go/file-rotatelogs"
+	"github.com/rifflock/lfshook"
+	"github.com/sirupsen/logrus"
+	"os"
+	"time"
+)
+
+var logger *Logger
+
+type Logger struct {
+	*logrus.Logger
+	Pid int
+}
+
+func init() {
+	logger = loggerInit("")
+
+}
+func NewPrivateLog(moduleName string) {
+	logger = loggerInit(moduleName)
+}
+
+func loggerInit(moduleName string) *Logger {
+	var logger = logrus.New()
+	//All logs will be printed
+	logger.SetLevel(logrus.TraceLevel)
+	//Log Style Setting
+	logger.SetFormatter(&nested.Formatter{
+		TimestampFormat: "2006-01-02 15:04:05",
+		HideKeys:        false,
+		FieldsOrder:     []string{"PID"},
+	})
+	//File name and line number display hook
+	logger.AddHook(newFileHook())
+
+	//Send logs to elasticsearch hook
+	if config.Config.Log.ElasticSearchSwitch == true {
+		logger.AddHook(newEsHook(moduleName))
+	}
+	//Log file segmentation hook
+	hook := NewLfsHook(config.Config.Log.StorageLocation+time.Now().Format("2006-01-02")+".log", 0, 5, moduleName)
+	logger.AddHook(hook)
+	return &Logger{
+		logger,
+		os.Getpid(),
+	}
+}
+func NewLfsHook(logName string, rotationTime time.Duration, maxRemainNum uint, moduleName string) logrus.Hook {
+	var fileNameSuffix string
+	if GetCurrentTimestamp() >= GetCurDayZeroTimestamp() && GetCurrentTimestamp() <= GetCurDayHalfTimestamp() {
+		fileNameSuffix = time.Now().Format("2006-01-02") + ".log"
+	} else {
+		fileNameSuffix = time.Now().Format("2006-01-02") + ".log"
+	}
+	writer, err := rotatelogs.New(
+		logName,
+		rotatelogs.WithRotationCount(maxRemainNum),
+	)
+	if err != nil {
+		panic(err)
+	}
+	writeInfo, err := rotatelogs.New(
+		config.Config.Log.StorageLocation+moduleName+"/info."+fileNameSuffix,
+		rotatelogs.WithRotationTime(time.Duration(60)*time.Second),
+		rotatelogs.WithRotationCount(maxRemainNum),
+	)
+	writeError, err := rotatelogs.New(
+		config.Config.Log.StorageLocation+moduleName+"/error."+fileNameSuffix,
+		rotatelogs.WithRotationTime(time.Minute),
+		rotatelogs.WithRotationCount(maxRemainNum),
+	)
+	writeDebug, err := rotatelogs.New(
+		config.Config.Log.StorageLocation+moduleName+"/debug."+fileNameSuffix,
+		rotatelogs.WithRotationCount(maxRemainNum),
+	)
+	writeWarn, err := rotatelogs.New(
+		config.Config.Log.StorageLocation+moduleName+"/warn."+fileNameSuffix,
+		rotatelogs.WithRotationTime(time.Minute),
+		rotatelogs.WithRotationCount(maxRemainNum),
+	)
+	if err != nil {
+		panic(err)
+	}
+	lfsHook := lfshook.NewHook(lfshook.WriterMap{
+		logrus.DebugLevel: writeDebug,
+		logrus.InfoLevel:  writeInfo,
+		logrus.WarnLevel:  writeWarn,
+		logrus.ErrorLevel: writeError,
+		logrus.FatalLevel: writer,
+		logrus.PanicLevel: writer,
+	}, &nested.Formatter{
+		TimestampFormat: "2006-01-02 15:04:05",
+		HideKeys:        false,
+		FieldsOrder:     []string{"PID"},
+	})
+
+	return lfsHook
+}
+
+func Info(token, OperationID, format string, args ...interface{}) {
+	if token == "" && OperationID == "" {
+		logger.WithFields(logrus.Fields{}).Infof(format, args...)
+	} else {
+		logger.WithFields(logrus.Fields{
+			"token":       token,
+			"OperationID": OperationID,
+		}).Infof(format, args...)
+	}
+}
+
+func Error(token, OperationID, format string, args ...interface{}) {
+	if token == "" && OperationID == "" {
+		logger.WithFields(logrus.Fields{}).Errorf(format, args...)
+	} else {
+		logger.WithFields(logrus.Fields{
+			"token":       token,
+			"OperationID": OperationID,
+		}).Errorf(format, args...)
+	}
+}
+
+func Debug(token, OperationID, format string, args ...interface{}) {
+	if token == "" && OperationID == "" {
+		logger.WithFields(logrus.Fields{}).Debugf(format, args...)
+	} else {
+		logger.WithFields(logrus.Fields{
+			"token":       token,
+			"OperationID": OperationID,
+		}).Debugf(format, args...)
+	}
+}
+
+func Warning(token, OperationID, format string, args ...interface{}) {
+	if token == "" && OperationID == "" {
+		logger.WithFields(logrus.Fields{}).Warningf(format, args...)
+	} else {
+		logger.WithFields(logrus.Fields{
+			"token":       token,
+			"OperationID": OperationID,
+		}).Warningf(format, args...)
+	}
+}
+
+func InfoByArgs(format string, args ...interface{}) {
+	logger.WithFields(logrus.Fields{}).Infof(format, args)
+}
+
+func ErrorByArgs(format string, args ...interface{}) {
+	logger.WithFields(logrus.Fields{}).Errorf(format, args...)
+}
+
+//Print log information in k, v format,
+//kv is best to appear in pairs. tipInfo is the log prompt information for printing,
+//and kv is the key and value for printing.
+func InfoByKv(tipInfo, OperationID string, args ...interface{}) {
+	fields := make(logrus.Fields)
+	argsHandle(OperationID, fields, args)
+	logger.WithFields(fields).Info(tipInfo)
+}
+func ErrorByKv(tipInfo, OperationID string, args ...interface{}) {
+	fields := make(logrus.Fields)
+	argsHandle(OperationID, fields, args)
+	logger.WithFields(fields).Error(tipInfo)
+}
+func DebugByKv(tipInfo, OperationID string, args ...interface{}) {
+	fields := make(logrus.Fields)
+	argsHandle(OperationID, fields, args)
+	logger.WithFields(fields).Debug(tipInfo)
+}
+func WarnByKv(tipInfo, OperationID string, args ...interface{}) {
+	fields := make(logrus.Fields)
+	argsHandle(OperationID, fields, args)
+	logger.WithFields(fields).Warn(tipInfo)
+}
+
+//internal method
+func argsHandle(OperationID string, fields logrus.Fields, args []interface{}) {
+	for i := 0; i < len(args); i += 2 {
+		if i+1 < len(args) {
+			fields[fmt.Sprintf("%v", args[i])] = args[i+1]
+		} else {
+			fields[fmt.Sprintf("%v", args[i])] = ""
+		}
+	}
+	fields["operationID"] = OperationID
+	fields["PID"] = logger.Pid
+}

src/common/log/time_format.go

+/*
+** description("").
+** copyright('tuoyun,www.tuoyun.net').
+** author("fg,Gordon@tuoyun.net").
+** time(2021/2/22 11:52).
+ */
+package log
+
+import (
+	"strconv"
+	"time"
+)
+
+const (
+	TimeOffset = 8 * 3600  //8 hour offset
+	HalfOffset = 12 * 3600 //Half-day hourly offset
+)
+
+//Get the current timestamp
+func GetCurrentTimestamp() int64 {
+	return time.Now().Unix()
+}
+
+//Get the timestamp at 0 o'clock of the day
+func GetCurDayZeroTimestamp() int64 {
+	timeStr := time.Now().Format("2006-01-02")
+	t, _ := time.Parse("2006-01-02", timeStr)
+	return t.Unix() - TimeOffset
+}
+
+//Get the timestamp at 12 o'clock on the day
+func GetCurDayHalfTimestamp() int64 {
+	return GetCurDayZeroTimestamp() + HalfOffset
+
+}
+
+//Get the formatted time at 0 o'clock of the day, the format is "2006-01-02_00-00-00"
+func GetCurDayZeroTimeFormat() string {
+	return time.Unix(GetCurDayZeroTimestamp(), 0).Format("2006-01-02_15-04-05")
+}
+
+//Get the formatted time at 12 o'clock of the day, the format is "2006-01-02_12-00-00"
+func GetCurDayHalfTimeFormat() string {
+	return time.Unix(GetCurDayZeroTimestamp()+HalfOffset, 0).Format("2006-01-02_15-04-05")
+}
+func GetTimeStampByFormat(datetime string) string {
+	timeLayout := "2006-01-02 15:04:05"  //Template required for transformation
+	loc, _ := time.LoadLocation("Local") //Get time zone
+	tmp, _ := time.ParseInLocation(timeLayout, datetime, loc)
+	timestamp := tmp.Unix() //Converted to timestamp type is int64
+	return strconv.FormatInt(timestamp, 10)
+}
+
+func TimeStringFormatTimeUnix(timeFormat string, timeSrc string) int64 {
+	tm, _ := time.Parse(timeFormat, timeSrc)
+	return tm.Unix()
+}

src/proto/push/push.pb.go

+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: push/push.proto
+
+package pbPush
+
+import (
+	context "context"
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+	math "math"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+type PushMsgReq struct {
+	SendID               string   `protobuf:"bytes,1,opt,name=SendID,proto3" json:"SendID,omitempty"`
+	RecvID               string   `protobuf:"bytes,2,opt,name=RecvID,proto3" json:"RecvID,omitempty"`
+	Content              string   `protobuf:"bytes,3,opt,name=Content,proto3" json:"Content,omitempty"`
+	RecvSeq              int64    `protobuf:"varint,4,opt,name=RecvSeq,proto3" json:"RecvSeq,omitempty"`
+	SendTime             int64    `protobuf:"varint,5,opt,name=SendTime,proto3" json:"SendTime,omitempty"`
+	MsgFrom              int32    `protobuf:"varint,6,opt,name=MsgFrom,proto3" json:"MsgFrom,omitempty"`
+	ContentType          int32    `protobuf:"varint,7,opt,name=ContentType,proto3" json:"ContentType,omitempty"`
+	SessionType          int32    `protobuf:"varint,8,opt,name=SessionType,proto3" json:"SessionType,omitempty"`
+	OperationID          string   `protobuf:"bytes,9,opt,name=OperationID,proto3" json:"OperationID,omitempty"`
+	MsgID                string   `protobuf:"bytes,10,opt,name=MsgID,proto3" json:"MsgID,omitempty"`
+	OfflineInfo          string   `protobuf:"bytes,11,opt,name=OfflineInfo,proto3" json:"OfflineInfo,omitempty"`
+	Options              string   `protobuf:"bytes,12,opt,name=Options,proto3" json:"Options,omitempty"`
+	PlatformID           int32    `protobuf:"varint,13,opt,name=PlatformID,proto3" json:"PlatformID,omitempty"`
+	IsEmphasize          bool     `protobuf:"varint,14,opt,name=IsEmphasize,proto3" json:"IsEmphasize,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *PushMsgReq) Reset()         { *m = PushMsgReq{} }
+func (m *PushMsgReq) String() string { return proto.CompactTextString(m) }
+func (*PushMsgReq) ProtoMessage()    {}
+func (*PushMsgReq) Descriptor() ([]byte, []int) {
+	return fileDescriptor_ae0042da44e9a7a7, []int{0}
+}
+
+func (m *PushMsgReq) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_PushMsgReq.Unmarshal(m, b)
+}
+func (m *PushMsgReq) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_PushMsgReq.Marshal(b, m, deterministic)
+}
+func (m *PushMsgReq) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_PushMsgReq.Merge(m, src)
+}
+func (m *PushMsgReq) XXX_Size() int {
+	return xxx_messageInfo_PushMsgReq.Size(m)
+}
+func (m *PushMsgReq) XXX_DiscardUnknown() {
+	xxx_messageInfo_PushMsgReq.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_PushMsgReq proto.InternalMessageInfo
+
+func (m *PushMsgReq) GetSendID() string {
+	if m != nil {
+		return m.SendID
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetRecvID() string {
+	if m != nil {
+		return m.RecvID
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetContent() string {
+	if m != nil {
+		return m.Content
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetRecvSeq() int64 {
+	if m != nil {
+		return m.RecvSeq
+	}
+	return 0
+}
+
+func (m *PushMsgReq) GetSendTime() int64 {
+	if m != nil {
+		return m.SendTime
+	}
+	return 0
+}
+
+func (m *PushMsgReq) GetMsgFrom() int32 {
+	if m != nil {
+		return m.MsgFrom
+	}
+	return 0
+}
+
+func (m *PushMsgReq) GetContentType() int32 {
+	if m != nil {
+		return m.ContentType
+	}
+	return 0
+}
+
+func (m *PushMsgReq) GetSessionType() int32 {
+	if m != nil {
+		return m.SessionType
+	}
+	return 0
+}
+
+func (m *PushMsgReq) GetOperationID() string {
+	if m != nil {
+		return m.OperationID
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetMsgID() string {
+	if m != nil {
+		return m.MsgID
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetOfflineInfo() string {
+	if m != nil {
+		return m.OfflineInfo
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetOptions() string {
+	if m != nil {
+		return m.Options
+	}
+	return ""
+}
+
+func (m *PushMsgReq) GetPlatformID() int32 {
+	if m != nil {
+		return m.PlatformID
+	}
+	return 0
+}
+
+func (m *PushMsgReq) GetIsEmphasize() bool {
+	if m != nil {
+		return m.IsEmphasize
+	}
+	return false
+}
+
+type PushMsgResp struct {
+	ResultCode           int32    `protobuf:"varint,1,opt,name=ResultCode,proto3" json:"ResultCode,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *PushMsgResp) Reset()         { *m = PushMsgResp{} }
+func (m *PushMsgResp) String() string { return proto.CompactTextString(m) }
+func (*PushMsgResp) ProtoMessage()    {}
+func (*PushMsgResp) Descriptor() ([]byte, []int) {
+	return fileDescriptor_ae0042da44e9a7a7, []int{1}
+}
+
+func (m *PushMsgResp) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_PushMsgResp.Unmarshal(m, b)
+}
+func (m *PushMsgResp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_PushMsgResp.Marshal(b, m, deterministic)
+}
+func (m *PushMsgResp) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_PushMsgResp.Merge(m, src)
+}
+func (m *PushMsgResp) XXX_Size() int {
+	return xxx_messageInfo_PushMsgResp.Size(m)
+}
+func (m *PushMsgResp) XXX_DiscardUnknown() {
+	xxx_messageInfo_PushMsgResp.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_PushMsgResp proto.InternalMessageInfo
+
+func (m *PushMsgResp) GetResultCode() int32 {
+	if m != nil {
+		return m.ResultCode
+	}
+	return 0
+}
+
+func init() {
+	proto.RegisterType((*PushMsgReq)(nil), "push.PushMsgReq")
+	proto.RegisterType((*PushMsgResp)(nil), "push.PushMsgResp")
+}
+
+func init() { proto.RegisterFile("push/push.proto", fileDescriptor_ae0042da44e9a7a7) }
+
+var fileDescriptor_ae0042da44e9a7a7 = []byte{
+	// 346 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x5c, 0x92, 0x4d, 0x6b, 0xf2, 0x40,
+	0x14, 0x85, 0xc9, 0xab, 0xf1, 0x63, 0xf2, 0x6a, 0xdb, 0xa1, 0x94, 0x8b, 0x8b, 0x12, 0x5c, 0x65,
+	0x53, 0x0b, 0xed, 0xb2, 0x9b, 0x52, 0xd3, 0x42, 0x16, 0xa2, 0x4c, 0x5c, 0x75, 0xe7, 0xc7, 0x8d,
+	0x06, 0xcc, 0xcc, 0x98, 0x89, 0x42, 0xfb, 0x6b, 0xfa, 0x53, 0xcb, 0x9d, 0x44, 0x0d, 0xdd, 0x88,
+	0xe7, 0x39, 0xe7, 0x1e, 0x2e, 0xb9, 0xc3, 0xae, 0xf4, 0xc1, 0x6c, 0x1f, 0xe9, 0x67, 0xa4, 0x73,
+	0x55, 0x28, 0xde, 0xa4, 0xff, 0xc3, 0x9f, 0x06, 0x63, 0xb3, 0x83, 0xd9, 0x4e, 0xcc, 0x46, 0xe0,
+	0x9e, 0xdf, 0xb1, 0x56, 0x8c, 0x72, 0x1d, 0x85, 0xe0, 0xf8, 0x4e, 0xd0, 0x15, 0x95, 0x22, 0x2e,
+	0x70, 0x75, 0x8c, 0x42, 0xf8, 0x57, 0xf2, 0x52, 0x71, 0x60, 0xed, 0xb1, 0x92, 0x05, 0xca, 0x02,
+	0x1a, 0xd6, 0x38, 0x49, 0x72, 0x28, 0x13, 0xe3, 0x1e, 0x9a, 0xbe, 0x13, 0x34, 0xc4, 0x49, 0xf2,
+	0x01, 0xeb, 0x50, 0xeb, 0x3c, 0xcd, 0x10, 0x5c, 0x6b, 0x9d, 0x35, 0x4d, 0x4d, 0xcc, 0xe6, 0x23,
+	0x57, 0x19, 0xb4, 0x7c, 0x27, 0x70, 0xc5, 0x49, 0x72, 0x9f, 0x79, 0x55, 0xf5, 0xfc, 0x4b, 0x23,
+	0xb4, 0xad, 0x5b, 0x47, 0x94, 0x88, 0xd1, 0x98, 0x54, 0x49, 0x9b, 0xe8, 0x94, 0x89, 0x1a, 0xa2,
+	0xc4, 0x54, 0x63, 0xbe, 0x28, 0x52, 0x25, 0xa3, 0x10, 0xba, 0x76, 0xe3, 0x3a, 0xe2, 0xb7, 0xcc,
+	0x9d, 0x98, 0x4d, 0x14, 0x02, 0xb3, 0x5e, 0x29, 0xec, 0x5c, 0x92, 0xec, 0x52, 0x89, 0x91, 0x4c,
+	0x14, 0x78, 0xd5, 0xdc, 0x05, 0xd1, 0xde, 0x53, 0x4d, 0x1d, 0x06, 0xfe, 0x97, 0xdf, 0xa1, 0x92,
+	0xfc, 0x9e, 0xb1, 0xd9, 0x6e, 0x51, 0x24, 0x2a, 0xcf, 0xa2, 0x10, 0x7a, 0x76, 0xa9, 0x1a, 0xa1,
+	0xee, 0xc8, 0xbc, 0x67, 0x7a, 0xbb, 0x30, 0xe9, 0x37, 0x42, 0xdf, 0x77, 0x82, 0x8e, 0xa8, 0xa3,
+	0xe1, 0x03, 0xf3, 0xce, 0x17, 0x32, 0x9a, 0x0a, 0x05, 0x9a, 0xc3, 0xae, 0x18, 0xab, 0x35, 0xda,
+	0x33, 0xb9, 0xa2, 0x46, 0x9e, 0x5e, 0x59, 0xbf, 0x8a, 0xc7, 0x98, 0x1f, 0xd3, 0x15, 0xf2, 0x11,
+	0x6b, 0x57, 0x84, 0x5f, 0x8f, 0xec, 0x0b, 0xb8, 0x5c, 0x7c, 0x70, 0xf3, 0x87, 0x18, 0xfd, 0xd6,
+	0xfb, 0xf4, 0x88, 0xbd, 0xe8, 0x25, 0xd1, 0x65, 0xcb, 0xbe, 0x97, 0xe7, 0xdf, 0x00, 0x00, 0x00,
+	0xff, 0xff, 0x20, 0x4b, 0x09, 0x6a, 0x42, 0x02, 0x00, 0x00,
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// PushMsgServiceClient is the client API for PushMsgService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type PushMsgServiceClient interface {
+	PushMsg(ctx context.Context, in *PushMsgReq, opts ...grpc.CallOption) (*PushMsgResp, error)
+}
+
+type pushMsgServiceClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewPushMsgServiceClient(cc *grpc.ClientConn) PushMsgServiceClient {
+	return &pushMsgServiceClient{cc}
+}
+
+func (c *pushMsgServiceClient) PushMsg(ctx context.Context, in *PushMsgReq, opts ...grpc.CallOption) (*PushMsgResp, error) {
+	out := new(PushMsgResp)
+	err := c.cc.Invoke(ctx, "/push.PushMsgService/PushMsg", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// PushMsgServiceServer is the server API for PushMsgService service.
+type PushMsgServiceServer interface {
+	PushMsg(context.Context, *PushMsgReq) (*PushMsgResp, error)
+}
+
+// UnimplementedPushMsgServiceServer can be embedded to have forward compatible implementations.
+type UnimplementedPushMsgServiceServer struct {
+}
+
+func (*UnimplementedPushMsgServiceServer) PushMsg(ctx context.Context, req *PushMsgReq) (*PushMsgResp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method PushMsg not implemented")
+}
+
+func RegisterPushMsgServiceServer(s *grpc.Server, srv PushMsgServiceServer) {
+	s.RegisterService(&_PushMsgService_serviceDesc, srv)
+}
+
+func _PushMsgService_PushMsg_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PushMsgReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(PushMsgServiceServer).PushMsg(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/push.PushMsgService/PushMsg",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(PushMsgServiceServer).PushMsg(ctx, req.(*PushMsgReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+var _PushMsgService_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "push.PushMsgService",
+	HandlerType: (*PushMsgServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "PushMsg",
+			Handler:    _PushMsgService_PushMsg_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "push/push.proto",
+}

src/proto/push/push.proto

+syntax = "proto3";
+option go_package = "push;pbPush";
+package push;
+
+message PushMsgReq {
+  string SendID = 1;
+  string RecvID = 2;
+  string Content = 3;
+  int64 RecvSeq = 4;
+  int64 SendTime = 5;
+  int32 MsgFrom = 6;
+  int32 ContentType = 7;
+  int32 SessionType = 8;
+  string OperationID = 9;
+  string MsgID = 10;
+  string OfflineInfo = 11;
+  string Options =12;
+  int32  PlatformID =13;
+  bool   IsEmphasize = 14;
+}
+message PushMsgResp{
+  int32   ResultCode = 1;
+}
+//message InternalPushMsgReq{
+//  int32 ReqIdentifier = 1;
+//  string Token = 2;
+//  string SendID = 3;
+//  string OperationID = 4;
+//  int32 MsgIncr = 5;
+//  int32 PlatformID = 6;
+//  int32 SessionType = 7;
+//  int32 MsgFrom = 8;
+//  int32 ContentType = 9;
+//  string RecvID = 10;
+//  repeated string ForceList = 11;
+//  string Content = 12;
+//  string Options = 13;
+//  string ClientMsgID = 14;
+//  string OffLineInfo = 15;
+//  string Ex = 16;
+//
+//}
+
+service PushMsgService {
+  rpc PushMsg(PushMsgReq) returns(PushMsgResp);
+//  rpc InternalPushMsg(InternalPushMsgReq)returns(PushMsgResp);
+}
+

src/proto/relay/relay.pb.go

+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: relay/relay.proto
+
+package pbRelay
+
+import (
+	context "context"
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+	math "math"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+type MsgToUserReq struct {
+	SendID               string   `protobuf:"bytes,1,opt,name=SendID,proto3" json:"SendID,omitempty"`
+	RecvID               string   `protobuf:"bytes,2,opt,name=RecvID,proto3" json:"RecvID,omitempty"`
+	Content              string   `protobuf:"bytes,5,opt,name=Content,proto3" json:"Content,omitempty"`
+	RecvSeq              int64    `protobuf:"varint,6,opt,name=RecvSeq,proto3" json:"RecvSeq,omitempty"`
+	SendTime             int64    `protobuf:"varint,7,opt,name=SendTime,proto3" json:"SendTime,omitempty"`
+	MsgFrom              int32    `protobuf:"varint,8,opt,name=MsgFrom,proto3" json:"MsgFrom,omitempty"`
+	ContentType          int32    `protobuf:"varint,9,opt,name=ContentType,proto3" json:"ContentType,omitempty"`
+	SessionType          int32    `protobuf:"varint,10,opt,name=SessionType,proto3" json:"SessionType,omitempty"`
+	OperationID          string   `protobuf:"bytes,11,opt,name=OperationID,proto3" json:"OperationID,omitempty"`
+	ServerMsgID          string   `protobuf:"bytes,12,opt,name=ServerMsgID,proto3" json:"ServerMsgID,omitempty"`
+	PlatformID           int32    `protobuf:"varint,13,opt,name=PlatformID,proto3" json:"PlatformID,omitempty"`
+	IsEmphasize          bool     `protobuf:"varint,14,opt,name=IsEmphasize,proto3" json:"IsEmphasize,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *MsgToUserReq) Reset()         { *m = MsgToUserReq{} }
+func (m *MsgToUserReq) String() string { return proto.CompactTextString(m) }
+func (*MsgToUserReq) ProtoMessage()    {}
+func (*MsgToUserReq) Descriptor() ([]byte, []int) {
+	return fileDescriptor_b6da3b5c0d1535b3, []int{0}
+}
+
+func (m *MsgToUserReq) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_MsgToUserReq.Unmarshal(m, b)
+}
+func (m *MsgToUserReq) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_MsgToUserReq.Marshal(b, m, deterministic)
+}
+func (m *MsgToUserReq) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_MsgToUserReq.Merge(m, src)
+}
+func (m *MsgToUserReq) XXX_Size() int {
+	return xxx_messageInfo_MsgToUserReq.Size(m)
+}
+func (m *MsgToUserReq) XXX_DiscardUnknown() {
+	xxx_messageInfo_MsgToUserReq.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_MsgToUserReq proto.InternalMessageInfo
+
+func (m *MsgToUserReq) GetSendID() string {
+	if m != nil {
+		return m.SendID
+	}
+	return ""
+}
+
+func (m *MsgToUserReq) GetRecvID() string {
+	if m != nil {
+		return m.RecvID
+	}
+	return ""
+}
+
+func (m *MsgToUserReq) GetContent() string {
+	if m != nil {
+		return m.Content
+	}
+	return ""
+}
+
+func (m *MsgToUserReq) GetRecvSeq() int64 {
+	if m != nil {
+		return m.RecvSeq
+	}
+	return 0
+}
+
+func (m *MsgToUserReq) GetSendTime() int64 {
+	if m != nil {
+		return m.SendTime
+	}
+	return 0
+}
+
+func (m *MsgToUserReq) GetMsgFrom() int32 {
+	if m != nil {
+		return m.MsgFrom
+	}
+	return 0
+}
+
+func (m *MsgToUserReq) GetContentType() int32 {
+	if m != nil {
+		return m.ContentType
+	}
+	return 0
+}
+
+func (m *MsgToUserReq) GetSessionType() int32 {
+	if m != nil {
+		return m.SessionType
+	}
+	return 0
+}
+
+func (m *MsgToUserReq) GetOperationID() string {
+	if m != nil {
+		return m.OperationID
+	}
+	return ""
+}
+
+func (m *MsgToUserReq) GetServerMsgID() string {
+	if m != nil {
+		return m.ServerMsgID
+	}
+	return ""
+}
+
+func (m *MsgToUserReq) GetPlatformID() int32 {
+	if m != nil {
+		return m.PlatformID
+	}
+	return 0
+}
+
+func (m *MsgToUserReq) GetIsEmphasize() bool {
+	if m != nil {
+		return m.IsEmphasize
+	}
+	return false
+}
+
+type MsgToUserResp struct {
+	Resp                 []*SingleMsgToUser `protobuf:"bytes,1,rep,name=resp,proto3" json:"resp,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
+	XXX_unrecognized     []byte             `json:"-"`
+	XXX_sizecache        int32              `json:"-"`
+}
+
+func (m *MsgToUserResp) Reset()         { *m = MsgToUserResp{} }
+func (m *MsgToUserResp) String() string { return proto.CompactTextString(m) }
+func (*MsgToUserResp) ProtoMessage()    {}
+func (*MsgToUserResp) Descriptor() ([]byte, []int) {
+	return fileDescriptor_b6da3b5c0d1535b3, []int{1}
+}
+
+func (m *MsgToUserResp) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_MsgToUserResp.Unmarshal(m, b)
+}
+func (m *MsgToUserResp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_MsgToUserResp.Marshal(b, m, deterministic)
+}
+func (m *MsgToUserResp) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_MsgToUserResp.Merge(m, src)
+}
+func (m *MsgToUserResp) XXX_Size() int {
+	return xxx_messageInfo_MsgToUserResp.Size(m)
+}
+func (m *MsgToUserResp) XXX_DiscardUnknown() {
+	xxx_messageInfo_MsgToUserResp.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_MsgToUserResp proto.InternalMessageInfo
+
+func (m *MsgToUserResp) GetResp() []*SingleMsgToUser {
+	if m != nil {
+		return m.Resp
+	}
+	return nil
+}
+
+//message SendMsgByWSReq{
+//  string SendID = 1;
+//  string RecvID = 2;
+//  string Content = 3;
+//  int64 SendTime = 4;
+//  int64  MsgFrom = 5;
+//  int64  ContentType = 6;
+//  int64  SessionType = 7;
+//  string OperationID = 8;
+//  int64  PlatformID = 9;
+//}
+type SingleMsgToUser struct {
+	ResultCode           int64    `protobuf:"varint,1,opt,name=ResultCode,proto3" json:"ResultCode,omitempty"`
+	RecvID               string   `protobuf:"bytes,2,opt,name=RecvID,proto3" json:"RecvID,omitempty"`
+	RecvPlatFormID       int32    `protobuf:"varint,3,opt,name=RecvPlatFormID,proto3" json:"RecvPlatFormID,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *SingleMsgToUser) Reset()         { *m = SingleMsgToUser{} }
+func (m *SingleMsgToUser) String() string { return proto.CompactTextString(m) }
+func (*SingleMsgToUser) ProtoMessage()    {}
+func (*SingleMsgToUser) Descriptor() ([]byte, []int) {
+	return fileDescriptor_b6da3b5c0d1535b3, []int{2}
+}
+
+func (m *SingleMsgToUser) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_SingleMsgToUser.Unmarshal(m, b)
+}
+func (m *SingleMsgToUser) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_SingleMsgToUser.Marshal(b, m, deterministic)
+}
+func (m *SingleMsgToUser) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_SingleMsgToUser.Merge(m, src)
+}
+func (m *SingleMsgToUser) XXX_Size() int {
+	return xxx_messageInfo_SingleMsgToUser.Size(m)
+}
+func (m *SingleMsgToUser) XXX_DiscardUnknown() {
+	xxx_messageInfo_SingleMsgToUser.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_SingleMsgToUser proto.InternalMessageInfo
+
+func (m *SingleMsgToUser) GetResultCode() int64 {
+	if m != nil {
+		return m.ResultCode
+	}
+	return 0
+}
+
+func (m *SingleMsgToUser) GetRecvID() string {
+	if m != nil {
+		return m.RecvID
+	}
+	return ""
+}
+
+func (m *SingleMsgToUser) GetRecvPlatFormID() int32 {
+	if m != nil {
+		return m.RecvPlatFormID
+	}
+	return 0
+}
+
+func init() {
+	proto.RegisterType((*MsgToUserReq)(nil), "relay.MsgToUserReq")
+	proto.RegisterType((*MsgToUserResp)(nil), "relay.MsgToUserResp")
+	proto.RegisterType((*SingleMsgToUser)(nil), "relay.SingleMsgToUser")
+}
+
+func init() { proto.RegisterFile("relay/relay.proto", fileDescriptor_b6da3b5c0d1535b3) }
+
+var fileDescriptor_b6da3b5c0d1535b3 = []byte{
+	// 381 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x92, 0xcb, 0x6a, 0xe3, 0x30,
+	0x18, 0x85, 0x71, 0x3c, 0xb9, 0x29, 0x37, 0x46, 0x33, 0x04, 0x4d, 0x16, 0x83, 0xc9, 0xa2, 0x98,
+	0x2e, 0x52, 0x48, 0xa1, 0x9b, 0xec, 0x1a, 0x37, 0xe0, 0x85, 0x49, 0x91, 0xd3, 0x4d, 0x77, 0x4e,
+	0xf2, 0xd7, 0x35, 0xd8, 0x92, 0x22, 0xb9, 0x81, 0xf4, 0xb1, 0xfb, 0x04, 0x45, 0x72, 0xd2, 0x88,
+	0x94, 0x6e, 0x8c, 0xcf, 0xf7, 0x1f, 0x1f, 0x1f, 0x5d, 0xd0, 0x6f, 0x09, 0x79, 0x72, 0xb8, 0x31,
+	0xcf, 0x89, 0x90, 0xbc, 0xe4, 0xb8, 0x6e, 0xc4, 0xf8, 0xa3, 0x86, 0xba, 0x91, 0x4a, 0x57, 0xfc,
+	0x49, 0x81, 0xa4, 0xb0, 0xc3, 0x43, 0xd4, 0x88, 0x81, 0x6d, 0xc3, 0x80, 0x38, 0x9e, 0xe3, 0xb7,
+	0xe9, 0x51, 0x69, 0x4e, 0x61, 0xb3, 0x0f, 0x03, 0x52, 0xab, 0x78, 0xa5, 0x30, 0x41, 0xcd, 0x39,
+	0x67, 0x25, 0xb0, 0x92, 0xd4, 0xcd, 0xe0, 0x24, 0xf5, 0x44, 0x7b, 0x62, 0xd8, 0x91, 0x86, 0xe7,
+	0xf8, 0x2e, 0x3d, 0x49, 0x3c, 0x42, 0x2d, 0x9d, 0xba, 0xca, 0x0a, 0x20, 0x4d, 0x33, 0xfa, 0xd2,
+	0xfa, 0xab, 0x48, 0xa5, 0x0b, 0xc9, 0x0b, 0xd2, 0xf2, 0x1c, 0xbf, 0x4e, 0x4f, 0x12, 0x7b, 0xa8,
+	0x73, 0x8c, 0x5e, 0x1d, 0x04, 0x90, 0xb6, 0x99, 0xda, 0x48, 0x3b, 0x62, 0x50, 0x2a, 0xe3, 0xcc,
+	0x38, 0x50, 0xe5, 0xb0, 0x90, 0x76, 0x2c, 0x05, 0xc8, 0xa4, 0xcc, 0x38, 0x0b, 0x03, 0xd2, 0x31,
+	0x8d, 0x6d, 0x54, 0x65, 0xc8, 0x3d, 0xc8, 0x48, 0xa5, 0x61, 0x40, 0xba, 0x95, 0xc3, 0x42, 0xf8,
+	0x3f, 0x42, 0x8f, 0x79, 0x52, 0xbe, 0x70, 0x59, 0x84, 0x01, 0xe9, 0x99, 0x9f, 0x58, 0x44, 0x27,
+	0x84, 0xea, 0xa1, 0x10, 0xaf, 0x89, 0xca, 0xde, 0x81, 0xf4, 0x3d, 0xc7, 0x6f, 0x51, 0x1b, 0x8d,
+	0x67, 0xa8, 0x67, 0xed, 0xb9, 0x12, 0xf8, 0x1a, 0xfd, 0x92, 0xa0, 0x04, 0x71, 0x3c, 0xd7, 0xef,
+	0x4c, 0x87, 0x93, 0xea, 0xa0, 0xe2, 0x8c, 0xa5, 0x39, 0x9c, 0x9d, 0xc6, 0x33, 0xde, 0xa1, 0xc1,
+	0xc5, 0x40, 0x37, 0xa2, 0xa0, 0xde, 0xf2, 0x72, 0xce, 0xb7, 0x60, 0xce, 0xcd, 0xa5, 0x16, 0xf9,
+	0xf1, 0xec, 0xae, 0x50, 0x5f, 0xbf, 0xe9, 0xee, 0x8b, 0x6a, 0x35, 0xae, 0x59, 0xcd, 0x05, 0x9d,
+	0xc6, 0xe8, 0xdf, 0x92, 0xe5, 0x19, 0x83, 0x08, 0x94, 0x4a, 0x52, 0xa0, 0xba, 0x9e, 0xde, 0x92,
+	0x6c, 0x03, 0xf8, 0x0e, 0xb5, 0xcf, 0x4d, 0xfe, 0x1c, 0xab, 0xdb, 0x57, 0x6a, 0xf4, 0xf7, 0x3b,
+	0x54, 0xe2, 0x7e, 0xf0, 0xdc, 0x33, 0x78, 0x26, 0xd6, 0x26, 0x6f, 0xdd, 0x30, 0x17, 0xf3, 0xf6,
+	0x33, 0x00, 0x00, 0xff, 0xff, 0x57, 0x9a, 0x71, 0x03, 0xad, 0x02, 0x00, 0x00,
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// OnlineMessageRelayServiceClient is the client API for OnlineMessageRelayService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type OnlineMessageRelayServiceClient interface {
+	MsgToUser(ctx context.Context, in *MsgToUserReq, opts ...grpc.CallOption) (*MsgToUserResp, error)
+}
+
+type onlineMessageRelayServiceClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewOnlineMessageRelayServiceClient(cc *grpc.ClientConn) OnlineMessageRelayServiceClient {
+	return &onlineMessageRelayServiceClient{cc}
+}
+
+func (c *onlineMessageRelayServiceClient) MsgToUser(ctx context.Context, in *MsgToUserReq, opts ...grpc.CallOption) (*MsgToUserResp, error) {
+	out := new(MsgToUserResp)
+	err := c.cc.Invoke(ctx, "/relay.OnlineMessageRelayService/MsgToUser", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// OnlineMessageRelayServiceServer is the server API for OnlineMessageRelayService service.
+type OnlineMessageRelayServiceServer interface {
+	MsgToUser(context.Context, *MsgToUserReq) (*MsgToUserResp, error)
+}
+
+// UnimplementedOnlineMessageRelayServiceServer can be embedded to have forward compatible implementations.
+type UnimplementedOnlineMessageRelayServiceServer struct {
+}
+
+func (*UnimplementedOnlineMessageRelayServiceServer) MsgToUser(ctx context.Context, req *MsgToUserReq) (*MsgToUserResp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method MsgToUser not implemented")
+}
+
+func RegisterOnlineMessageRelayServiceServer(s *grpc.Server, srv OnlineMessageRelayServiceServer) {
+	s.RegisterService(&_OnlineMessageRelayService_serviceDesc, srv)
+}
+
+func _OnlineMessageRelayService_MsgToUser_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(MsgToUserReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(OnlineMessageRelayServiceServer).MsgToUser(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/relay.OnlineMessageRelayService/MsgToUser",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(OnlineMessageRelayServiceServer).MsgToUser(ctx, req.(*MsgToUserReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+var _OnlineMessageRelayService_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "relay.OnlineMessageRelayService",
+	HandlerType: (*OnlineMessageRelayServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "MsgToUser",
+			Handler:    _OnlineMessageRelayService_MsgToUser_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "relay/relay.proto",
+}

src/proto/relay/relay.proto

+syntax = "proto3";
+option go_package = "relay;pbRelay";
+package relay;
+
+message MsgToUserReq {
+  string SendID = 1;
+  string RecvID = 2;
+  string Content = 5;
+  int64  RecvSeq = 6;
+  int64 SendTime = 7;
+  int32  MsgFrom = 8;
+  int32  ContentType = 9;
+  int32  SessionType = 10;
+  string OperationID = 11;
+  string ServerMsgID = 12;
+  int32  PlatformID = 13;
+  bool   IsEmphasize = 14;
+}
+message MsgToUserResp{
+repeated SingleMsgToUser resp = 1;
+}
+//message SendMsgByWSReq{
+//  string SendID = 1;
+//  string RecvID = 2;
+//  string Content = 3;
+//  int64 SendTime = 4;
+//  int64  MsgFrom = 5;
+//  int64  ContentType = 6;
+//  int64  SessionType = 7;
+//  string OperationID = 8;
+//  int64  PlatformID = 9;
+//}
+message SingleMsgToUser{
+  int64   ResultCode = 1;
+  string  RecvID = 2;
+  int32   RecvPlatFormID = 3;
+}
+service OnlineMessageRelayService {
+  rpc MsgToUser(MsgToUserReq) returns(MsgToUserResp);
+//  rpc SendMsgByWS(SendMsgByWSReq)  returns(MsgToUserResp);
+}
+

src/rpc/auth/open_im_auth.go

+package main
+
+import (
+	rpcAuth "Open_IM/src/rpc/auth/auth"
+	"flag"
+)
+
+func main() {
+	rpcPort := flag.Int("port", 10600, "RpcToken default listen port 10800")
+	flag.Parse()
+	rpcServer := rpcAuth.NewRpcAuthServer(*rpcPort)
+	rpcServer.Run()
+}

src/rpc/chat/Makefile

+.PHONY: all build run gotool install clean help
+
+BINARY_NAME=open_im_msg
+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/rpc/chat/chat/pull_message.go

+//实现pb定义的rpc服务
+package rpcChat
+
+import (
+	"context"
+
+	commonDB "Open_IM/src/common/db"
+	"Open_IM/src/common/log"
+
+	"sort"
+	"strings"
+
+	pbMsg "Open_IM/src/proto/chat"
+)
+
+func (rpc *rpcChat) GetNewSeq(_ context.Context, in *pbMsg.GetNewSeqReq) (*pbMsg.GetNewSeqResp, error) {
+	log.InfoByKv("rpc getNewSeq is arriving", in.OperationID, in.String())
+	//seq, err := model.GetBiggestSeqFromReceive(in.UserID)
+	seq, err := commonDB.DB.GetUserSeq(in.UserID)
+	resp := new(pbMsg.GetNewSeqResp)
+	if err == nil {
+		resp.Seq = seq
+		resp.ErrCode = 0
+		resp.ErrMsg = ""
+		return resp, err
+	} else {
+		log.ErrorByKv("getSeq from redis error", in.OperationID, "args", in.String(), "err", err.Error())
+		resp.Seq = 0
+		resp.ErrCode = 0
+		resp.ErrMsg = ""
+		return resp, nil
+	}
+
+}
+
+//func (s *MsgServer) PullMessage(_ context.Context, in *pbMsg.PullMessageReq) (*pbMsg.PullMessageResp, error) {
+//	log.InfoByArgs(fmt.Sprintf("rpc pullMessage is arriving,args=%s", in.String()))
+//	resp := new(pbMsg.PullMessageResp)
+//	var respMsgFormat []*pbMsg.MsgFormat
+//	var respUserMsgFormat []*pbMsg.UserMsgFormat
+//	conn := db.NewDbConnection()
+//	rows, err := conn.Table("receive r").Select("c.sender_id,c.receiver_id,"+
+//		"c.msg_type,c.push_msg_type,c.chat_type,c.msg_id,c.send_content,r.seq,c.send_time,c.sender_nickname,c.receiver_nickname,c.sender_head_url,c.receiver_head_url").
+//		Joins("INNER JOIN chat_log c ON r.msg_id = c.msg_id AND r.user_id = ? AND seq BETWEEN ? AND ?",
+//			in.UserID, in.SeqBegin, in.SeqEnd).Rows()
+//	if err != nil {
+//		fmt.Printf("pullMsg data error: %v\n", err)
+//		resp.ErrCode = 1
+//		resp.ErrMsg = err.Error()
+//		return resp, nil
+//	}
+//	defer rows.Close()
+//	for rows.Next() {
+//		tempResp := new(pbMsg.MsgFormat)
+//		rows.Scan(&tempResp.SendID, &tempResp.RecvID, &tempResp.MsgType, &tempResp.PushMsgType, &tempResp.ChatType,
+//			&tempResp.MsgID, &tempResp.Msg, &tempResp.Seq, &tempResp.Time, &tempResp.SendNickName, &tempResp.RecvNickName,
+//			&tempResp.SendHeadUrl, &tempResp.RecvHeadUrl)
+//		respMsgFormat = append(respMsgFormat, tempResp)
+//	}
+//	respUserMsgFormat = msgHandleByUser(respMsgFormat, in.UserID)
+//	return &pbMsg.PullMessageResp{
+//		ErrCode: 0,
+//		ErrMsg:  "",
+//		UserMsg: respUserMsgFormat,
+//	}, nil
+//}
+func (rpc *rpcChat) PullMessage(_ context.Context, in *pbMsg.PullMessageReq) (*pbMsg.PullMessageResp, error) {
+	log.InfoByKv("rpc pullMessage is arriving", in.OperationID, "args", in.String())
+	resp := new(pbMsg.PullMessageResp)
+	var respSingleMsgFormat []*pbMsg.GatherFormat
+	var respGroupMsgFormat []*pbMsg.GatherFormat
+	SingleMsgFormat, GroupMsgFormat, MaxSeq, MinSeq, err := commonDB.DB.GetUserChat(in.UserID, in.SeqBegin, in.SeqEnd)
+	if err != nil {
+		log.ErrorByKv("pullMsg data error", in.OperationID, in.String())
+		resp.ErrCode = 1
+		resp.ErrMsg = err.Error()
+		return resp, nil
+	}
+	respSingleMsgFormat = singleMsgHandleByUser(SingleMsgFormat, in.UserID)
+	respGroupMsgFormat = groupMsgHandleByUser(GroupMsgFormat)
+	return &pbMsg.PullMessageResp{
+		ErrCode:       0,
+		ErrMsg:        "",
+		MaxSeq:        MaxSeq,
+		MinSeq:        MinSeq,
+		SingleUserMsg: respSingleMsgFormat,
+		GroupUserMsg:  respGroupMsgFormat,
+	}, nil
+}
+func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.GatherFormat {
+	var userid string
+	var respMsgFormat []*pbMsg.GatherFormat
+	m := make(map[string]MsgFormats)
+	//将消息以用户为维度聚集
+	for _, v := range allMsg {
+		if v.RecvID != ownerId {
+			userid = v.RecvID
+		} else {
+			userid = v.SendID
+		}
+		if value, ok := m[userid]; !ok {
+			var t MsgFormats
+			m[userid] = append(t, v)
+		} else {
+			m[userid] = append(value, v)
+		}
+	}
+	//形成pb格式返回
+	for user, msg := range m {
+		tempUserMsg := new(pbMsg.GatherFormat)
+		tempUserMsg.ID = user
+		tempUserMsg.List = msg
+		sort.Sort(msg)
+		respMsgFormat = append(respMsgFormat, tempUserMsg)
+	}
+	return respMsgFormat
+}
+func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
+	var respMsgFormat []*pbMsg.GatherFormat
+	m := make(map[string]MsgFormats)
+	//将消息以用户为维度聚集
+	for _, v := range allMsg {
+		//获得群ID
+		groupID := strings.Split(v.RecvID, " ")[1]
+		if value, ok := m[groupID]; !ok {
+			var t MsgFormats
+			m[groupID] = append(t, v)
+		} else {
+			m[groupID] = append(value, v)
+		}
+
+	}
+	//形成pb格式返回
+	for groupID, msg := range m {
+		tempUserMsg := new(pbMsg.GatherFormat)
+		tempUserMsg.ID = groupID
+		tempUserMsg.List = msg
+		sort.Sort(msg)
+		respMsgFormat = append(respMsgFormat, tempUserMsg)
+	}
+	return respMsgFormat
+}
+
+type MsgFormats []*pbMsg.MsgFormat
+
+// 实现sort.Interface接口取元素数量方法
+func (s MsgFormats) Len() int {
+	return len(s)
+}
+
+// 实现sort.Interface接口比较元素方法
+func (s MsgFormats) Less(i, j int) bool {
+	return s[i].SendTime < s[j].SendTime
+}
+
+// 实现sort.Interface接口交换元素方法
+func (s MsgFormats) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}

src/rpc/chat/chat/rpcChat.go

+package rpcChat
+
+import (
+	"Open_IM/src/common/config"
+	"Open_IM/src/common/kafka"
+	log2 "Open_IM/src/common/log"
+	pbChat "Open_IM/src/proto/chat"
+	"Open_IM/src/utils"
+	"github.com/skiffer-git/grpc-etcdv3/getcdv3"
+	"google.golang.org/grpc"
+	"net"
+	"strconv"
+	"strings"
+)
+
+type rpcChat struct {
+	rpcPort         int
+	rpcRegisterName string
+	etcdSchema      string
+	etcdAddr        []string
+	producer        *kafka.Producer
+}
+
+func NewRpcChatServer(port int) *rpcChat {
+	rc := rpcChat{
+		rpcPort:         port,
+		rpcRegisterName: config.Config.RpcRegisterName.OpenImOfflineMessageName,
+		etcdSchema:      config.Config.Etcd.EtcdSchema,
+		etcdAddr:        config.Config.Etcd.EtcdAddr,
+	}
+	rc.producer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
+	return &rc
+}
+
+func (rpc *rpcChat) Run() {
+	log2.Info("", "", "rpc get_token init...")
+
+	address := utils.ServerIP + ":" + strconv.Itoa(rpc.rpcPort)
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		log2.Error("", "", "listen network failed, err = %s, address = %s", err.Error(), address)
+		return
+	}
+	log2.Info("", "", "listen network success, address = %s", address)
+
+	//grpc server
+	srv := grpc.NewServer()
+	defer srv.GracefulStop()
+
+	//service registers with etcd
+
+	pbChat.RegisterChatServer(srv, rpc)
+	err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName, 10)
+	if err != nil {
+		log2.Error("", "", "register rpc get_token to etcd failed, err = %s", err.Error())
+		return
+	}
+
+	err = srv.Serve(listener)
+	if err != nil {
+		log2.Info("", "", "rpc get_token fail, err = %s", err.Error())
+		return
+	}
+	log2.Info("", "", "rpc get_token init success")
+}

src/rpc/chat/chat/send_msg.go

+package rpcChat
+
+import (
+	"Open_IM/src/common/log"
+	pbChat "Open_IM/src/proto/chat"
+	"Open_IM/src/utils"
+	"context"
+	"math/rand"
+	"strconv"
+	"time"
+)
+
+func (rpc *rpcChat) UserSendMsg(_ context.Context, pb *pbChat.UserSendMsgReq) (*pbChat.UserSendMsgResp, error) {
+
+	serverMsgID := GetMsgID(pb.SendID)
+	pbData := pbChat.WSToMsgSvrChatMsg{}
+	pbData.MsgFrom = pb.MsgFrom
+	pbData.SessionType = pb.SessionType
+	pbData.ContentType = pb.ContentType
+	pbData.Content = pb.Content
+	pbData.RecvID = pb.RecvID
+	pbData.ForceList = pb.ForceList
+	pbData.OfflineInfo = pb.OffLineInfo
+	pbData.Options = pb.Options
+	pbData.PlatformID = pb.PlatformID
+	pbData.SendID = pb.SendID
+	pbData.MsgID = serverMsgID
+	pbData.OperationID = pb.OperationID
+	pbData.Token = pb.Token
+	pbData.SendTime = utils.GetCurrentTimestampBySecond()
+	rpc.sendMsgToKafka(&pbData, pbData.RecvID)
+	rpc.sendMsgToKafka(&pbData, pbData.SendID)
+	replay := pbChat.UserSendMsgResp{}
+	replay.ReqIdentifier = pb.ReqIdentifier
+	replay.MsgIncr = pb.MsgIncr
+	replay.ClientMsgID = pb.ClientMsgID
+	replay.ServerMsgID = serverMsgID
+
+	return &replay, nil
+}
+func (rpc *rpcChat) sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string) {
+	pid, offset, err := rpc.producer.SendMessage(m, key)
+	if err != nil {
+		log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error())
+	}
+}
+func GetMsgID(sendID string) string {
+	t := time.Now().Format("2006-01-02 15:04:05")
+	return t + "-" + sendID + "-" + strconv.Itoa(rand.Int())
+}

src/rpc/chat/open_im_msg.go

+package main
+
+import (
+	rpcChat "Open_IM/src/rpc/chat/chat"
+	"Open_IM/src/utils"
+	"flag"
+)
+
+func main() {
+	rpcPort := flag.String("port", "", "rpc listening port")
+	flag.Parse()
+	rpcServer := rpcChat.NewRpcChatServer(utils.StringToInt(*rpcPort))
+	rpcServer.Run()
+}