Merge remote-tracking branch 'origin/modify' into v2.3.0release

# Conflicts:
#	config/config.yaml
#	pkg/common/constant/constant.go
#	pkg/proto/sdk_ws/ws.pb.go
#	pkg/proto/sdk_ws/ws.proto

internal/api/msg/extend_msg.go

@@ -0,0 +1,203 @@
+package msg
+
+import (
+	api "Open_IM/pkg/base_info"
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/token_verify"
+	"Open_IM/pkg/grpc-etcdv3/getcdv3"
+	rpc "Open_IM/pkg/proto/msg"
+	"Open_IM/pkg/utils"
+	"context"
+	"github.com/gin-gonic/gin"
+	"net/http"
+	"strings"
+)
+
+func SetMessageReactionExtensions(c *gin.Context) {
+	var (
+		req   api.SetMessageReactionExtensionsCallbackReq
+		resp  api.SetMessageReactionExtensionsCallbackResp
+		reqPb rpc.SetMessageReactionExtensionsReq
+	)
+
+	if err := c.BindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
+
+	if err := utils.CopyStructFields(&reqPb, &req); err != nil {
+		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
+	}
+	var ok bool
+	var errInfo string
+	ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
+	if !ok {
+		errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+
+	grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
+	if grpcConn == nil {
+		errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+	msgClient := rpc.NewMsgClient(grpcConn)
+	respPb, err := msgClient.SetMessageReactionExtensions(context.Background(), &reqPb)
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
+		return
+	}
+	resp.ErrCode = respPb.ErrCode
+	resp.ErrMsg = respPb.ErrMsg
+	resp.Data.ResultKeyValue = respPb.Result
+	resp.Data.MsgFirstModifyTime = reqPb.MsgFirstModifyTime
+	resp.Data.IsReact = reqPb.IsReact
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
+	c.JSON(http.StatusOK, resp)
+
+}
+
+func GetMessageListReactionExtensions(c *gin.Context) {
+	var (
+		req   api.GetMessageListReactionExtensionsReq
+		resp  api.GetMessageListReactionExtensionsResp
+		reqPb rpc.GetMessageListReactionExtensionsReq
+	)
+	if err := c.BindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
+
+	if err := utils.CopyStructFields(&reqPb, &req); err != nil {
+		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
+	}
+	var ok bool
+	var errInfo string
+	ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
+	if !ok {
+		errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+
+	grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
+	if grpcConn == nil {
+		errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+	msgClient := rpc.NewMsgClient(grpcConn)
+	respPb, err := msgClient.GetMessageListReactionExtensions(context.Background(), &reqPb)
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
+		return
+	}
+	resp.ErrCode = respPb.ErrCode
+	resp.ErrMsg = respPb.ErrMsg
+	resp.Data = respPb.SingleMessageResult
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
+	c.JSON(http.StatusOK, resp)
+}
+
+func AddMessageReactionExtensions(c *gin.Context) {
+	var (
+		req   api.AddMessageReactionExtensionsReq
+		resp  api.AddMessageReactionExtensionsResp
+		reqPb rpc.ModifyMessageReactionExtensionsReq
+	)
+	if err := c.BindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
+
+	if err := utils.CopyStructFields(&reqPb, &req); err != nil {
+		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
+	}
+	var ok bool
+	var errInfo string
+	ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
+	if !ok {
+		errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+
+	grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
+	if grpcConn == nil {
+		errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+	msgClient := rpc.NewMsgClient(grpcConn)
+	respPb, err := msgClient.AddMessageReactionExtensions(context.Background(), &reqPb)
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
+		return
+	}
+	resp.ErrCode = respPb.ErrCode
+	resp.ErrMsg = respPb.ErrMsg
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
+	c.JSON(http.StatusOK, resp)
+}
+
+func DeleteMessageReactionExtensions(c *gin.Context) {
+	var (
+		req   api.DeleteMessageReactionExtensionsReq
+		resp  api.DeleteMessageReactionExtensionsResp
+		reqPb rpc.DeleteMessageListReactionExtensionsReq
+	)
+	if err := c.BindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req:", req)
+
+	if err := utils.CopyStructFields(&reqPb, &req); err != nil {
+		log.NewDebug(req.OperationID, utils.GetSelfFuncName(), "CopyStructFields", err.Error())
+	}
+	var ok bool
+	var errInfo string
+	ok, reqPb.OpUserID, errInfo = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
+	if !ok {
+		errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+
+	grpcConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, req.OperationID)
+	if grpcConn == nil {
+		errMsg := req.OperationID + " getcdv3.GetDefaultConn == nil"
+		log.NewError(req.OperationID, errMsg)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
+		return
+	}
+	msgClient := rpc.NewMsgClient(grpcConn)
+	respPb, err := msgClient.DeleteMessageReactionExtensions(context.Background(), &reqPb)
+	if err != nil {
+		log.NewError(req.OperationID, utils.GetSelfFuncName(), "DelMsgList failed", err.Error(), reqPb)
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": constant.ErrServer.ErrCode, "errMsg": constant.ErrServer.ErrMsg + err.Error()})
+		return
+	}
+	resp.ErrCode = respPb.ErrCode
+	resp.ErrMsg = respPb.ErrMsg
+	resp.Data = respPb.Result
+	log.NewInfo(req.OperationID, utils.GetSelfFuncName(), resp)
+	c.JSON(http.StatusOK, resp)
+}

internal/msg_transfer/logic/init.go

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

internal/msg_transfer/logic/modify_msg_handler.go

@@ -0,0 +1,121 @@
+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 {
+		isReactionFromCache := utils.GetSwitchFromOptions(msgDataToMQ.MsgData.Options, constant.IsReactionFromCache)
+		if !isReactionFromCache {
+			continue
+		}
+		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
+}

internal/msg_transfer/logic/online_history_msg_handler.go

@@ -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)
@@ -83,11 +84,15 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 						if !(!isSenderSync && msgChannelValue.aggregationID == v.MsgData.SendID) {
 							notStoragePushMsgList = append(notStoragePushMsgList, v)
 						}
-
 					}
 
+					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:
@@ -107,6 +112,7 @@ func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) {
 						singleMsgSuccessCount += uint64(len(storageMsgList))
 						singleMsgSuccessCountMutex.Unlock()
 						och.SendMessageToMongoCH(msgChannelValue.aggregationID, triggerID, storageMsgList, lastSeq)
+
 						for _, v := range storageMsgList {
 							sendMessageToPushMQ(v, msgChannelValue.aggregationID)
 						}
@@ -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

internal/msg_transfer/logic/online_msg_to_mongo_handler.go

@@ -61,7 +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)
 			}
-			//if v.MsgData.ContentType == ? {}
 		}
 	}
 }

internal/push/push_interface.go

@@ -2,7 +2,7 @@ package push
 
 import "Open_IM/pkg/common/constant"
 
-var PushTerminal = []int{constant.IOSPlatformID, constant.AndroidPlatformID}
+var PushTerminal = []int{constant.IOSPlatformID, constant.AndroidPlatformID, constant.WebPlatformID}
 
 type OfflinePusher interface {
 	Push(userIDList []string, title, detailContent, operationID string, opts PushOpts) (resp string, err error)

internal/rpc/msg/callback.go

@@ -221,9 +221,3 @@ func callbackMsgModify(msg *pbChat.SendMsgReq) cbApi.CommonCallbackResp {
 	log.NewDebug(msg.OperationID, utils.GetSelfFuncName(), string(msg.MsgData.Content))
 	return callbackResp
 }
-
-func CallbackBeforeExtendMsgModify() cbApi.CommonCallbackResp {
-	callbackResp := cbApi.CommonCallbackResp{OperationID: ""}
-
-	return callbackResp
-}

internal/rpc/msg/extend_msg.go

@@ -1 +1,424 @@
 package msg
+
+import (
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/db"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/proto/msg"
+	"Open_IM/pkg/proto/sdk_ws"
+	"Open_IM/pkg/utils"
+	"context"
+	go_redis "github.com/go-redis/redis/v8"
+
+	"time"
+)
+
+func (rpc *rpcChat) SetMessageReactionExtensions(ctx context.Context, req *msg.SetMessageReactionExtensionsReq) (resp *msg.SetMessageReactionExtensionsResp, err error) {
+	log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
+	var rResp msg.SetMessageReactionExtensionsResp
+	rResp.ClientMsgID = req.ClientMsgID
+	rResp.MsgFirstModifyTime = req.MsgFirstModifyTime
+	callbackResp := callbackSetMessageReactionExtensions(req)
+	if callbackResp.ActionCode != constant.ActionAllow || callbackResp.ErrCode != 0 {
+		rResp.ErrCode = int32(callbackResp.ErrCode)
+		rResp.ErrMsg = callbackResp.ErrMsg
+		for _, value := range req.ReactionExtensionList {
+			temp := new(msg.KeyValueResp)
+			temp.KeyValue = value
+			temp.ErrMsg = callbackResp.ErrMsg
+			temp.ErrCode = 100
+			rResp.Result = append(rResp.Result, temp)
+		}
+		return &rResp, nil
+	}
+	//if ExternalExtension
+	if req.IsExternalExtensions {
+		var isHistory bool
+		if req.IsReact {
+			isHistory = false
+		} else {
+			isHistory = true
+		}
+		rResp.MsgFirstModifyTime = callbackResp.MsgFirstModifyTime
+		rResp.Result = callbackResp.ResultReactionExtensionList
+		ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, isHistory, false)
+		return &rResp, nil
+	}
+	for _, v := range callbackResp.ResultReactionExtensionList {
+		if v.ErrCode == 0 {
+			req.ReactionExtensionList[v.KeyValue.TypeKey] = v.KeyValue
+		} else {
+			delete(req.ReactionExtensionList, v.KeyValue.TypeKey)
+			rResp.Result = append(rResp.Result, v)
+		}
+	}
+	isExists, err := db.DB.JudgeMessageReactionEXISTS(req.ClientMsgID, req.SessionType)
+	if err != nil {
+		rResp.ErrCode = 100
+		rResp.ErrMsg = err.Error()
+		for _, value := range req.ReactionExtensionList {
+			temp := new(msg.KeyValueResp)
+			temp.KeyValue = value
+			temp.ErrMsg = err.Error()
+			temp.ErrCode = 100
+			rResp.Result = append(rResp.Result, temp)
+		}
+		return &rResp, nil
+	}
+
+	if !isExists {
+		if !req.IsReact {
+			log.Debug(req.OperationID, "redis handle firstly", req.String())
+			rResp.MsgFirstModifyTime = utils.GetCurrentTimestampByMill()
+			for k, v := range req.ReactionExtensionList {
+				err := rpc.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
+				if err != nil {
+					setKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, k, v)
+					continue
+				}
+				v.LatestUpdateTime = utils.GetCurrentTimestampByMill()
+				newerr := db.DB.SetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, k, utils.StructToJsonString(v))
+				if newerr != nil {
+					setKeyResultInfo(&rResp, 201, newerr.Error(), req.ClientMsgID, k, v)
+					continue
+				}
+				setKeyResultInfo(&rResp, 0, "", req.ClientMsgID, k, v)
+			}
+			rResp.IsReact = true
+			_, err := db.DB.SetMessageReactionExpire(req.ClientMsgID, req.SessionType, time.Duration(24*3)*time.Hour)
+			if err != nil {
+				log.Error(req.OperationID, "SetMessageReactionExpire err:", err.Error(), req.String())
+			}
+		} else {
+			err := rpc.dMessageLocker.LockGlobalMessage(req.ClientMsgID)
+			if err != nil {
+				rResp.ErrCode = 100
+				rResp.ErrMsg = err.Error()
+				for _, value := range req.ReactionExtensionList {
+					temp := new(msg.KeyValueResp)
+					temp.KeyValue = value
+					temp.ErrMsg = err.Error()
+					temp.ErrCode = 100
+					rResp.Result = append(rResp.Result, temp)
+				}
+				return &rResp, nil
+			}
+			mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime)
+			if err != nil {
+				rResp.ErrCode = 200
+				rResp.ErrMsg = err.Error()
+				for _, value := range req.ReactionExtensionList {
+					temp := new(msg.KeyValueResp)
+					temp.KeyValue = value
+					temp.ErrMsg = err.Error()
+					temp.ErrCode = 100
+					rResp.Result = append(rResp.Result, temp)
+				}
+				return &rResp, nil
+			}
+			setValue := make(map[string]*server_api_params.KeyValue)
+			for k, v := range req.ReactionExtensionList {
+
+				temp := new(server_api_params.KeyValue)
+				if vv, ok := mongoValue.ReactionExtensionList[k]; ok {
+					utils.CopyStructFields(temp, &vv)
+					if v.LatestUpdateTime != vv.LatestUpdateTime {
+						setKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, k, temp)
+						continue
+					}
+				}
+				temp.TypeKey = k
+				temp.Value = v.Value
+				temp.LatestUpdateTime = utils.GetCurrentTimestampByMill()
+				setValue[k] = temp
+			}
+			err = db.DB.InsertOrUpdateReactionExtendMsgSet(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime, setValue)
+			if err != nil {
+				for _, value := range setValue {
+					temp := new(msg.KeyValueResp)
+					temp.KeyValue = value
+					temp.ErrMsg = err.Error()
+					temp.ErrCode = 100
+					rResp.Result = append(rResp.Result, temp)
+				}
+			} else {
+				for _, value := range setValue {
+					temp := new(msg.KeyValueResp)
+					temp.KeyValue = value
+					rResp.Result = append(rResp.Result, temp)
+				}
+			}
+			lockErr := rpc.dMessageLocker.UnLockGlobalMessage(req.ClientMsgID)
+			if lockErr != nil {
+				log.Error(req.OperationID, "UnLockGlobalMessage err:", lockErr.Error())
+			}
+		}
+
+	} else {
+		log.Debug(req.OperationID, "redis handle secondly", req.String())
+
+		for k, v := range req.ReactionExtensionList {
+			err := rpc.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, k)
+			if err != nil {
+				setKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, k, v)
+				continue
+			}
+			redisValue, err := db.DB.GetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, k)
+			if err != nil && err != go_redis.Nil {
+				setKeyResultInfo(&rResp, 200, err.Error(), req.ClientMsgID, k, v)
+				continue
+			}
+			temp := new(server_api_params.KeyValue)
+			utils.JsonStringToStruct(redisValue, temp)
+			if v.LatestUpdateTime != temp.LatestUpdateTime {
+				setKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, k, temp)
+				continue
+			} else {
+				v.LatestUpdateTime = utils.GetCurrentTimestampByMill()
+				newerr := db.DB.SetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, k, utils.StructToJsonString(v))
+				if newerr != nil {
+					setKeyResultInfo(&rResp, 201, newerr.Error(), req.ClientMsgID, k, temp)
+					continue
+				}
+				setKeyResultInfo(&rResp, 0, "", req.ClientMsgID, k, v)
+			}
+
+		}
+	}
+	if !isExists {
+		if !req.IsReact {
+			ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, true, true)
+		} else {
+			ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, false)
+		}
+	} else {
+		ExtendMessageUpdatedNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, true)
+	}
+	log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", rResp.String())
+	return &rResp, nil
+
+}
+func setKeyResultInfo(r *msg.SetMessageReactionExtensionsResp, errCode int32, errMsg, clientMsgID, typeKey string, keyValue *server_api_params.KeyValue) {
+	temp := new(msg.KeyValueResp)
+	temp.KeyValue = keyValue
+	temp.ErrCode = errCode
+	temp.ErrMsg = errMsg
+	r.Result = append(r.Result, temp)
+	_ = db.DB.UnLockMessageTypeKey(clientMsgID, typeKey)
+}
+func setDeleteKeyResultInfo(r *msg.DeleteMessageListReactionExtensionsResp, errCode int32, errMsg, clientMsgID, typeKey string, keyValue *server_api_params.KeyValue) {
+	temp := new(msg.KeyValueResp)
+	temp.KeyValue = keyValue
+	temp.ErrCode = errCode
+	temp.ErrMsg = errMsg
+	r.Result = append(r.Result, temp)
+	_ = db.DB.UnLockMessageTypeKey(clientMsgID, typeKey)
+}
+
+func (rpc *rpcChat) GetMessageListReactionExtensions(ctx context.Context, req *msg.GetMessageListReactionExtensionsReq) (resp *msg.GetMessageListReactionExtensionsResp, err error) {
+	log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
+	var rResp msg.GetMessageListReactionExtensionsResp
+	for _, messageValue := range req.MessageReactionKeyList {
+		var oneMessage msg.SingleMessageExtensionResult
+		oneMessage.ClientMsgID = messageValue.ClientMsgID
+
+		isExists, err := db.DB.JudgeMessageReactionEXISTS(messageValue.ClientMsgID, req.SessionType)
+		if err != nil {
+			rResp.ErrCode = 100
+			rResp.ErrMsg = err.Error()
+			return &rResp, nil
+		}
+		if isExists {
+			redisValue, err := db.DB.GetOneMessageAllReactionList(messageValue.ClientMsgID, req.SessionType)
+			if err != nil {
+				oneMessage.ErrCode = 100
+				oneMessage.ErrMsg = err.Error()
+				rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
+				continue
+			}
+			keyMap := make(map[string]*server_api_params.KeyValue)
+
+			for k, v := range redisValue {
+				temp := new(server_api_params.KeyValue)
+				utils.JsonStringToStruct(v, temp)
+				keyMap[k] = temp
+			}
+			oneMessage.ReactionExtensionList = keyMap
+
+		} else {
+			mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, messageValue.ClientMsgID, messageValue.MsgFirstModifyTime)
+			if err != nil {
+				oneMessage.ErrCode = 100
+				oneMessage.ErrMsg = err.Error()
+				rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
+				continue
+			}
+			keyMap := make(map[string]*server_api_params.KeyValue)
+
+			for k, v := range mongoValue.ReactionExtensionList {
+				temp := new(server_api_params.KeyValue)
+				temp.TypeKey = v.TypeKey
+				temp.Value = v.Value
+				temp.LatestUpdateTime = v.LatestUpdateTime
+				keyMap[k] = temp
+			}
+			oneMessage.ReactionExtensionList = keyMap
+		}
+		rResp.SingleMessageResult = append(rResp.SingleMessageResult, &oneMessage)
+	}
+	log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", rResp.String())
+	return &rResp, nil
+
+}
+
+func (rpc *rpcChat) AddMessageReactionExtensions(ctx context.Context, req *msg.ModifyMessageReactionExtensionsReq) (resp *msg.ModifyMessageReactionExtensionsResp, err error) {
+	return
+}
+
+func (rpc *rpcChat) DeleteMessageReactionExtensions(ctx context.Context, req *msg.DeleteMessageListReactionExtensionsReq) (resp *msg.DeleteMessageListReactionExtensionsResp, err error) {
+	log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc args is:", req.String())
+	var rResp msg.DeleteMessageListReactionExtensionsResp
+	callbackResp := callbackDeleteMessageReactionExtensions(req)
+	if callbackResp.ActionCode != constant.ActionAllow || callbackResp.ErrCode != 0 {
+		rResp.ErrCode = int32(callbackResp.ErrCode)
+		rResp.ErrMsg = callbackResp.ErrMsg
+		for _, value := range req.ReactionExtensionList {
+			temp := new(msg.KeyValueResp)
+			temp.KeyValue = value
+			temp.ErrMsg = callbackResp.ErrMsg
+			temp.ErrCode = 100
+			rResp.Result = append(rResp.Result, temp)
+		}
+		return &rResp, nil
+	}
+	//if ExternalExtension
+	if req.IsExternalExtensions {
+		rResp.Result = callbackResp.ResultReactionExtensionList
+		ExtendMessageDeleteNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, false)
+		return &rResp, nil
+
+	}
+	for _, v := range callbackResp.ResultReactionExtensionList {
+		if v.ErrCode != 0 {
+			func(req *[]*server_api_params.KeyValue, typeKey string) {
+				for i := 0; i < len(*req); i++ {
+					if (*req)[i].TypeKey == typeKey {
+						*req = append((*req)[:i], (*req)[i+1:]...)
+					}
+				}
+			}(&req.ReactionExtensionList, v.KeyValue.TypeKey)
+			rResp.Result = append(rResp.Result, v)
+		}
+	}
+	isExists, err := db.DB.JudgeMessageReactionEXISTS(req.ClientMsgID, req.SessionType)
+	if err != nil {
+		rResp.ErrCode = 100
+		rResp.ErrMsg = err.Error()
+		for _, value := range req.ReactionExtensionList {
+			temp := new(msg.KeyValueResp)
+			temp.KeyValue = value
+			temp.ErrMsg = err.Error()
+			temp.ErrCode = 100
+			rResp.Result = append(rResp.Result, temp)
+		}
+		return &rResp, nil
+	}
+
+	if isExists {
+		log.Debug(req.OperationID, "redis handle this delete", req.String())
+		for _, v := range req.ReactionExtensionList {
+			err := rpc.dMessageLocker.LockMessageTypeKey(req.ClientMsgID, v.TypeKey)
+			if err != nil {
+				setDeleteKeyResultInfo(&rResp, 100, err.Error(), req.ClientMsgID, v.TypeKey, v)
+				continue
+			}
+
+			redisValue, err := db.DB.GetMessageTypeKeyValue(req.ClientMsgID, req.SessionType, v.TypeKey)
+			if err != nil && err != go_redis.Nil {
+				setDeleteKeyResultInfo(&rResp, 200, err.Error(), req.ClientMsgID, v.TypeKey, v)
+				continue
+			}
+			temp := new(server_api_params.KeyValue)
+			utils.JsonStringToStruct(redisValue, temp)
+			if v.LatestUpdateTime != temp.LatestUpdateTime {
+				setDeleteKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, v.TypeKey, temp)
+				continue
+			} else {
+				newErr := db.DB.DeleteOneMessageKey(req.ClientMsgID, req.SessionType, v.TypeKey)
+				if newErr != nil {
+					setDeleteKeyResultInfo(&rResp, 201, newErr.Error(), req.ClientMsgID, v.TypeKey, temp)
+					continue
+				}
+				setDeleteKeyResultInfo(&rResp, 0, "", req.ClientMsgID, v.TypeKey, v)
+			}
+		}
+	} else {
+		err := rpc.dMessageLocker.LockGlobalMessage(req.ClientMsgID)
+		if err != nil {
+			rResp.ErrCode = 100
+			rResp.ErrMsg = err.Error()
+			for _, value := range req.ReactionExtensionList {
+				temp := new(msg.KeyValueResp)
+				temp.KeyValue = value
+				temp.ErrMsg = err.Error()
+				temp.ErrCode = 100
+				rResp.Result = append(rResp.Result, temp)
+			}
+			return &rResp, nil
+		}
+		mongoValue, err := db.DB.GetExtendMsg(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime)
+		if err != nil {
+			rResp.ErrCode = 200
+			rResp.ErrMsg = err.Error()
+			for _, value := range req.ReactionExtensionList {
+				temp := new(msg.KeyValueResp)
+				temp.KeyValue = value
+				temp.ErrMsg = err.Error()
+				temp.ErrCode = 100
+				rResp.Result = append(rResp.Result, temp)
+			}
+			return &rResp, nil
+		}
+		setValue := make(map[string]*server_api_params.KeyValue)
+		for _, v := range req.ReactionExtensionList {
+
+			temp := new(server_api_params.KeyValue)
+			if vv, ok := mongoValue.ReactionExtensionList[v.TypeKey]; ok {
+				utils.CopyStructFields(temp, &vv)
+				if v.LatestUpdateTime != vv.LatestUpdateTime {
+					setDeleteKeyResultInfo(&rResp, 300, "message have update", req.ClientMsgID, v.TypeKey, temp)
+					continue
+				}
+			} else {
+				setDeleteKeyResultInfo(&rResp, 400, "key not in", req.ClientMsgID, v.TypeKey, v)
+				continue
+			}
+			temp.TypeKey = v.TypeKey
+			setValue[v.TypeKey] = temp
+		}
+		err = db.DB.DeleteReactionExtendMsgSet(req.SourceID, req.SessionType, req.ClientMsgID, req.MsgFirstModifyTime, setValue)
+		if err != nil {
+			for _, value := range setValue {
+				temp := new(msg.KeyValueResp)
+				temp.KeyValue = value
+				temp.ErrMsg = err.Error()
+				temp.ErrCode = 100
+				rResp.Result = append(rResp.Result, temp)
+			}
+		} else {
+			for _, value := range setValue {
+				temp := new(msg.KeyValueResp)
+				temp.KeyValue = value
+				rResp.Result = append(rResp.Result, temp)
+			}
+		}
+		lockErr := rpc.dMessageLocker.UnLockGlobalMessage(req.ClientMsgID)
+		if lockErr != nil {
+			log.Error(req.OperationID, "UnLockGlobalMessage err:", lockErr.Error())
+		}
+
+	}
+	ExtendMessageDeleteNotification(req.OperationID, req.OpUserID, req.SourceID, req.SessionType, req, &rResp, false, isExists)
+	log.Debug(req.OperationID, utils.GetSelfFuncName(), "rpc return is:", rResp.String())
+	return &rResp, nil
+}

internal/rpc/msg/extend_msg.notification.go

@@ -0,0 +1,106 @@
+package msg
+
+import (
+	"Open_IM/pkg/base_info"
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/grpc-etcdv3/getcdv3"
+	"Open_IM/pkg/proto/msg"
+	open_im_sdk "Open_IM/pkg/proto/sdk_ws"
+	"Open_IM/pkg/utils"
+	"context"
+	"strings"
+)
+
+func ExtendMessageUpdatedNotification(operationID, sendID string, sourceID string, sessionType int32,
+	req *msg.SetMessageReactionExtensionsReq, resp *msg.SetMessageReactionExtensionsResp, isHistory bool, isReactionFromCache bool) {
+	var m base_info.ReactionMessageModifierNotification
+	m.SourceID = req.SourceID
+	m.OpUserID = req.OpUserID
+	m.SessionType = req.SessionType
+	keyMap := make(map[string]*open_im_sdk.KeyValue)
+	for _, valueResp := range resp.Result {
+		if valueResp.ErrCode == 0 {
+			keyMap[valueResp.KeyValue.TypeKey] = valueResp.KeyValue
+		}
+	}
+	if len(keyMap) == 0 {
+		log.NewWarn(operationID, "all key set failed can not send notification", *req)
+		return
+	}
+	m.SuccessReactionExtensionList = keyMap
+	m.ClientMsgID = req.ClientMsgID
+	m.IsReact = resp.IsReact
+	m.IsExternalExtensions = req.IsExternalExtensions
+	m.MsgFirstModifyTime = resp.MsgFirstModifyTime
+	messageReactionSender(operationID, sendID, sourceID, sessionType, constant.ReactionMessageModifier, utils.StructToJsonString(m), isHistory, isReactionFromCache)
+}
+func ExtendMessageDeleteNotification(operationID, sendID string, sourceID string, sessionType int32,
+	req *msg.DeleteMessageListReactionExtensionsReq, resp *msg.DeleteMessageListReactionExtensionsResp, isHistory bool, isReactionFromCache bool) {
+	var m base_info.ReactionMessageDeleteNotification
+	m.SourceID = req.SourceID
+	m.OpUserID = req.OpUserID
+	m.SessionType = req.SessionType
+	keyMap := make(map[string]*open_im_sdk.KeyValue)
+	for _, valueResp := range resp.Result {
+		if valueResp.ErrCode == 0 {
+			keyMap[valueResp.KeyValue.TypeKey] = valueResp.KeyValue
+		}
+	}
+	if len(keyMap) == 0 {
+		log.NewWarn(operationID, "all key set failed can not send notification", *req)
+		return
+	}
+	m.SuccessReactionExtensionList = keyMap
+	m.ClientMsgID = req.ClientMsgID
+	m.MsgFirstModifyTime = req.MsgFirstModifyTime
+
+	messageReactionSender(operationID, sendID, sourceID, sessionType, constant.ReactionMessageDeleter, utils.StructToJsonString(m), isHistory, isReactionFromCache)
+}
+func messageReactionSender(operationID, sendID string, sourceID string, sessionType, contentType int32, content string, isHistory bool, isReactionFromCache bool) {
+	options := make(map[string]bool, 5)
+	utils.SetSwitchFromOptions(options, constant.IsOfflinePush, false)
+	utils.SetSwitchFromOptions(options, constant.IsConversationUpdate, false)
+	utils.SetSwitchFromOptions(options, constant.IsSenderConversationUpdate, false)
+	utils.SetSwitchFromOptions(options, constant.IsUnreadCount, false)
+	utils.SetSwitchFromOptions(options, constant.IsReactionFromCache, isReactionFromCache)
+	if !isHistory {
+		utils.SetSwitchFromOptions(options, constant.IsHistory, false)
+		utils.SetSwitchFromOptions(options, constant.IsPersistent, false)
+	}
+	pbData := msg.SendMsgReq{
+		OperationID: operationID,
+		MsgData: &open_im_sdk.MsgData{
+			SendID:      sendID,
+			ClientMsgID: utils.GetMsgID(sendID),
+			SessionType: sessionType,
+			MsgFrom:     constant.SysMsgType,
+			ContentType: contentType,
+			Content:     []byte(content),
+			//	ForceList:        params.ForceList,
+			CreateTime: utils.GetCurrentTimestampByMill(),
+			Options:    options,
+		},
+	}
+	switch sessionType {
+	case constant.SingleChatType, constant.NotificationChatType:
+		pbData.MsgData.RecvID = sourceID
+	case constant.GroupChatType, constant.SuperGroupChatType:
+		pbData.MsgData.GroupID = sourceID
+	}
+	etcdConn := getcdv3.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImMsgName, operationID)
+	if etcdConn == nil {
+		errMsg := operationID + "getcdv3.GetDefaultConn == nil"
+		log.NewError(operationID, errMsg)
+		return
+	}
+	client := msg.NewMsgClient(etcdConn)
+	reply, err := client.SendMsg(context.Background(), &pbData)
+	if err != nil {
+		log.NewError(operationID, "SendMsg rpc failed, ", pbData.String(), err.Error())
+	} else if reply.ErrCode != 0 {
+		log.NewError(operationID, "SendMsg rpc failed, ", pbData.String(), reply.ErrCode, reply.ErrMsg)
+	}
+
+}

internal/rpc/msg/extend_msg_callback.go

@@ -0,0 +1,60 @@
+package msg
+
+import (
+	cbApi "Open_IM/pkg/call_back_struct"
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/http"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/proto/msg"
+	"Open_IM/pkg/utils"
+	http2 "net/http"
+)
+
+func callbackSetMessageReactionExtensions(setReq *msg.SetMessageReactionExtensionsReq) *cbApi.CallbackBeforeSetMessageReactionExtResp {
+	callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
+	log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
+	req := cbApi.CallbackBeforeSetMessageReactionExtReq{
+		OperationID:           setReq.OperationID,
+		CallbackCommand:       constant.CallbackBeforeSetMessageReactionExtensionCommand,
+		SourceID:              setReq.SourceID,
+		OpUserID:              setReq.OpUserID,
+		SessionType:           setReq.SessionType,
+		ReactionExtensionList: setReq.ReactionExtensionList,
+		ClientMsgID:           setReq.ClientMsgID,
+		IsReact:               setReq.IsReact,
+		IsExternalExtensions:  setReq.IsExternalExtensions,
+		MsgFirstModifyTime:    setReq.MsgFirstModifyTime,
+	}
+	resp := &cbApi.CallbackBeforeSetMessageReactionExtResp{CommonCallbackResp: &callbackResp}
+	defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp)
+	if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeSetMessageReactionExtensionCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
+		callbackResp.ErrCode = http2.StatusInternalServerError
+		callbackResp.ErrMsg = err.Error()
+	}
+	return resp
+
+}
+
+func callbackDeleteMessageReactionExtensions(setReq *msg.DeleteMessageListReactionExtensionsReq) *cbApi.CallbackDeleteMessageReactionExtResp {
+	callbackResp := cbApi.CommonCallbackResp{OperationID: setReq.OperationID}
+	log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), setReq.String())
+	req := cbApi.CallbackDeleteMessageReactionExtReq{
+		OperationID:           setReq.OperationID,
+		CallbackCommand:       constant.CallbackBeforeDeleteMessageReactionExtensionsCommand,
+		SourceID:              setReq.SourceID,
+		OpUserID:              setReq.OpUserID,
+		SessionType:           setReq.SessionType,
+		ReactionExtensionList: setReq.ReactionExtensionList,
+		ClientMsgID:           setReq.ClientMsgID,
+		IsExternalExtensions:  setReq.IsExternalExtensions,
+		MsgFirstModifyTime:    setReq.MsgFirstModifyTime,
+	}
+	resp := &cbApi.CallbackDeleteMessageReactionExtResp{CommonCallbackResp: &callbackResp}
+	defer log.NewDebug(setReq.OperationID, utils.GetSelfFuncName(), req, *resp)
+	if err := http.CallBackPostReturn(config.Config.Callback.CallbackUrl, constant.CallbackBeforeDeleteMessageReactionExtensionsCommand, req, resp, config.Config.Callback.CallbackAfterSendGroupMsg.CallbackTimeOut); err != nil {
+		callbackResp.ErrCode = http2.StatusInternalServerError
+		callbackResp.ErrMsg = err.Error()
+	}
+	return resp
+}

internal/rpc/msg/lock.go

@@ -0,0 +1,52 @@
+package msg
+
+import (
+	"Open_IM/pkg/common/db"
+	"time"
+)
+
+const GlOBLLOCK = "GLOBAL_LOCK"
+
+type MessageLocker interface {
+	LockMessageTypeKey(clientMsgID, typeKey string) (err error)
+	UnLockMessageTypeKey(clientMsgID string, typeKey string) error
+	LockGlobalMessage(clientMsgID string) (err error)
+	UnLockGlobalMessage(clientMsgID string) (err error)
+}
+type LockerMessage struct{}
+
+func NewLockerMessage() *LockerMessage {
+	return &LockerMessage{}
+}
+func (l *LockerMessage) LockMessageTypeKey(clientMsgID, typeKey string) (err error) {
+	for i := 0; i < 3; i++ {
+		err = db.DB.LockMessageTypeKey(clientMsgID, typeKey)
+		if err != nil {
+			time.Sleep(time.Millisecond * 100)
+			continue
+		} else {
+			break
+		}
+	}
+	return err
+
+}
+func (l *LockerMessage) LockGlobalMessage(clientMsgID string) (err error) {
+	for i := 0; i < 3; i++ {
+		err = db.DB.LockMessageTypeKey(clientMsgID, GlOBLLOCK)
+		if err != nil {
+			time.Sleep(time.Millisecond * 100)
+			continue
+		} else {
+			break
+		}
+	}
+	return err
+
+}
+func (l *LockerMessage) UnLockMessageTypeKey(clientMsgID string, typeKey string) error {
+	return db.DB.UnLockMessageTypeKey(clientMsgID, typeKey)
+}
+func (l *LockerMessage) UnLockGlobalMessage(clientMsgID string) error {
+	return db.DB.UnLockMessageTypeKey(clientMsgID, GlOBLLOCK)
+}

internal/rpc/msg/rpcChat.go

@@ -31,7 +31,8 @@ type rpcChat struct {
 	etcdAddr        []string
 	messageWriter   MessageWriter
 	//offlineProducer *kafka.Producer
-	delMsgCh chan deleteMsg
+	delMsgCh       chan deleteMsg
+	dMessageLocker MessageLocker
 }
 
 type deleteMsg struct {
@@ -48,6 +49,7 @@ func NewRpcChatServer(port int) *rpcChat {
 		rpcRegisterName: config.Config.RpcRegisterName.OpenImMsgName,
 		etcdSchema:      config.Config.Etcd.EtcdSchema,
 		etcdAddr:        config.Config.Etcd.EtcdAddr,
+		dMessageLocker:  NewLockerMessage(),
 	}
 	rc.messageWriter = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
 	//rc.offlineProducer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschatOffline.Addr, config.Config.Kafka.Ws2mschatOffline.Topic)