send msg file modify

internal/api/chat/send_msg.go

@@ -34,8 +34,8 @@ type paramsUserSendMsg struct {
 	}
 }
 
-func newUserSendMsgReq(token string, params *paramsUserSendMsg) *pbChat.UserSendMsgReq {
-	pbData := pbChat.UserSendMsgReq{
+func newUserSendMsgReq(token string, params *paramsUserSendMsg) *pbChat.SendMsgReq {
+	pbData := pbChat.SendMsgReq{
 		ReqIdentifier:  params.ReqIdentifier,
 		Token:          token,
 		SendID:         params.SendID,
@@ -70,20 +70,20 @@ func UserSendMsg(c *gin.Context) {
 	log.InfoByKv("api call success to sendMsgReq", params.OperationID, "Parameters", params)
 
 	pbData := newUserSendMsgReq(token, &params)
-	log.Info("", "", "api UserSendMsg call start..., [data: %s]", pbData.String())
+	log.Info("", "", "api SendMsg call start..., [data: %s]", pbData.String())
 
 	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
 	client := pbChat.NewChatClient(etcdConn)
 
-	log.Info("", "", "api UserSendMsg call, api call rpc...")
+	log.Info("", "", "api SendMsg call, api call rpc...")
 
-	reply, err := client.UserSendMsg(context.Background(), pbData)
+	reply, err := client.SendMsg(context.Background(), pbData)
 	if err != nil {
-		log.NewError(params.OperationID, "UserSendMsg rpc failed, ", params, err.Error())
-		c.JSON(http.StatusBadRequest, gin.H{"errCode": 401, "errMsg": "UserSendMsg rpc failed, " + err.Error()})
+		log.NewError(params.OperationID, "SendMsg rpc failed, ", params, err.Error())
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 401, "errMsg": "SendMsg rpc failed, " + err.Error()})
 		return
 	}
-	log.Info("", "", "api UserSendMsg call end..., [data: %s] [reply: %s]", pbData.String(), reply.String())
+	log.Info("", "", "api SendMsg call end..., [data: %s] [reply: %s]", pbData.String(), reply.String())
 
 	c.JSON(http.StatusOK, gin.H{
 		"errCode":       reply.ErrCode,

internal/msg_gateway/gate/logic.go

@@ -6,7 +6,7 @@ import (
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	pbChat "Open_IM/pkg/proto/chat"
-	open_im_sdk "Open_IM/pkg/proto/sdk_ws"
+	sdk_ws "Open_IM/pkg/proto/sdk_ws"
 	"Open_IM/pkg/utils"
 	"bytes"
 	"context"
@@ -86,7 +86,7 @@ func (ws *WServer) getSeqReq(conn *UserConn, m *Req) {
 	}
 }
 func (ws *WServer) getSeqResp(conn *UserConn, m *Req, pb *pbChat.GetMaxAndMinSeqResp) {
-	var mReplyData open_im_sdk.GetMaxAndMinSeqResp
+	var mReplyData sdk_ws.GetMaxAndMinSeqResp
 	mReplyData.MaxSeq = pb.GetMaxSeq()
 	mReplyData.MinSeq = pb.GetMinSeq()
 	b, _ := proto.Marshal(&mReplyData)
@@ -131,7 +131,7 @@ func (ws *WServer) pullMsgReq(conn *UserConn, m *Req) {
 }
 func (ws *WServer) pullMsgResp(conn *UserConn, m *Req, pb *pbChat.PullMessageResp) {
 	log.NewInfo(m.OperationID, "pullMsgResp come  here ", pb.String())
-	var mReplyData open_im_sdk.PullMessageBySeqListResp
+	var mReplyData sdk_ws.PullMessageBySeqListResp
 	a, err := json.Marshal(pb.SingleUserMsg)
 	if err != nil {
 		log.NewError(m.OperationID, "GetSingleUserMsg,json marshal,err", err.Error())
@@ -170,10 +170,10 @@ func (ws *WServer) pullMsgBySeqListReq(conn *UserConn, m *Req) {
 	log.NewInfo(m.OperationID, "Ws call success to pullMsgBySeqListReq start", m.SendID, m.ReqIdentifier, m.MsgIncr)
 	nReply := new(pbChat.PullMessageResp)
 	isPass, errCode, errMsg, data := ws.argsValidate(m, constant.WSPullMsgBySeqList)
-	log.NewInfo(m.OperationID, "Ws call success to pullMsgBySeqListReq middle", m.SendID, m.ReqIdentifier, m.MsgIncr, data.(open_im_sdk.PullMessageBySeqListReq).SeqList)
+	log.NewInfo(m.OperationID, "Ws call success to pullMsgBySeqListReq middle", m.SendID, m.ReqIdentifier, m.MsgIncr, data.(sdk_ws.PullMessageBySeqListReq).SeqList)
 	if isPass {
 		pbData := pbChat.PullMessageBySeqListReq{}
-		pbData.SeqList = data.(open_im_sdk.PullMessageBySeqListReq).SeqList
+		pbData.SeqList = data.(sdk_ws.PullMessageBySeqListReq).SeqList
 		pbData.UserID = m.SendID
 		pbData.OperationID = m.OperationID
 		grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
@@ -196,32 +196,20 @@ func (ws *WServer) pullMsgBySeqListReq(conn *UserConn, m *Req) {
 }
 func (ws *WServer) sendMsgReq(conn *UserConn, m *Req, sendTime int64) {
 	log.NewInfo(m.OperationID, "Ws call success to sendMsgReq start", m.MsgIncr, m.ReqIdentifier, m.SendID, sendTime)
-	nReply := new(pbChat.UserSendMsgResp)
+	nReply := new(pbChat.SendMsgResp)
 	isPass, errCode, errMsg, pData := ws.argsValidate(m, constant.WSSendMsg)
 	if isPass {
-		data := pData.(open_im_sdk.UserSendMsgReq)
-		pbData := pbChat.UserSendMsgReq{
-			ReqIdentifier:  m.ReqIdentifier,
-			Token:          m.Token,
-			SendID:         m.SendID,
-			OperationID:    m.OperationID,
-			PlatformID:     data.PlatformID,
-			SessionType:    data.SessionType,
-			MsgFrom:        data.MsgFrom,
-			ContentType:    data.ContentType,
-			RecvID:         data.RecvID,
-			ForceList:      data.ForceList,
-			SenderNickName: data.SenderNickName,
-			SenderFaceURL:  data.SenderFaceURL,
-			Content:        data.Content,
-			Options:        utils.MapIntToJsonString(data.Options),
-			ClientMsgID:    data.ClientMsgID,
-			SendTime:       sendTime,
+		data := pData.(sdk_ws.MsgData)
+		data.SendTime = sendTime
+		pbData := pbChat.SendMsgReq{
+			Token:       m.Token,
+			OperationID: m.OperationID,
+			MsgData:     &data,
 		}
 		log.NewInfo(m.OperationID, "Ws call success to sendMsgReq middle", m.ReqIdentifier, m.SendID, m.MsgIncr, data)
 		etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
 		client := pbChat.NewChatClient(etcdConn)
-		reply, err := client.UserSendMsg(context.Background(), &pbData)
+		reply, err := client.SendMsg(context.Background(), &pbData)
 		if err != nil {
 			log.NewError(pbData.OperationID, "UserSendMsg err", err.Error())
 			nReply.ErrCode = 200
@@ -239,10 +227,10 @@ func (ws *WServer) sendMsgReq(conn *UserConn, m *Req, sendTime int64) {
 	}
 
 }
-func (ws *WServer) sendMsgResp(conn *UserConn, m *Req, pb *pbChat.UserSendMsgResp, sendTime int64) {
+func (ws *WServer) sendMsgResp(conn *UserConn, m *Req, pb *pbChat.SendMsgResp, sendTime int64) {
 	// := make(map[string]interface{})
 
-	var mReplyData open_im_sdk.UserSendMsgResp
+	var mReplyData sdk_ws.UserSendMsgResp
 	mReplyData.ClientMsgID = pb.GetClientMsgID()
 	mReplyData.ServerMsgID = pb.GetServerMsgID()
 	mReplyData.SendTime = sendTime
@@ -263,12 +251,14 @@ func (ws *WServer) sendMsg(conn *UserConn, mReply interface{}) {
 	enc := gob.NewEncoder(&b)
 	err := enc.Encode(mReply)
 	if err != nil {
-		log.NewError(mReply.(Resp).OperationID, mReply.(Resp).ReqIdentifier, mReply.(Resp).ErrCode, mReply.(Resp).ErrMsg, "Encode Msg error", conn.RemoteAddr().String(), ws.getUserUid(conn), err.Error())
+		uid, platform := ws.getUserUid(conn)
+		log.NewError(mReply.(Resp).OperationID, mReply.(Resp).ReqIdentifier, mReply.(Resp).ErrCode, mReply.(Resp).ErrMsg, "Encode Msg error", conn.RemoteAddr().String(), uid, platform, err.Error())
 		return
 	}
 	err = ws.writeMsg(conn, websocket.BinaryMessage, b.Bytes())
 	if err != nil {
-		log.NewError(mReply.(Resp).OperationID, mReply.(Resp).ReqIdentifier, mReply.(Resp).ErrCode, mReply.(Resp).ErrMsg, "WS WriteMsg error", conn.RemoteAddr().String(), ws.getUserUid(conn), err.Error())
+		uid, platform := ws.getUserUid(conn)
+		log.NewError(mReply.(Resp).OperationID, mReply.(Resp).ReqIdentifier, mReply.(Resp).ErrCode, mReply.(Resp).ErrMsg, "WS WriteMsg error", conn.RemoteAddr().String(), uid, platform, err.Error())
 	}
 }
 func (ws *WServer) sendErrMsg(conn *UserConn, errCode int32, errMsg string, reqIdentifier int32, msgIncr string, operationID string) {

internal/msg_gateway/gate/rpc_server.go

@@ -6,7 +6,6 @@ import (
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	pbRelay "Open_IM/pkg/proto/relay"
-	open_im_sdk "Open_IM/pkg/proto/sdk_ws"
 	"Open_IM/pkg/utils"
 	"bytes"
 	"context"
@@ -55,26 +54,10 @@ func (r *RPCServer) run() {
 		return
 	}
 }
-func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbRelay.MsgToUserResp, error) {
+func (r *RPCServer) OnlinePushMsg(_ context.Context, in *pbRelay.OnlinePushMsgReq) (*pbRelay.OnlinePushMsgResp, error) {
 	log.InfoByKv("PushMsgToUser is arriving", in.OperationID, "args", in.String())
 	var resp []*pbRelay.SingleMsgToUser
-	var RecvID string
-	msg := open_im_sdk.MsgData{
-		SendID:           in.SendID,
-		RecvID:           in.RecvID,
-		MsgFrom:          in.MsgFrom,
-		ContentType:      in.ContentType,
-		SessionType:      in.SessionType,
-		SenderNickName:   in.SenderNickName,
-		SenderFaceURL:    in.SenderFaceURL,
-		ClientMsgID:      in.ClientMsgID,
-		ServerMsgID:      in.ServerMsgID,
-		Content:          in.Content,
-		Seq:              in.RecvSeq,
-		SendTime:         in.SendTime,
-		SenderPlatformID: in.PlatformID,
-	}
-	msgBytes, _ := proto.Marshal(&msg)
+	msgBytes, _ := proto.Marshal(in.MsgData)
 	mReply := Resp{
 		ReqIdentifier: constant.WSPushMsg,
 		OperationID:   in.OperationID,
@@ -86,65 +69,52 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
 	if err != nil {
 		log.NewError(in.OperationID, "data encode err", err.Error())
 	}
-	switch in.GetSessionType() {
-	case constant.SingleChatType:
-		RecvID = in.GetRecvID()
-	case constant.GroupChatType:
-		RecvID = strings.Split(in.GetRecvID(), " ")[0]
-	}
 	var tag bool
-	var UIDAndPID []string
-	userIDList := genUidPlatformArray(RecvID)
-	for _, v := range userIDList {
-		UIDAndPID = strings.Split(v, " ")
-		if conn := ws.getUserConn(v); conn != nil {
+	recvID := in.MsgData.RecvID
+	platformList := genPlatformArray()
+	for _, v := range platformList {
+		if conn := ws.getUserConn(recvID, v); conn != nil {
 			tag = true
-			resultCode := sendMsgToUser(conn, replyBytes.Bytes(), in, UIDAndPID[1], UIDAndPID[0])
+			resultCode := sendMsgToUser(conn, replyBytes.Bytes(), in, v, recvID)
 			temp := &pbRelay.SingleMsgToUser{
 				ResultCode:     resultCode,
-				RecvID:         UIDAndPID[0],
-				RecvPlatFormID: constant.PlatformNameToID(UIDAndPID[1]),
+				RecvID:         recvID,
+				RecvPlatFormID: constant.PlatformNameToID(v),
 			}
 			resp = append(resp, temp)
 		} else {
 			temp := &pbRelay.SingleMsgToUser{
 				ResultCode:     -1,
-				RecvID:         UIDAndPID[0],
-				RecvPlatFormID: constant.PlatformNameToID(UIDAndPID[1]),
+				RecvID:         recvID,
+				RecvPlatFormID: constant.PlatformNameToID(v),
 			}
 			resp = append(resp, temp)
 		}
 	}
 	//Single chat sender synchronization message
-	if in.GetSessionType() == constant.SingleChatType && in.ContentType <= constant.Quote && in.ContentType != constant.Typing && in.ContentType != constant.HasReadReceipt {
-		userIDList = genUidPlatformArray(in.SendID)
-		for _, v := range userIDList {
-			UIDAndPID = strings.Split(v, " ")
-			if conn := ws.getUserConn(v); conn != nil {
-				_ = sendMsgToUser(conn, replyBytes.Bytes(), in, UIDAndPID[1], UIDAndPID[0])
-			}
+	if in.MsgData.GetSessionType() == constant.SingleChatType {
+		for k, v := range ws.getSingleUserAllConn(recvID) {
+			_ = sendMsgToUser(v, replyBytes.Bytes(), in, k, recvID)
 		}
 	}
 	if !tag {
 		log.NewError(in.OperationID, "push err ,no matched ws conn not in map", in.String())
 	}
-	return &pbRelay.MsgToUserResp{
+	return &pbRelay.OnlinePushMsgResp{
 		Resp: resp,
 	}, nil
 }
 func (r *RPCServer) GetUsersOnlineStatus(_ context.Context, req *pbRelay.GetUsersOnlineStatusReq) (*pbRelay.GetUsersOnlineStatusResp, error) {
 	log.NewDebug(req.OperationID, "rpc GetUsersOnlineStatus arrived server", req.String())
-	var UIDAndPID []string
 	var resp pbRelay.GetUsersOnlineStatusResp
-	for _, v1 := range req.UserIDList {
-		userIDList := genUidPlatformArray(v1)
+	for _, userID := range req.UserIDList {
+		platformList := genPlatformArray()
 		temp := new(pbRelay.GetUsersOnlineStatusResp_SuccessResult)
-		temp.UserID = v1
-		for _, v2 := range userIDList {
-			UIDAndPID = strings.Split(v2, " ")
-			if conn := ws.getUserConn(v2); conn != nil {
+		temp.UserID = userID
+		for _, platform := range platformList {
+			if conn := ws.getUserConn(userID, platform); conn != nil {
 				ps := new(pbRelay.GetUsersOnlineStatusResp_SuccessDetail)
-				ps.Platform = UIDAndPID[1]
+				ps.Platform = platform
 				ps.Status = constant.OnlineStatus
 				temp.Status = constant.OnlineStatus
 				temp.DetailPlatformStatus = append(temp.DetailPlatformStatus, ps)
@@ -157,11 +127,11 @@ func (r *RPCServer) GetUsersOnlineStatus(_ context.Context, req *pbRelay.GetUser
 	}
 	return &resp, nil
 }
-func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.MsgToUserReq, RecvPlatForm, RecvID string) (ResultCode int64) {
+func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.OnlinePushMsgReq, RecvPlatForm, RecvID string) (ResultCode int64) {
 	err := ws.writeMsg(conn, websocket.BinaryMessage, bMsg)
 	if err != nil {
 		log.ErrorByKv("PushMsgToUser is failed By Ws", "", "Addr", conn.RemoteAddr().String(),
-			"error", err, "senderPlatform", constant.PlatformIDToName(in.PlatformID), "recvPlatform", RecvPlatForm, "args", in.String(), "recvID", RecvID)
+			"error", err, "senderPlatform", constant.PlatformIDToName(in.MsgData.SenderPlatformID), "recvPlatform", RecvPlatForm, "args", in.String(), "recvID", RecvID)
 		ResultCode = -2
 		return ResultCode
 	} else {
@@ -171,9 +141,9 @@ func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.MsgToUserReq, RecvPl
 	}
 
 }
-func genUidPlatformArray(uid string) (array []string) {
+func genPlatformArray() (array []string) {
 	for i := 1; i <= constant.LinuxPlatformID; i++ {
-		array = append(array, uid+" "+constant.PlatformIDToName(int32(i)))
+		array = append(array, constant.PlatformIDToName(int32(i)))
 	}
 	return array
 }

internal/msg_gateway/gate/ws_server.go

@@ -9,6 +9,7 @@ import (
 	"Open_IM/pkg/utils"
 	"bytes"
 	"encoding/gob"
+	"github.com/garyburd/redigo/redis"
 	"net/http"
 	"sync"
 	"time"
@@ -24,15 +25,15 @@ type WServer struct {
 	wsAddr       string
 	wsMaxConnNum int
 	wsUpGrader   *websocket.Upgrader
-	wsConnToUser map[*UserConn]string
-	wsUserToConn map[string]*UserConn
+	wsConnToUser map[*UserConn]map[string]string
+	wsUserToConn map[string]map[string]*UserConn
 }
 
 func (ws *WServer) onInit(wsPort int) {
 	ws.wsAddr = ":" + utils.IntToString(wsPort)
 	ws.wsMaxConnNum = config.Config.LongConnSvr.WebsocketMaxConnNum
-	ws.wsConnToUser = make(map[*UserConn]string)
-	ws.wsUserToConn = make(map[string]*UserConn)
+	ws.wsConnToUser = make(map[*UserConn]map[string]string)
+	ws.wsUserToConn = make(map[string]map[string]*UserConn)
 	ws.wsUpGrader = &websocket.Upgrader{
 		HandshakeTimeout: time.Duration(config.Config.LongConnSvr.WebsocketTimeOut) * time.Second,
 		ReadBufferSize:   config.Config.LongConnSvr.WebsocketMaxMsgLen,
@@ -74,7 +75,8 @@ func (ws *WServer) readMsg(conn *UserConn) {
 			log.NewInfo("", "this is a  pingMessage")
 		}
 		if err != nil {
-			log.ErrorByKv("WS ReadMsg error", "", "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn), "error", err)
+			uid, platform := ws.getUserUid(conn)
+			log.ErrorByKv("WS ReadMsg error", "", "userIP", conn.RemoteAddr().String(), "userUid", uid, "platform", platform, "error", err.Error())
 			ws.delUserConn(conn)
 			return
 		} else {
@@ -94,34 +96,43 @@ func (ws *WServer) writeMsg(conn *UserConn, a int, msg []byte) error {
 func (ws *WServer) MultiTerminalLoginChecker(uid string, platformID int32, newConn *UserConn, token string) {
 	switch config.Config.MultiLoginPolicy {
 	case constant.AllLoginButSameTermKick:
-		if oldConn, ok := ws.wsUserToConn[genMapKey(uid, platformID)]; ok {
-			log.NewDebug("", uid, platformID, "kick old conn")
-			ws.sendKickMsg(oldConn, newConn)
-			m, err := db.DB.GetTokenMapByUidPid(uid, constant.PlatformIDToName(platformID))
-			if err != nil {
-				log.NewError("", "get token from redis err", err.Error())
-				return
-			}
-			if m == nil {
-				log.NewError("", "get token from redis err", "m is nil")
-				return
-			}
-			for k, _ := range m {
-				if k != token {
-					m[k] = constant.KickedToken
+		if oldConnMap, ok := ws.wsUserToConn[uid]; ok {
+			if oldConn, ok := oldConnMap[constant.PlatformIDToName(platformID)]; ok {
+				log.NewDebug("", uid, platformID, "kick old conn")
+				ws.sendKickMsg(oldConn, newConn)
+				m, err := db.DB.GetTokenMapByUidPid(uid, constant.PlatformIDToName(platformID))
+				if err != nil && err != redis.ErrNil {
+					log.NewError("", "get token from redis err", err.Error())
+					return
 				}
+				if m == nil {
+					log.NewError("", "get token from redis err", "m is nil")
+					return
+				}
+				for k, _ := range m {
+					if k != token {
+						m[k] = constant.KickedToken
+					}
+				}
+				log.NewDebug("get map is ", m)
+				err = db.DB.SetTokenMapByUidPid(uid, platformID, m)
+				if err != nil {
+					log.NewError("", "SetTokenMapByUidPid err", err.Error())
+					return
+				}
+				err = oldConn.Close()
+				delete(oldConnMap, constant.PlatformIDToName(platformID))
+				ws.wsUserToConn[uid] = oldConnMap
+				if len(oldConnMap) == 0 {
+					delete(ws.wsUserToConn, uid)
+				}
+				delete(ws.wsConnToUser, oldConn)
+				if err != nil {
+					log.NewError("", "conn close err", err.Error())
+				}
+
 			}
-			log.NewDebug("get map is ", m)
-			err = db.DB.SetTokenMapByUidPid(uid, platformID, m)
-			if err != nil {
-				log.NewError("", "SetTokenMapByUidPid err", err.Error())
-				return
-			}
-			err = oldConn.Close()
-			delete(ws.wsConnToUser, oldConn)
-			if err != nil {
-				log.NewError("", "conn close err", err.Error())
-			}
+
 		} else {
 			log.NewDebug("no other conn", ws.wsUserToConn)
 		}
@@ -149,53 +160,97 @@ func (ws *WServer) sendKickMsg(oldConn, newConn *UserConn) {
 	}
 }
 func (ws *WServer) addUserConn(uid string, platformID int32, conn *UserConn, token string) {
-	key := genMapKey(uid, platformID)
 	rwLock.Lock()
 	defer rwLock.Unlock()
 	ws.MultiTerminalLoginChecker(uid, platformID, conn, token)
-	ws.wsConnToUser[conn] = key
-	ws.wsUserToConn[key] = conn
-	log.WarnByKv("WS Add operation", "", "wsUser added", ws.wsUserToConn, "uid", uid, "online_num", len(ws.wsUserToConn))
+	if oldConnMap, ok := ws.wsUserToConn[uid]; ok {
+		oldConnMap[constant.PlatformIDToName(platformID)] = conn
+		ws.wsUserToConn[uid] = oldConnMap
+	} else {
+		i := make(map[string]*UserConn)
+		i[constant.PlatformIDToName(platformID)] = conn
+		ws.wsUserToConn[uid] = i
+	}
+	if oldStringMap, ok := ws.wsConnToUser[conn]; ok {
+		oldStringMap[constant.PlatformIDToName(platformID)] = uid
+		ws.wsConnToUser[conn] = oldStringMap
+	} else {
+		i := make(map[string]string)
+		i[constant.PlatformIDToName(platformID)] = uid
+		ws.wsConnToUser[conn] = i
+	}
+	count := 0
+	for _, v := range ws.wsUserToConn {
+		count = count + len(v)
+	}
+	log.WarnByKv("WS Add operation", "", "wsUser added", ws.wsUserToConn, "uid", uid, "online_user_num", len(ws.wsUserToConn), "online_conn_num", count)
 
 }
 
 func (ws *WServer) delUserConn(conn *UserConn) {
 	rwLock.Lock()
 	defer rwLock.Unlock()
-	var uidPlatform string
-	if uid, ok := ws.wsConnToUser[conn]; ok {
-		uidPlatform = uid
-		if _, ok = ws.wsUserToConn[uid]; ok {
-			delete(ws.wsUserToConn, uid)
-			log.WarnByKv("WS delete operation", "", "wsUser deleted", ws.wsUserToConn, "uid", uid, "online_num", len(ws.wsUserToConn))
+	var platform, uid string
+	if oldStringMap, ok := ws.wsConnToUser[conn]; ok {
+		for k, v := range oldStringMap {
+			platform = k
+			uid = v
+		}
+		if oldConnMap, ok := ws.wsUserToConn[uid]; ok {
+			delete(oldConnMap, platform)
+			ws.wsUserToConn[uid] = oldConnMap
+			if len(oldConnMap) == 0 {
+				delete(ws.wsUserToConn, uid)
+			}
+			count := 0
+			for _, v := range ws.wsUserToConn {
+				count = count + len(v)
+			}
+			log.WarnByKv("WS delete operation", "", "wsUser deleted", ws.wsUserToConn, "uid", uid, "online_user_num", len(ws.wsUserToConn), "online_conn_num", count)
 		} else {
-			log.WarnByKv("uid not exist", "", "wsUser deleted", ws.wsUserToConn, "uid", uid, "online_num", len(ws.wsUserToConn))
+			log.WarnByKv("WS delete operation", "", "wsUser deleted", ws.wsUserToConn, "uid", uid, "online_user_num", len(ws.wsUserToConn))
 		}
 		delete(ws.wsConnToUser, conn)
+
 	}
 	err := conn.Close()
 	if err != nil {
-		log.ErrorByKv("close err", "", "uid", uidPlatform)
+		log.ErrorByKv("close err", "", "uid", uid, "platform", platform)
+
 	}
 
 }
 
-func (ws *WServer) getUserConn(uid string) *UserConn {
+func (ws *WServer) getUserConn(uid string, platform string) *UserConn {
 	rwLock.RLock()
 	defer rwLock.RUnlock()
-	if conn, ok := ws.wsUserToConn[uid]; ok {
-		return conn
+	if connMap, ok := ws.wsUserToConn[uid]; ok {
+		if conn, flag := connMap[platform]; flag {
+			return conn
+		}
 	}
 	return nil
 }
-func (ws *WServer) getUserUid(conn *UserConn) string {
+func (ws *WServer) getSingleUserAllConn(uid string) map[string]*UserConn {
+	rwLock.RLock()
+	defer rwLock.RUnlock()
+	if connMap, ok := ws.wsUserToConn[uid]; ok {
+		return connMap
+	}
+	return nil
+}
+func (ws *WServer) getUserUid(conn *UserConn) (uid, platform string) {
 	rwLock.RLock()
 	defer rwLock.RUnlock()
 
-	if uid, ok := ws.wsConnToUser[conn]; ok {
-		return uid
+	if stringMap, ok := ws.wsConnToUser[conn]; ok {
+		for k, v := range stringMap {
+			platform = k
+			uid = v
+		}
+		return uid, platform
 	}
-	return ""
+	return "", ""
 }
 func (ws *WServer) headerCheck(w http.ResponseWriter, r *http.Request) bool {
 	status := http.StatusUnauthorized

internal/msg_transfer/logic/db.go

@@ -8,16 +8,18 @@ import (
 	"Open_IM/pkg/utils"
 )
 
-func saveUserChat(uid string, pbMsg *pbMsg.MsgSvrToPushSvrChatMsg) error {
+func saveUserChat(uid string, msg *pbMsg.MsgDataToMQ) error {
 	time := utils.GetCurrentTimestampByMill()
 	seq, err := db.DB.IncrUserSeq(uid)
 	if err != nil {
-		log.NewError(pbMsg.OperationID, "data insert to redis err", err.Error(), pbMsg.String())
+		log.NewError(msg.OperationID, "data insert to redis err", err.Error(), msg.String())
 		return err
 	}
-	pbMsg.RecvSeq = seq
-	log.NewInfo(pbMsg.OperationID, "IncrUserSeq cost time", utils.GetCurrentTimestampByMill()-time)
-	return db.DB.SaveUserChat(uid, pbMsg.SendTime, pbMsg)
+	msg.MsgData.Seq = seq
+	pbSaveData := pbMsg.MsgDataToDB{}
+	pbSaveData.MsgData = msg.MsgData
+	log.NewInfo(msg.OperationID, "IncrUserSeq cost time", utils.GetCurrentTimestampByMill()-time)
+	return db.DB.SaveUserChat(uid, pbSaveData.MsgData.SendTime, &pbSaveData)
 }
 
 func getGroupList(groupID string) ([]string, error) {

internal/msg_transfer/logic/history_msg_handler.go

@@ -32,80 +32,60 @@ func (mc *HistoryConsumerHandler) Init() {
 }
 
 func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string) {
-	log.InfoByKv("chat come mongo!!!", "", "chat", string(msg))
+	log.InfoByKv("msg come mongo!!!", "", "msg", string(msg))
 	time := utils.GetCurrentTimestampByNano()
-	pbData := pbMsg.WSToMsgSvrChatMsg{}
-	err := proto.Unmarshal(msg, &pbData)
+	msgFromMQ := pbMsg.MsgDataToMQ{}
+	err := proto.Unmarshal(msg, &msgFromMQ)
 	if err != nil {
-		log.ErrorByKv("msg_transfer Unmarshal chat err", "", "chat", string(msg), "err", err.Error())
+		log.ErrorByKv("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
 		return
 	}
-	pbSaveData := pbMsg.MsgSvrToPushSvrChatMsg{}
-	pbSaveData.SendID = pbData.SendID
-	pbSaveData.SenderNickName = pbData.SenderNickName
-	pbSaveData.SenderFaceURL = pbData.SenderFaceURL
-	pbSaveData.ClientMsgID = pbData.ClientMsgID
-	pbSaveData.SendTime = pbData.SendTime
-	pbSaveData.Content = pbData.Content
-	pbSaveData.MsgFrom = pbData.MsgFrom
-	pbSaveData.ContentType = pbData.ContentType
-	pbSaveData.SessionType = pbData.SessionType
-	pbSaveData.MsgID = pbData.MsgID
-	pbSaveData.OperationID = pbData.OperationID
-	pbSaveData.RecvID = pbData.RecvID
-	pbSaveData.PlatformID = pbData.PlatformID
-	options := utils.JsonStringToMap(pbData.Options)
+	operationID := msgFromMQ.OperationID
 	//Control whether to store offline messages (mongo)
-	isHistory := utils.GetSwitchFromOptions(options, "history")
+	isHistory := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsHistory)
 	//Control whether to store history messages (mysql)
-	isPersist := utils.GetSwitchFromOptions(options, "persistent")
-	switch pbData.SessionType {
+	isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
+	switch msgFromMQ.MsgData.SessionType {
 	case constant.SingleChatType:
-		log.NewDebug(pbSaveData.OperationID, "msg_transfer chat type = SingleChatType", isHistory, isPersist)
+		log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = SingleChatType", isHistory, isPersist)
 		if isHistory {
-			if msgKey == pbSaveData.RecvID {
-				err := saveUserChat(pbData.RecvID, &pbSaveData)
+			if msgKey == msgFromMQ.MsgData.RecvID {
+				err := saveUserChat(msgFromMQ.MsgData.RecvID, &msgFromMQ)
 				if err != nil {
-					log.NewError(pbSaveData.OperationID, "single data insert to mongo err", err.Error(), pbSaveData.String())
+					log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
 					return
 				}
 
-			} else if msgKey == pbSaveData.SendID {
-				err := saveUserChat(pbData.SendID, &pbSaveData)
+			} else if msgKey == msgFromMQ.MsgData.SendID {
+				err := saveUserChat(msgFromMQ.MsgData.SendID, &msgFromMQ)
 				if err != nil {
-					log.NewError(pbSaveData.OperationID, "single data insert to mongo err", err.Error(), pbSaveData.String())
+					log.NewError(operationID, "single data insert to mongo err", err.Error(), msgFromMQ.String())
 					return
 				}
-
 			}
 
-			log.NewDebug(pbSaveData.OperationID, "saveUserChat cost time ", utils.GetCurrentTimestampByNano()-time)
+			log.NewDebug(operationID, "saveUserChat cost time ", utils.GetCurrentTimestampByNano()-time)
 		}
-		if msgKey == pbSaveData.RecvID {
-			pbSaveData.Options = pbData.Options
-			pbSaveData.OfflineInfo = pbData.OfflineInfo
-			go sendMessageToPush(&pbSaveData)
-			log.NewDebug(pbSaveData.OperationID, "sendMessageToPush cost time ", utils.GetCurrentTimestampByNano()-time)
+		if msgKey == msgFromMQ.MsgData.RecvID {
+			go sendMessageToPush(&msgFromMQ)
+			log.NewDebug(msgFromMQ.OperationID, "sendMessageToPush cost time ", utils.GetCurrentTimestampByNano()-time)
 		}
 
 	case constant.GroupChatType:
-		log.NewDebug(pbSaveData.OperationID, "msg_transfer chat type = GroupChatType", isHistory, isPersist)
+		log.NewDebug(msgFromMQ.OperationID, "msg_transfer msg type = GroupChatType", isHistory, isPersist)
 		if isHistory {
-			uidAndGroupID := strings.Split(pbData.RecvID, " ")
-			err := saveUserChat(uidAndGroupID[0], &pbSaveData)
+			err := saveUserChat(msgFromMQ.MsgData.RecvID, &msgFromMQ)
 			if err != nil {
-				log.NewError(pbSaveData.OperationID, "group data insert to mongo err", pbSaveData.String(), uidAndGroupID[0], err.Error())
+				log.NewError(operationID, "group data insert to mongo err", msgFromMQ.String(), msgFromMQ.MsgData.RecvID, err.Error())
 				return
 			}
 		}
-		pbSaveData.Options = pbData.Options
-		pbSaveData.OfflineInfo = pbData.OfflineInfo
-		go sendMessageToPush(&pbSaveData)
+		go sendMessageToPush(&msgFromMQ)
 	default:
-		log.NewError(pbSaveData.OperationID, "SessionType error", pbSaveData.String())
+		log.NewError(msgFromMQ.OperationID, "SessionType error", msgFromMQ.String())
 		return
 	}
-	log.NewDebug(pbSaveData.OperationID, "msg_transfer handle topic data to database success...", pbSaveData.String())
+	log.NewDebug(msgFromMQ.OperationID, "msg_transfer handle topic data to database success...", msgFromMQ.String())
 }
 
 func (HistoryConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
@@ -113,50 +93,35 @@ func (HistoryConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { ret
 func (mc *HistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	claim sarama.ConsumerGroupClaim) error {
 	for msg := range claim.Messages() {
-		log.InfoByKv("kafka get info to mongo", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "chat", string(msg.Value))
+		log.InfoByKv("kafka get info to mongo", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
 		mc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
 		sess.MarkMessage(msg, "")
 	}
 	return nil
 }
-func sendMessageToPush(message *pbMsg.MsgSvrToPushSvrChatMsg) {
+func sendMessageToPush(message *pbMsg.MsgDataToMQ) {
 	log.InfoByKv("msg_transfer send message to push", message.OperationID, "message", message.String())
-	msg := pbPush.PushMsgReq{}
-	msg.OperationID = message.OperationID
-	msg.PlatformID = message.PlatformID
-	msg.Content = message.Content
-	msg.ContentType = message.ContentType
-	msg.SessionType = message.SessionType
-	msg.RecvID = message.RecvID
-	msg.SendID = message.SendID
-	msg.SenderNickName = message.SenderNickName
-	msg.SenderFaceURL = message.SenderFaceURL
-	msg.ClientMsgID = message.ClientMsgID
-	msg.MsgFrom = message.MsgFrom
-	msg.Options = message.Options
-	msg.RecvSeq = message.RecvSeq
-	msg.SendTime = message.SendTime
-	msg.MsgID = message.MsgID
-	msg.OfflineInfo = message.OfflineInfo
+	rpcPushMsg := pbPush.PushMsgReq{OperationID: message.OperationID, MsgData: message.MsgData}
+	mqPushMsg := pbMsg.PushMsgDataToMQ{OperationID: message.OperationID, MsgData: message.MsgData}
 	grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImPushName)
 	if grpcConn == nil {
-		log.ErrorByKv("rpc dial failed", msg.OperationID, "push data", msg.String())
-		pid, offset, err := producer.SendMessage(message)
+		log.ErrorByKv("rpc dial failed", rpcPushMsg.OperationID, "push data", rpcPushMsg.String())
+		pid, offset, err := producer.SendMessage(&mqPushMsg)
 		if err != nil {
-			log.ErrorByKv("kafka send failed", msg.OperationID, "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
+			log.ErrorByKv("kafka send failed", mqPushMsg.OperationID, "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
 		}
 		return
 	}
 	msgClient := pbPush.NewPushMsgServiceClient(grpcConn)
-	_, err := msgClient.PushMsg(context.Background(), &msg)
+	_, err := msgClient.PushMsg(context.Background(), &rpcPushMsg)
 	if err != nil {
-		log.ErrorByKv("rpc send failed", msg.OperationID, "push data", msg.String(), "err", err.Error())
-		pid, offset, err := producer.SendMessage(message)
+		log.ErrorByKv("rpc send failed", rpcPushMsg.OperationID, "push data", rpcPushMsg.String(), "err", err.Error())
+		pid, offset, err := producer.SendMessage(&mqPushMsg)
 		if err != nil {
-			log.ErrorByKv("kafka send failed", msg.OperationID, "send data", message.String(), "pid", pid, "offset", offset, "err", err.Error())
+			log.ErrorByKv("kafka send failed", mqPushMsg.OperationID, "send data", mqPushMsg.String(), "pid", pid, "offset", offset, "err", err.Error())
 		}
 	} else {
-		log.InfoByKv("rpc send success", msg.OperationID, "push data", msg.String())
+		log.InfoByKv("rpc send success", rpcPushMsg.OperationID, "push data", rpcPushMsg.String())
 
 	}
 }

internal/msg_transfer/logic/persistent_msg_handler.go

@@ -16,7 +16,6 @@ import (
 	"Open_IM/pkg/utils"
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
-	"strings"
 )
 
 type PersistentConsumerHandler struct {
@@ -32,30 +31,29 @@ func (pc *PersistentConsumerHandler) Init() {
 		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql)
 
 }
+
 func (pc *PersistentConsumerHandler) handleChatWs2Mysql(msg []byte, msgKey string) {
-	log.InfoByKv("chat come here mysql!!!", "", "chat", string(msg))
-	pbData := pbMsg.WSToMsgSvrChatMsg{}
-	err := proto.Unmarshal(msg, &pbData)
+	log.InfoByKv("msg come here mysql!!!", "", "msg", string(msg))
+	msgFromMQ := pbMsg.MsgDataToMQ{}
+	err := proto.Unmarshal(msg, &msgFromMQ)
 	if err != nil {
-		log.ErrorByKv("msg_transfer Unmarshal chat err", "", "chat", string(msg), "err", err.Error())
+		log.ErrorByKv("msg_transfer Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
 		return
 	}
-	options := utils.JsonStringToMap(pbData.Options)
 	//Control whether to store history messages (mysql)
-	isPersist := utils.GetSwitchFromOptions(options, "persistent")
+	isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
 	//Only process receiver data
 	if isPersist {
-		if msgKey == pbData.RecvID && pbData.SessionType == constant.SingleChatType {
-			log.InfoByKv("msg_transfer chat persisting", pbData.OperationID)
-			if err = im_mysql_msg_model.InsertMessageToChatLog(pbData); err != nil {
-				log.ErrorByKv("Message insert failed", pbData.OperationID, "err", err.Error(), "chat", pbData.String())
+		if msgKey == msgFromMQ.MsgData.RecvID && msgFromMQ.MsgData.SessionType == constant.SingleChatType {
+			log.InfoByKv("msg_transfer msg persisting", msgFromMQ.OperationID)
+			if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil {
+				log.ErrorByKv("Message insert failed", msgFromMQ.OperationID, "err", err.Error(), "msg", msgFromMQ.String())
 				return
 			}
-		} else if pbData.SessionType == constant.GroupChatType && msgKey == "0" {
-			pbData.RecvID = strings.Split(pbData.RecvID, " ")[1]
-			log.InfoByKv("msg_transfer chat persisting", pbData.OperationID)
-			if err = im_mysql_msg_model.InsertMessageToChatLog(pbData); err != nil {
-				log.ErrorByKv("Message insert failed", pbData.OperationID, "err", err.Error(), "chat", pbData.String())
+		} else if msgFromMQ.MsgData.SessionType == constant.GroupChatType && msgKey == msgFromMQ.MsgData.SendID {
+			log.InfoByKv("msg_transfer msg persisting", msgFromMQ.OperationID)
+			if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil {
+				log.ErrorByKv("Message insert failed", msgFromMQ.OperationID, "err", err.Error(), "msg", msgFromMQ.String())
 				return
 			}
 		}
@@ -67,7 +65,7 @@ func (PersistentConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error {
 func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
 	claim sarama.ConsumerGroupClaim) error {
 	for msg := range claim.Messages() {
-		log.InfoByKv("kafka get info to mysql", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "chat", string(msg.Value))
+		log.InfoByKv("kafka get info to mysql", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
 		pc.msgHandle[msg.Topic](msg.Value, string(msg.Key))
 		sess.MarkMessage(msg, "")
 	}

internal/push/logic/push_handler.go

@@ -11,8 +11,7 @@ import (
 	kfk "Open_IM/pkg/common/kafka"
 	"Open_IM/pkg/common/log"
 	pbChat "Open_IM/pkg/proto/chat"
-	pbRelay "Open_IM/pkg/proto/relay"
-	"Open_IM/pkg/utils"
+	pbPush "Open_IM/pkg/proto/push"
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
 )
@@ -33,28 +32,13 @@ func (ms *PushConsumerHandler) Init() {
 }
 func (ms *PushConsumerHandler) handleMs2PsChat(msg []byte) {
 	log.InfoByKv("msg come from kafka  And push!!!", "", "msg", string(msg))
-	pbData := pbChat.MsgSvrToPushSvrChatMsg{}
-	if err := proto.Unmarshal(msg, &pbData); err != nil {
+	msgFromMQ := pbChat.PushMsgDataToMQ{}
+	if err := proto.Unmarshal(msg, &msgFromMQ); err != nil {
 		log.ErrorByKv("push Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
 		return
 	}
-	sendPbData := pbRelay.MsgToUserReq{}
-	sendPbData.SendTime = pbData.SendTime
-	sendPbData.OperationID = pbData.OperationID
-	sendPbData.ServerMsgID = pbData.MsgID
-	sendPbData.MsgFrom = pbData.MsgFrom
-	sendPbData.ContentType = pbData.ContentType
-	sendPbData.SessionType = pbData.SessionType
-	sendPbData.RecvID = pbData.RecvID
-	sendPbData.Content = pbData.Content
-	sendPbData.SendID = pbData.SendID
-	sendPbData.SenderNickName = pbData.SenderNickName
-	sendPbData.SenderFaceURL = pbData.SenderFaceURL
-	sendPbData.ClientMsgID = pbData.ClientMsgID
-	sendPbData.PlatformID = pbData.PlatformID
-	sendPbData.RecvSeq = pbData.RecvSeq
 	//Call push module to send message to the user
-	MsgToUser(&sendPbData, pbData.OfflineInfo, utils.JsonStringToMap(pbData.Options))
+	MsgToUser((*pbPush.PushMsgReq)(&msgFromMQ))
 }
 func (PushConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
 func (PushConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

internal/push/logic/push_rpc_server.go

@@ -5,7 +5,6 @@ import (
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	"Open_IM/pkg/proto/push"
-	pbRelay "Open_IM/pkg/proto/relay"
 	"Open_IM/pkg/utils"
 	"context"
 	"google.golang.org/grpc"
@@ -49,23 +48,8 @@ func (r *RPCServer) run() {
 	}
 }
 func (r *RPCServer) PushMsg(_ context.Context, pbData *pbPush.PushMsgReq) (*pbPush.PushMsgResp, error) {
-	sendPbData := pbRelay.MsgToUserReq{}
-	sendPbData.SendTime = pbData.SendTime
-	sendPbData.OperationID = pbData.OperationID
-	sendPbData.ServerMsgID = pbData.MsgID
-	sendPbData.MsgFrom = pbData.MsgFrom
-	sendPbData.ContentType = pbData.ContentType
-	sendPbData.SenderNickName = pbData.SenderNickName
-	sendPbData.SenderFaceURL = pbData.SenderFaceURL
-	sendPbData.ClientMsgID = pbData.ClientMsgID
-	sendPbData.SessionType = pbData.SessionType
-	sendPbData.RecvID = pbData.RecvID
-	sendPbData.Content = pbData.Content
-	sendPbData.SendID = pbData.SendID
-	sendPbData.PlatformID = pbData.PlatformID
-	sendPbData.RecvSeq = pbData.RecvSeq
 	//Call push module to send message to the user
-	MsgToUser(&sendPbData, pbData.OfflineInfo, utils.JsonStringToMap(pbData.Options))
+	MsgToUser(pbData)
 	return &pbPush.PushMsgResp{
 		ResultCode: 0,
 	}, nil

internal/push/logic/push_to_client.go

@@ -8,13 +8,11 @@ package logic
 
 import (
 	push "Open_IM/internal/push/jpush"
-	rpcChat "Open_IM/internal/rpc/chat"
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/log"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
-	pbChat "Open_IM/pkg/proto/chat"
-	pbGroup "Open_IM/pkg/proto/group"
+	pbPush "Open_IM/pkg/proto/push"
 	pbRelay "Open_IM/pkg/proto/relay"
 	"Open_IM/pkg/utils"
 	"context"
@@ -34,122 +32,119 @@ type AtContent struct {
 	IsAtSelf   bool     `json:"isAtSelf"`
 }
 
-func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo string, Options map[string]int32) {
+func MsgToUser(pushMsg *pbPush.PushMsgReq) {
 	var wsResult []*pbRelay.SingleMsgToUser
-	isOfflinePush := utils.GetSwitchFromOptions(Options, "offlinePush")
-	log.InfoByKv("Get chat from msg_transfer And push chat", sendPbData.OperationID, "PushData", sendPbData, Options, isOfflinePush)
+	isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush)
+	log.InfoByKv("Get msg from msg_transfer And push msg", pushMsg.OperationID, "PushData", pushMsg.String())
 	grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
 	//Online push message
-	log.InfoByKv("test", sendPbData.OperationID, "len  grpc", len(grpcCons), "data", sendPbData)
+	log.InfoByKv("test", pushMsg.OperationID, "len  grpc", len(grpcCons), "data", pushMsg.String())
 	for _, v := range grpcCons {
 		msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v)
-		reply, err := msgClient.MsgToUser(context.Background(), sendPbData)
+		reply, err := msgClient.OnlinePushMsg(context.Background(), &pbRelay.OnlinePushMsgReq{OperationID: pushMsg.OperationID, MsgData: pushMsg.MsgData})
 		if err != nil {
-			log.InfoByKv("push data to client rpc err", sendPbData.OperationID, "err", err)
+			log.InfoByKv("push data to client rpc err", pushMsg.OperationID, "err", err)
 			continue
 		}
 		if reply != nil && reply.Resp != nil {
 			wsResult = append(wsResult, reply.Resp...)
 		}
 	}
-	log.InfoByKv("push_result", sendPbData.OperationID, "result", wsResult, "sendData", sendPbData)
-	if sendPbData.ContentType != constant.Typing && sendPbData.ContentType != constant.HasReadReceipt {
-		if isOfflinePush {
-			for _, v := range wsResult {
-				if v.ResultCode == 0 {
-					continue
-				}
-				//supported terminal
-				for _, t := range pushTerminal {
-					if v.RecvPlatFormID == t {
-						//Use offline push messaging
-						var UIDList []string
-						UIDList = append(UIDList, v.RecvID)
-						customContent := OpenIMContent{
-							SessionType: int(sendPbData.SessionType),
-							From:        sendPbData.SendID,
-							To:          sendPbData.RecvID,
-							Seq:         sendPbData.RecvSeq,
-						}
-						bCustomContent, _ := json.Marshal(customContent)
-						jsonCustomContent := string(bCustomContent)
-						var content string
-						switch sendPbData.ContentType {
-						case constant.Text:
-							content = constant.ContentType2PushContent[constant.Text]
-						case constant.Picture:
-							content = constant.ContentType2PushContent[constant.Picture]
-						case constant.Voice:
-							content = constant.ContentType2PushContent[constant.Voice]
-						case constant.Video:
-							content = constant.ContentType2PushContent[constant.Video]
-						case constant.File:
-							content = constant.ContentType2PushContent[constant.File]
-						case constant.AtText:
-							a := AtContent{}
-							_ = utils.JsonStringToStruct(sendPbData.Content, &a)
-							if utils.IsContain(v.RecvID, a.AtUserList) {
-								content = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
-							} else {
-								content = constant.ContentType2PushContent[constant.GroupMsg]
-							}
-						default:
-							content = constant.ContentType2PushContent[constant.Common]
-						}
-						pushResult, err := push.JGAccountListPush(UIDList, content, jsonCustomContent, constant.PlatformIDToName(t))
-						if err != nil {
-							log.NewError(sendPbData.OperationID, "offline push error", sendPbData.String(), err.Error(), constant.PlatformIDToName(t))
-						} else {
-							log.NewDebug(sendPbData.OperationID, "offline push return result is ", string(pushResult), sendPbData, constant.PlatformIDToName(t))
-						}
-
+	log.InfoByKv("push_result", pushMsg.OperationID, "result", wsResult, "sendData", pushMsg.MsgData)
+	if isOfflinePush {
+		for _, v := range wsResult {
+			if v.ResultCode == 0 {
+				continue
+			}
+			//supported terminal
+			for _, t := range pushTerminal {
+				if v.RecvPlatFormID == t {
+					//Use offline push messaging
+					var UIDList []string
+					UIDList = append(UIDList, v.RecvID)
+					customContent := OpenIMContent{
+						SessionType: int(pushMsg.MsgData.SessionType),
+						From:        pushMsg.MsgData.SendID,
+						To:          pushMsg.MsgData.RecvID,
+						Seq:         pushMsg.MsgData.Seq,
 					}
+					bCustomContent, _ := json.Marshal(customContent)
+					jsonCustomContent := string(bCustomContent)
+					var content string
+					switch pushMsg.MsgData.ContentType {
+					case constant.Text:
+						content = constant.ContentType2PushContent[constant.Text]
+					case constant.Picture:
+						content = constant.ContentType2PushContent[constant.Picture]
+					case constant.Voice:
+						content = constant.ContentType2PushContent[constant.Voice]
+					case constant.Video:
+						content = constant.ContentType2PushContent[constant.Video]
+					case constant.File:
+						content = constant.ContentType2PushContent[constant.File]
+					case constant.AtText:
+						a := AtContent{}
+						_ = utils.JsonStringToStruct(string(pushMsg.MsgData.Content), &a)
+						if utils.IsContain(v.RecvID, a.AtUserList) {
+							content = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
+						} else {
+							content = constant.ContentType2PushContent[constant.GroupMsg]
+						}
+					default:
+						content = constant.ContentType2PushContent[constant.Common]
+					}
+					pushResult, err := push.JGAccountListPush(UIDList, content, jsonCustomContent, constant.PlatformIDToName(t))
+					if err != nil {
+						log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error(), constant.PlatformIDToName(t))
+					} else {
+						log.NewDebug(pushMsg.OperationID, "offline push return result is ", string(pushResult), pushMsg.MsgData, constant.PlatformIDToName(t))
+					}
+
 				}
 			}
-
 		}
-	}
-
-}
-
-func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) {
-	m.MsgID = rpcChat.GetMsgID(m.SendID)
-	m.ClientMsgID = m.MsgID
-	switch m.SessionType {
-	case constant.SingleChatType:
-		sendMsgToKafka(m, m.SendID, "msgKey--sendID")
-		sendMsgToKafka(m, m.RecvID, "msgKey--recvID")
-	case constant.GroupChatType:
-		etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
-		client := pbGroup.NewGroupClient(etcdConn)
-		req := &pbGroup.GetGroupAllMemberReq{
-			GroupID:     m.RecvID,
-			Token:       config.Config.Secret,
-			OperationID: m.OperationID,
-		}
-		reply, err := client.GetGroupAllMember(context.Background(), req)
-		if err != nil {
-			log.Error(m.Token, m.OperationID, "rpc  getGroupInfo failed, err = %s", err.Error())
-			return
-		}
-		if reply.ErrorCode != 0 {
-			log.Error(m.Token, m.OperationID, "rpc  getGroupInfo failed, err = %s", reply.ErrorMsg)
-			return
-		}
-		groupID := m.RecvID
-		for i, v := range reply.MemberList {
-			m.RecvID = v.UserId + " " + groupID
-			sendMsgToKafka(m, utils.IntToString(i), "msgKey--recvID+\" \"+groupID")
-		}
-	default:
 
 	}
 }
 
-func sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string, flag string) {
-	pid, offset, err := 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(), flag, key)
-	}
-
-}
+//func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) {
+//	m.MsgID = rpcChat.GetMsgID(m.SendID)
+//	m.ClientMsgID = m.MsgID
+//	switch m.SessionType {
+//	case constant.SingleChatType:
+//		sendMsgToKafka(m, m.SendID, "msgKey--sendID")
+//		sendMsgToKafka(m, m.RecvID, "msgKey--recvID")
+//	case constant.GroupChatType:
+//		etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
+//		client := pbGroup.NewGroupClient(etcdConn)
+//		req := &pbGroup.GetGroupAllMemberReq{
+//			GroupID:     m.RecvID,
+//			Token:       config.Config.Secret,
+//			OperationID: m.OperationID,
+//		}
+//		reply, err := client.GetGroupAllMember(context.Background(), req)
+//		if err != nil {
+//			log.Error(m.Token, m.OperationID, "rpc  getGroupInfo failed, err = %s", err.Error())
+//			return
+//		}
+//		if reply.ErrorCode != 0 {
+//			log.Error(m.Token, m.OperationID, "rpc  getGroupInfo failed, err = %s", reply.ErrorMsg)
+//			return
+//		}
+//		groupID := m.RecvID
+//		for i, v := range reply.MemberList {
+//			m.RecvID = v.UserId + " " + groupID
+//			sendMsgToKafka(m, utils.IntToString(i), "msgKey--recvID+\" \"+groupID")
+//		}
+//	default:
+//
+//	}
+//}
+//
+//func sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string, flag string) {
+//	pid, offset, err := 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(), flag, key)
+//	}
+//
+//}

internal/rpc/msg/pull_message.go

@@ -1,4 +1,4 @@
-package chat
+package msg
 
 import (
 	"context"
@@ -6,7 +6,7 @@ import (
 
 	commonDB "Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
-
+	"Open_IM/pkg/proto/sdk_ws"
 	"sort"
 	"strings"
 
@@ -44,8 +44,8 @@ func (rpc *rpcChat) GetMaxAndMinSeq(_ context.Context, in *pbMsg.GetMaxAndMinSeq
 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
+	var respSingleMsgFormat []*open_im_sdk.GatherFormat
+	var respGroupMsgFormat []*open_im_sdk.GatherFormat
 	SingleMsgFormat, GroupMsgFormat, MaxSeq, MinSeq, err := commonDB.DB.GetMsgBySeqRange(in.UserID, in.SeqBegin, in.SeqEnd)
 	if err != nil {
 		log.ErrorByKv("pullMsg data error", in.OperationID, in.String())
@@ -60,15 +60,15 @@ func (rpc *rpcChat) PullMessage(_ context.Context, in *pbMsg.PullMessageReq) (*p
 		ErrMsg:        "",
 		MaxSeq:        MaxSeq,
 		MinSeq:        MinSeq,
-		SingleUserMsg: respSingleMsgFormat,
+		SingleUserMsg: []*pbMsg.GatherFormat(respSingleMsgFormat),
 		GroupUserMsg:  respGroupMsgFormat,
 	}, nil
 }
-func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *pbMsg.PullMessageBySeqListReq) (*pbMsg.PullMessageResp, error) {
+func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *open_im_sdk.PullMessageBySeqListReq) (*open_im_sdk.PullMessageBySeqListResp, error) {
 	log.NewInfo(in.OperationID, "rpc PullMessageBySeqList is arriving", in.String())
-	resp := new(pbMsg.PullMessageResp)
-	var respSingleMsgFormat []*pbMsg.GatherFormat
-	var respGroupMsgFormat []*pbMsg.GatherFormat
+	resp := new(open_im_sdk.PullMessageBySeqListResp)
+	var respSingleMsgFormat []*open_im_sdk.GatherFormat
+	var respGroupMsgFormat []*open_im_sdk.GatherFormat
 	SingleMsgFormat, GroupMsgFormat, MaxSeq, MinSeq, err := commonDB.DB.GetMsgBySeqList(in.UserID, in.SeqList)
 	if err != nil {
 		log.ErrorByKv("PullMessageBySeqList data error", in.OperationID, in.String())
@@ -78,7 +78,7 @@ func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *pbMsg.PullMessag
 	}
 	respSingleMsgFormat = singleMsgHandleByUser(SingleMsgFormat, in.UserID)
 	respGroupMsgFormat = groupMsgHandleByUser(GroupMsgFormat)
-	return &pbMsg.PullMessageResp{
+	return &open_im_sdk.PullMessageBySeqListResp{
 		ErrCode:       0,
 		ErrMsg:        "",
 		MaxSeq:        MaxSeq,
@@ -87,9 +87,9 @@ func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *pbMsg.PullMessag
 		GroupUserMsg:  respGroupMsgFormat,
 	}, nil
 }
-func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.GatherFormat {
+func singleMsgHandleByUser(allMsg []*open_im_sdk.MsgData, ownerId string) []*open_im_sdk.GatherFormat {
 	var userid string
-	var respMsgFormat []*pbMsg.GatherFormat
+	var respMsgFormat []*open_im_sdk.GatherFormat
 	m := make(map[string]MsgFormats)
 	//Gather messages in the dimension of users
 	for _, v := range allMsg {
@@ -107,7 +107,7 @@ func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.G
 	}
 	//Return in pb format
 	for user, msg := range m {
-		tempUserMsg := new(pbMsg.GatherFormat)
+		tempUserMsg := new(open_im_sdk.GatherFormat)
 		tempUserMsg.ID = user
 		tempUserMsg.List = msg
 		sort.Sort(msg)
@@ -115,8 +115,8 @@ func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.G
 	}
 	return respMsgFormat
 }
-func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
-	var respMsgFormat []*pbMsg.GatherFormat
+func groupMsgHandleByUser(allMsg []*open_im_sdk.MsgData) []*open_im_sdk.GatherFormat {
+	var respMsgFormat []*open_im_sdk.GatherFormat
 	m := make(map[string]MsgFormats)
 	//Gather messages in the dimension of users
 	for _, v := range allMsg {
@@ -132,7 +132,7 @@ func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
 	}
 	//Return in pb format
 	for groupID, msg := range m {
-		tempUserMsg := new(pbMsg.GatherFormat)
+		tempUserMsg := new(open_im_sdk.GatherFormat)
 		tempUserMsg.ID = groupID
 		tempUserMsg.List = msg
 		sort.Sort(msg)
@@ -141,7 +141,7 @@ func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
 	return respMsgFormat
 }
 
-type MsgFormats []*pbMsg.MsgFormat
+type MsgFormats []*open_im_sdk.MsgData
 
 // Implement the sort.Interface interface to get the number of elements method
 func (s MsgFormats) Len() int {

internal/rpc/msg/rpcChat.go

@@ -1,4 +1,4 @@
-package chat
+package msg
 
 import (
 	"Open_IM/pkg/common/config"