merge all branch and change project structure

internal/msg_gateway/gate/logic.go

@@ -6,62 +6,67 @@ 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"
 	"Open_IM/pkg/utils"
+	"bytes"
 	"context"
+	"encoding/gob"
 	"encoding/json"
+	"github.com/golang/protobuf/proto"
 	"github.com/gorilla/websocket"
+	"runtime"
 	"strings"
 )
 
-func (ws *WServer) msgParse(conn *websocket.Conn, jsonMsg []byte) {
+func (ws *WServer) msgParse(conn *UserConn, binaryMsg []byte) {
 	//ws online debug data
 	//{"ReqIdentifier":1001,"Token":"123","SendID":"c4ca4238a0b923820dcc509a6f75849b","Time":"123","OperationID":"123","MsgIncr":0}
 	//{"ReqIdentifier":1002,"Token":"123","SendID":"c4ca4238a0b923820dcc509a6f75849b","Time":"123","OperationID":"123","MsgIncr":0,"SeqBegin":1,"SeqEnd":6}
 	//{"ReqIdentifier":1003,"Token":"123","SendID":"c4ca4238a0b923820dcc509a6f75849b",
 	//"RecvID":"a87ff679a2f3e71d9181a67b7542122c","ClientMsgID":"2343","Time":"147878787","OperationID":
 	//"123","MsgIncr":0,"SubMsgType":101,"MsgType":100,"MsgFrom":1,"Content":"sdfsdf"}
+	b := bytes.NewBuffer(binaryMsg)
 	m := Req{}
-	if err := json.Unmarshal(jsonMsg, &m); err != nil {
+	dec := gob.NewDecoder(b)
+	err := dec.Decode(&m)
+	if err != nil {
 		log.ErrorByKv("ws json Unmarshal err", "", "err", err.Error())
-		ws.sendErrMsg(conn, 200, err.Error())
+		ws.sendErrMsg(conn, 200, err.Error(), constant.WSDataError, "", "")
+		err = conn.Close()
+		if err != nil {
+			log.NewError("", "ws close err", err.Error())
+		}
 		return
 	}
 	if err := validate.Struct(m); err != nil {
 		log.ErrorByKv("ws args validate  err", "", "err", err.Error())
-		ws.sendErrMsg(conn, 201, err.Error())
+		ws.sendErrMsg(conn, 201, err.Error(), m.ReqIdentifier, m.MsgIncr, m.OperationID)
 		return
 	}
-
-	if !utils.VerifyToken(m.Token, m.SendID) {
-		ws.sendErrMsg(conn, 202, "token validate err")
-		return
-	}
-	log.InfoByKv("Basic Info Authentication Success", m.OperationID, "reqIdentifier", m.ReqIdentifier, "sendID", m.SendID)
+	//if !utils.VerifyToken(m.Token, m.SendID) {
+	//	ws.sendErrMsg(conn, 202, "token validate err", m.ReqIdentifier, m.MsgIncr,m.OperationID)
+	//	return
+	//}
+	log.InfoByKv("Basic Info Authentication Success", m.OperationID, "reqIdentifier", m.ReqIdentifier, "sendID", m.SendID, "msgIncr", m.MsgIncr)
 
 	switch m.ReqIdentifier {
 	case constant.WSGetNewestSeq:
-		ws.newestSeqReq(conn, &m)
+		go ws.getSeqReq(conn, &m)
 	case constant.WSPullMsg:
-		ws.pullMsgReq(conn, &m)
+		go ws.pullMsgReq(conn, &m)
 	case constant.WSSendMsg:
-		ws.sendMsgReq(conn, &m)
+		sendTime := utils.GetCurrentTimestampByNano()
+		go ws.sendMsgReq(conn, &m, sendTime)
+	case constant.WSPullMsgBySeqList:
+		go ws.pullMsgBySeqListReq(conn, &m)
 	default:
 	}
+	log.NewInfo("", "goroutine num is ", runtime.NumGoroutine())
 }
-func (ws *WServer) newestSeqResp(conn *websocket.Conn, m *Req, pb *pbChat.GetNewSeqResp) {
-	mReply := make(map[string]interface{})
-	mData := make(map[string]interface{})
-	mReply["reqIdentifier"] = m.ReqIdentifier
-	mReply["msgIncr"] = m.MsgIncr
-	mReply["errCode"] = pb.GetErrCode()
-	mReply["errMsg"] = pb.GetErrMsg()
-	mData["seq"] = pb.GetSeq()
-	mReply["data"] = mData
-	ws.sendMsg(conn, mReply)
-}
-func (ws *WServer) newestSeqReq(conn *websocket.Conn, m *Req) {
-	log.InfoByKv("Ws call success to getNewSeq", m.OperationID, "Parameters", m)
-	pbData := pbChat.GetNewSeqReq{}
+func (ws *WServer) getSeqReq(conn *UserConn, m *Req) {
+	log.NewInfo(m.OperationID, "Ws call success to getNewSeq", m.MsgIncr, m.SendID, m.ReqIdentifier)
+	pbData := pbChat.GetMaxAndMinSeqReq{}
+	nReply := new(pbChat.GetMaxAndMinSeqResp)
 	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)
@@ -69,44 +74,35 @@ func (ws *WServer) newestSeqReq(conn *websocket.Conn, m *Req) {
 		log.ErrorByKv("get grpcConn err", pbData.OperationID, "args", m)
 	}
 	msgClient := pbChat.NewChatClient(grpcConn)
-	reply, err := msgClient.GetNewSeq(context.Background(), &pbData)
+	reply, err := msgClient.GetMaxAndMinSeq(context.Background(), &pbData)
 	if err != nil {
-		log.ErrorByKv("rpc call failed to getNewSeq", pbData.OperationID, "err", err, "pbData", pbData.String())
-		return
+		log.ErrorByKv("rpc call failed to getSeqReq", pbData.OperationID, "err", err, "pbData", pbData.String())
+		nReply.ErrCode = 200
+		nReply.ErrMsg = err.Error()
+		ws.getSeqResp(conn, m, nReply)
+	} else {
+		log.InfoByKv("rpc call success to getSeqReq", pbData.OperationID, "replyData", reply.String())
+		ws.getSeqResp(conn, m, reply)
 	}
-	log.InfoByKv("rpc call success to getNewSeq", pbData.OperationID, "replyData", reply.String())
-	ws.newestSeqResp(conn, m, reply)
-
 }
-
-func (ws *WServer) pullMsgResp(conn *websocket.Conn, m *Req, pb *pbChat.PullMessageResp) {
-	mReply := make(map[string]interface{})
-	msg := make(map[string]interface{})
-	mReply["reqIdentifier"] = m.ReqIdentifier
-	mReply["msgIncr"] = m.MsgIncr
-	mReply["errCode"] = pb.GetErrCode()
-	mReply["errMsg"] = pb.GetErrMsg()
-	//空切片
-	if v := pb.GetSingleUserMsg(); v != nil {
-		msg["single"] = v
-	} else {
-		msg["single"] = []pbChat.GatherFormat{}
+func (ws *WServer) getSeqResp(conn *UserConn, m *Req, pb *pbChat.GetMaxAndMinSeqResp) {
+	var mReplyData open_im_sdk.GetMaxAndMinSeqResp
+	mReplyData.MaxSeq = pb.GetMaxSeq()
+	mReplyData.MinSeq = pb.GetMinSeq()
+	b, _ := proto.Marshal(&mReplyData)
+	mReply := Resp{
+		ReqIdentifier: m.ReqIdentifier,
+		MsgIncr:       m.MsgIncr,
+		ErrCode:       pb.GetErrCode(),
+		ErrMsg:        pb.GetErrMsg(),
+		OperationID:   m.OperationID,
+		Data:          b,
 	}
-	if v := pb.GetGroupUserMsg(); v != nil {
-		msg["group"] = v
-	} else {
-		msg["group"] = []pbChat.GatherFormat{}
-	}
-	msg["maxSeq"] = pb.GetMaxSeq()
-	msg["minSeq"] = pb.GetMinSeq()
-	mReply["data"] = msg
 	ws.sendMsg(conn, mReply)
-
 }
-
-func (ws *WServer) pullMsgReq(conn *websocket.Conn, m *Req) {
-	log.InfoByKv("Ws call success to pullMsgReq", m.OperationID, "Parameters", m)
-	reply := new(pbChat.PullMessageResp)
+func (ws *WServer) pullMsgReq(conn *UserConn, m *Req) {
+	log.NewInfo(m.OperationID, "Ws call success to pullMsgReq", m.ReqIdentifier, m.MsgIncr, m.SendID)
+	nReply := new(pbChat.PullMessageResp)
 	isPass, errCode, errMsg, data := ws.argsValidate(m, constant.WSPullMsg)
 	if isPass {
 		pbData := pbChat.PullMessageReq{}
@@ -119,78 +115,169 @@ func (ws *WServer) pullMsgReq(conn *websocket.Conn, m *Req) {
 		reply, err := msgClient.PullMessage(context.Background(), &pbData)
 		if err != nil {
 			log.ErrorByKv("PullMessage error", pbData.OperationID, "err", err.Error())
-			return
+			nReply.ErrCode = 200
+			nReply.ErrMsg = err.Error()
+			ws.pullMsgResp(conn, m, nReply)
+		} else {
+			log.InfoByKv("rpc call success to pullMsgRep", pbData.OperationID, "ReplyArgs", reply.String(), "maxSeq", reply.GetMaxSeq(),
+				"MinSeq", reply.GetMinSeq(), "singLen", len(reply.GetSingleUserMsg()), "groupLen", len(reply.GetGroupUserMsg()))
+			ws.pullMsgResp(conn, m, reply)
 		}
-		log.InfoByKv("rpc call success to pullMsgRep", pbData.OperationID, "ReplyArgs", reply.String(), "maxSeq", reply.GetMaxSeq(),
-			"MinSeq", reply.GetMinSeq(), "singLen", len(reply.GetSingleUserMsg()), "groupLen", len(reply.GetGroupUserMsg()))
-		ws.pullMsgResp(conn, m, reply)
 	} else {
-		reply.ErrCode = errCode
-		reply.ErrMsg = errMsg
-		ws.pullMsgResp(conn, m, reply)
+		nReply.ErrCode = errCode
+		nReply.ErrMsg = errMsg
+		ws.pullMsgResp(conn, m, nReply)
 	}
 }
+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
+	a, err := json.Marshal(pb.SingleUserMsg)
+	if err != nil {
+		log.NewError(m.OperationID, "GetSingleUserMsg,json marshal,err", err.Error())
+	}
+	log.NewInfo(m.OperationID, "pullMsgResp json is ", len(pb.SingleUserMsg))
+	err = json.Unmarshal(a, &mReplyData.SingleUserMsg)
+	if err != nil {
+		log.NewError(m.OperationID, "SingleUserMsg,json Unmarshal,err", err.Error())
+	}
+	b, err := json.Marshal(pb.GroupUserMsg)
+	if err != nil {
+		log.NewError(m.OperationID, "mReplyData,json marshal,err", err.Error())
+	}
+	err = json.Unmarshal(b, &mReplyData.GroupUserMsg)
+	if err != nil {
+		log.NewError(m.OperationID, "test SingleUserMsg,json Unmarshal,err", err.Error())
+	}
+	c, err := proto.Marshal(&mReplyData)
+	log.NewInfo(m.OperationID, "test info is ", len(mReplyData.SingleUserMsg), mReplyData.SingleUserMsg)
+
+	mReply := Resp{
+		ReqIdentifier: m.ReqIdentifier,
+		MsgIncr:       m.MsgIncr,
+		ErrCode:       pb.GetErrCode(),
+		ErrMsg:        pb.GetErrMsg(),
+		OperationID:   m.OperationID,
+		Data:          c,
+	}
+	log.NewInfo(m.OperationID, "pullMsgResp all data  is ", mReply.ReqIdentifier, mReply.MsgIncr, mReply.ErrCode, mReply.ErrMsg,
+		len(mReply.Data))
 
-func (ws *WServer) sendMsgResp(conn *websocket.Conn, m *Req, pb *pbChat.UserSendMsgResp) {
-	mReply := make(map[string]interface{})
-	mReplyData := make(map[string]interface{})
-	mReply["reqIdentifier"] = m.ReqIdentifier
-	mReply["msgIncr"] = m.MsgIncr
-	mReply["errCode"] = pb.GetErrCode()
-	mReply["errMsg"] = pb.GetErrMsg()
-	mReplyData["clientMsgID"] = pb.GetClientMsgID()
-	mReplyData["serverMsgID"] = pb.GetServerMsgID()
-	mReply["data"] = mReplyData
 	ws.sendMsg(conn, mReply)
-}
 
-func (ws *WServer) sendMsgReq(conn *websocket.Conn, m *Req) {
-	log.InfoByKv("Ws call success to sendMsgReq", m.OperationID, "Parameters", m)
-	reply := new(pbChat.UserSendMsgResp)
+}
+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)
+	if isPass {
+		pbData := pbChat.PullMessageBySeqListReq{}
+		pbData.SeqList = data.(open_im_sdk.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)
+		msgClient := pbChat.NewChatClient(grpcConn)
+		reply, err := msgClient.PullMessageBySeqList(context.Background(), &pbData)
+		if err != nil {
+			log.NewError(pbData.OperationID, "pullMsgBySeqListReq err", err.Error())
+			nReply.ErrCode = 200
+			nReply.ErrMsg = err.Error()
+			ws.pullMsgResp(conn, m, nReply)
+		} else {
+			log.NewInfo(pbData.OperationID, "rpc call success to pullMsgBySeqListReq", reply.String(), reply.GetMaxSeq(), reply.GetMinSeq(), len(reply.GetSingleUserMsg()), len(reply.GetGroupUserMsg()))
+			ws.pullMsgResp(conn, m, reply)
+		}
+	} else {
+		nReply.ErrCode = errCode
+		nReply.ErrMsg = errMsg
+		ws.pullMsgResp(conn, m, nReply)
+	}
+}
+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)
 	isPass, errCode, errMsg, pData := ws.argsValidate(m, constant.WSSendMsg)
 	if isPass {
-		data := pData.(MsgData)
+		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,
-			Content:       data.Content,
-			Options:       utils.MapToJsonString(data.Options),
-			ClientMsgID:   data.ClientMsgID,
-			OffLineInfo:   utils.MapToJsonString(data.OfflineInfo),
+			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,
 		}
-		log.InfoByKv("Ws call success to sendMsgReq", m.OperationID, "Parameters", m)
+		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)
-		log.Info("", "", "api UserSendMsg call, api call rpc...")
-		reply, _ := client.UserSendMsg(context.Background(), &pbData)
-		log.Info("", "", "api UserSendMsg call end..., [data: %s] [reply: %s]", pbData.String(), reply.String())
-		ws.sendMsgResp(conn, m, reply)
+		reply, err := client.UserSendMsg(context.Background(), &pbData)
+		if err != nil {
+			log.NewError(pbData.OperationID, "UserSendMsg err", err.Error())
+			nReply.ErrCode = 200
+			nReply.ErrMsg = err.Error()
+			ws.sendMsgResp(conn, m, nReply, sendTime)
+		} else {
+			log.NewInfo(pbData.OperationID, "rpc call success to sendMsgReq", reply.String())
+			ws.sendMsgResp(conn, m, reply, sendTime)
+		}
+
 	} else {
-		reply.ErrCode = errCode
-		reply.ErrMsg = errMsg
-		ws.sendMsgResp(conn, m, reply)
+		nReply.ErrCode = errCode
+		nReply.ErrMsg = errMsg
+		ws.sendMsgResp(conn, m, nReply, sendTime)
 	}
 
 }
+func (ws *WServer) sendMsgResp(conn *UserConn, m *Req, pb *pbChat.UserSendMsgResp, sendTime int64) {
+	// := make(map[string]interface{})
 
-func (ws *WServer) sendMsg(conn *websocket.Conn, mReply map[string]interface{}) {
-	bMsg, _ := json.Marshal(mReply)
-	err := ws.writeMsg(conn, websocket.TextMessage, bMsg)
-	if err != nil {
-		log.ErrorByKv("WS WriteMsg error", "", "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn), "error", err, "mReply", mReply)
+	var mReplyData open_im_sdk.UserSendMsgResp
+	mReplyData.ClientMsgID = pb.GetClientMsgID()
+	mReplyData.ServerMsgID = pb.GetServerMsgID()
+	mReplyData.SendTime = sendTime
+	b, _ := proto.Marshal(&mReplyData)
+	mReply := Resp{
+		ReqIdentifier: m.ReqIdentifier,
+		MsgIncr:       m.MsgIncr,
+		ErrCode:       pb.GetErrCode(),
+		ErrMsg:        pb.GetErrMsg(),
+		OperationID:   m.OperationID,
+		Data:          b,
+	}
+	ws.sendMsg(conn, mReply)
+}
+
+func (ws *WServer) sendMsg(conn *UserConn, mReply interface{}) {
+	var b bytes.Buffer
+	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())
+		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())
+	}
+}
+func (ws *WServer) sendErrMsg(conn *UserConn, errCode int32, errMsg string, reqIdentifier int32, msgIncr string, operationID string) {
+	mReply := Resp{
+		ReqIdentifier: reqIdentifier,
+		MsgIncr:       msgIncr,
+		ErrCode:       errCode,
+		ErrMsg:        errMsg,
+		OperationID:   operationID,
 	}
-}
-func (ws *WServer) sendErrMsg(conn *websocket.Conn, errCode int32, errMsg string) {
-	mReply := make(map[string]interface{})
-	mReply["errCode"] = errCode
-	mReply["errMsg"] = errMsg
 	ws.sendMsg(conn, mReply)
 }

internal/msg_gateway/gate/rpc_server.go

@@ -6,14 +6,18 @@ 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"
-	"encoding/json"
+	"encoding/gob"
 	"fmt"
-	"github.com/gorilla/websocket"
-	"google.golang.org/grpc"
+	"github.com/golang/protobuf/proto"
 	"net"
 	"strings"
+
+	"github.com/gorilla/websocket"
+	"google.golang.org/grpc"
 )
 
 type RPCServer struct {
@@ -55,105 +59,87 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
 	log.InfoByKv("PushMsgToUser is arriving", in.OperationID, "args", in.String())
 	var resp []*pbRelay.SingleMsgToUser
 	var RecvID string
-	msg := make(map[string]interface{})
-	mReply := make(map[string]interface{})
-	mReply["reqIdentifier"] = constant.WSPushMsg
-	mReply["errCode"] = 0
-	mReply["errMsg"] = ""
-	msg["sendID"] = in.SendID
-	msg["recvID"] = in.RecvID
-	msg["msgFrom"] = in.MsgFrom
-	msg["contentType"] = in.ContentType
-	msg["sessionType"] = in.SessionType
-	msg["senderNickName"] = in.SenderNickName
-	msg["senderFaceUrl"] = in.SenderFaceURL
-	msg["clientMsgID"] = in.ClientMsgID
-	msg["serverMsgID"] = in.ServerMsgID
-	msg["content"] = in.Content
-	msg["seq"] = in.RecvSeq
-	msg["sendTime"] = in.SendTime
-	msg["senderPlatformID"] = in.PlatformID
-	mReply["data"] = msg
-	bMsg, _ := json.Marshal(mReply)
+	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)
+	mReply := Resp{
+		ReqIdentifier: constant.WSPushMsg,
+		OperationID:   in.OperationID,
+		Data:          msgBytes,
+	}
+	var replyBytes bytes.Buffer
+	enc := gob.NewEncoder(&replyBytes)
+	err := enc.Encode(mReply)
+	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]
 	}
-	log.InfoByKv("test", in.OperationID, "wsUserToConn", ws.wsUserToConn)
-	for key, conn := range ws.wsUserToConn {
-		UIDAndPID := strings.Split(key, " ")
-		if UIDAndPID[0] == RecvID {
-			resultCode := sendMsgToUser(conn, bMsg, in, UIDAndPID[1], UIDAndPID[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 {
+			tag = true
+			resultCode := sendMsgToUser(conn, replyBytes.Bytes(), in, UIDAndPID[1], UIDAndPID[0])
 			temp := &pbRelay.SingleMsgToUser{
 				ResultCode:     resultCode,
 				RecvID:         UIDAndPID[0],
 				RecvPlatFormID: utils.PlatformNameToID(UIDAndPID[1]),
 			}
 			resp = append(resp, temp)
+		} else {
+			temp := &pbRelay.SingleMsgToUser{
+				ResultCode:     -1,
+				RecvID:         UIDAndPID[0],
+				RecvPlatFormID: utils.PlatformNameToID(UIDAndPID[1]),
+			}
+			resp = append(resp, temp)
 		}
 	}
-	//switch in.GetContentType() {
-	//case constant.SyncSenderMsg:
-	//	log.InfoByKv("come sync", in.OperationID, "args", in.String())
-	//	RecvID = in.GetSendID()
-	//	if in.MsgFrom != constant.SysMsgType {
-	//		for key, conn := range ws.wsUserToConn {
-	//			UIDAndPID := strings.Split(key, " ")
-	//			if UIDAndPID[0] == RecvID && utils.PlatformIDToName(in.GetPlatformID()) != UIDAndPID[1] {
-	//				resultCode := sendMsgToUser(conn, bMsg, in, UIDAndPID[1], UIDAndPID[0])
-	//				temp := &pbRelay.SingleMsgToUser{
-	//					ResultCode:     resultCode,
-	//					RecvID:         UIDAndPID[0],
-	//					RecvPlatFormID: utils.PlatformNameToID(UIDAndPID[1]),
-	//				}
-	//				resp = append(resp, temp)
-	//			}
-	//
-	//		}
-	//	}
-	//default:
-	//	log.InfoByKv("not come sync", in.OperationID, "args", in.String())
-	//	switch in.SessionType {
-	//	case constant.SingleChatType:
-	//		log.InfoByKv("come single", in.OperationID, "args", in.String())
-	//		RecvID = in.GetRecvID()
-	//	case constant.GroupChatType:
-	//		RecvID = strings.Split(in.GetRecvID(), " ")[0]
-	//	default:
-	//	}
-	//	log.InfoByKv("come for range", in.OperationID, "args", in.String())
-	//
-	//	for key, conn := range ws.wsUserToConn {
-	//		UIDAndPID := strings.Split(key, " ")
-	//		if UIDAndPID[0] == RecvID {
-	//			resultCode := sendMsgToUser(conn, bMsg, in, UIDAndPID[1], UIDAndPID[0])
-	//			temp := &pbRelay.SingleMsgToUser{
-	//				ResultCode:     resultCode,
-	//				RecvID:         UIDAndPID[0],
-	//				RecvPlatFormID: utils.PlatformNameToID(UIDAndPID[1]),
-	//			}
-	//			resp = append(resp, temp)
-	//		}
-	//	}
-	//}
+	if !tag {
+		log.NewError(in.OperationID, "push err ,no matched ws conn not in map", in.String())
+	}
 	return &pbRelay.MsgToUserResp{
 		Resp: resp,
 	}, nil
 }
 
-func sendMsgToUser(conn *websocket.Conn, bMsg []byte, in *pbRelay.MsgToUserReq, RecvPlatForm, RecvID string) (ResultCode int64) {
-	err := ws.writeMsg(conn, websocket.TextMessage, bMsg)
+func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.MsgToUserReq, 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", utils.PlatformIDToName(in.PlatformID), "recvPlatform", RecvPlatForm, "args", in.String(), "recvID", RecvID)
 		ResultCode = -2
 		return ResultCode
 	} else {
-		log.InfoByKv("PushMsgToUser is success By Ws", in.OperationID, "args", in.String())
+		log.InfoByKv("PushMsgToUser is success By Ws", in.OperationID, "args", in.String(), "recvPlatForm", RecvPlatForm, "recvID", RecvID)
 		ResultCode = 0
 		return ResultCode
 	}
 
 }
+func genUidPlatformArray(uid string) (array []string) {
+	for i := 1; i <= utils.LinuxPlatformID; i++ {
+		array = append(array, uid+" "+utils.PlatformIDToName(int32(i)))
+	}
+	return array
+}

internal/msg_gateway/gate/validate.go

@@ -9,17 +9,27 @@ package gate
 import (
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/log"
-	"github.com/mitchellh/mapstructure"
+	open_im_sdk "Open_IM/pkg/proto/sdk_ws"
+	"github.com/golang/protobuf/proto"
 )
 
 type Req struct {
-	ReqIdentifier int32                  `json:"reqIdentifier" validate:"required"`
-	Token         string                 `json:"token" validate:"required"`
-	SendID        string                 `json:"sendID" validate:"required"`
-	OperationID   string                 `json:"operationID" validate:"required"`
-	MsgIncr       int32                  `json:"msgIncr" validate:"required"`
-	Data          map[string]interface{} `json:"data"`
+	ReqIdentifier int32  `json:"reqIdentifier" validate:"required"`
+	Token         string `json:"token" `
+	SendID        string `json:"sendID" validate:"required"`
+	OperationID   string `json:"operationID" validate:"required"`
+	MsgIncr       string `json:"msgIncr" validate:"required"`
+	Data          []byte `json:"data"`
 }
+type Resp struct {
+	ReqIdentifier int32  `json:"reqIdentifier"`
+	MsgIncr       string `json:"msgIncr"`
+	OperationID   string `json:"operationID"`
+	ErrCode       int32  `json:"errCode"`
+	ErrMsg        string `json:"errMsg"`
+	Data          []byte `json:"data"`
+}
+
 type SeqData struct {
 	SeqBegin int64 `mapstructure:"seqBegin" validate:"required"`
 	SeqEnd   int64 `mapstructure:"seqEnd" validate:"required"`
@@ -30,31 +40,64 @@ type MsgData struct {
 	MsgFrom     int32                  `mapstructure:"msgFrom" validate:"required"`
 	ContentType int32                  `mapstructure:"contentType" validate:"required"`
 	RecvID      string                 `mapstructure:"recvID" validate:"required"`
-	ForceList   []string               `mapstructure:"forceList" validate:"required"`
+	ForceList   []string               `mapstructure:"forceList"`
 	Content     string                 `mapstructure:"content" validate:"required"`
 	Options     map[string]interface{} `mapstructure:"options" validate:"required"`
 	ClientMsgID string                 `mapstructure:"clientMsgID" validate:"required"`
 	OfflineInfo map[string]interface{} `mapstructure:"offlineInfo" validate:"required"`
 	Ext         map[string]interface{} `mapstructure:"ext"`
 }
+type MaxSeqResp struct {
+	MaxSeq int64 `json:"maxSeq"`
+}
+type PullMessageResp struct {
+}
+type SeqListData struct {
+	SeqList []int64 `mapstructure:"seqList" validate:"required"`
+}
 
-func (ws *WServer) argsValidate(m *Req, r int32) (isPass bool, errCode int32, errMsg string, data interface{}) {
+func (ws *WServer) argsValidate(m *Req, r int32) (isPass bool, errCode int32, errMsg string, returnData interface{}) {
 	switch r {
-	case constant.WSPullMsg:
-		data = SeqData{}
 	case constant.WSSendMsg:
-		data = MsgData{}
+		data := open_im_sdk.UserSendMsgReq{}
+		if err := proto.Unmarshal(m.Data, &data); err != nil {
+			log.ErrorByKv("Decode Data struct  err", "", "err", err.Error(), "reqIdentifier", r)
+			return false, 203, err.Error(), nil
+		}
+		if err := validate.Struct(data); err != nil {
+			log.ErrorByKv("data args validate  err", "", "err", err.Error(), "reqIdentifier", r)
+			return false, 204, err.Error(), nil
+
+		}
+		return true, 0, "", data
+	case constant.WSPullMsgBySeqList:
+		data := open_im_sdk.PullMessageBySeqListReq{}
+		if err := proto.Unmarshal(m.Data, &data); err != nil {
+			log.ErrorByKv("Decode Data struct  err", "", "err", err.Error(), "reqIdentifier", r)
+			return false, 203, err.Error(), nil
+		}
+		if err := validate.Struct(data); err != nil {
+			log.ErrorByKv("data args validate  err", "", "err", err.Error(), "reqIdentifier", r)
+			return false, 204, err.Error(), nil
+
+		}
+		return true, 0, "", data
+
 	default:
 	}
-	if err := mapstructure.WeakDecode(m.Data, &data); err != nil {
-		log.ErrorByKv("map to Data struct  err", "", "err", err.Error(), "reqIdentifier", r)
-		return false, 203, err.Error(), nil
-	} else if err := validate.Struct(data); err != nil {
-		log.ErrorByKv("data args validate  err", "", "err", err.Error(), "reqIdentifier", r)
-		return false, 204, err.Error(), nil
 
-	} else {
-		return true, 0, "", data
-	}
+	return false, 204, "args err", nil
+
+	//b := bytes.NewBuffer(m.Data)
+	//dec := gob.NewDecoder(b)
+	//err := dec.Decode(&data)
+	//if err != nil {
+	//	log.ErrorByKv("Decode Data struct  err", "", "err", err.Error(), "reqIdentifier", r)
+	//	return false, 203, err.Error(), nil
+	//}
+	//if err := mapstructure.WeakDecode(m.Data, &data); err != nil {
+	//	log.ErrorByKv("map to Data struct  err", "", "err", err.Error(), "reqIdentifier", r)
+	//	return false, 203, err.Error(), nil
+	//} else
 
 }

internal/msg_gateway/gate/ws_server.go

@@ -6,23 +6,28 @@ import (
 	"Open_IM/pkg/utils"
 	"github.com/gorilla/websocket"
 	"net/http"
+	"sync"
 	"time"
 )
 
+type UserConn struct {
+	*websocket.Conn
+	w *sync.Mutex
+}
 type WServer struct {
 	wsAddr       string
 	wsMaxConnNum int
 	wsUpGrader   *websocket.Upgrader
-	wsConnToUser map[*websocket.Conn]string
-	wsUserToConn map[string]*websocket.Conn
+	wsConnToUser map[*UserConn]string
+	wsUserToConn map[string]*UserConn
 }
 
 func (ws *WServer) onInit(wsPort int) {
 	ip := utils.ServerIP
 	ws.wsAddr = ip + ":" + utils.IntToString(wsPort)
 	ws.wsMaxConnNum = config.Config.LongConnSvr.WebsocketMaxConnNum
-	ws.wsConnToUser = make(map[*websocket.Conn]string)
-	ws.wsUserToConn = make(map[string]*websocket.Conn)
+	ws.wsConnToUser = make(map[*UserConn]string)
+	ws.wsUserToConn = make(map[string]*UserConn)
 	ws.wsUpGrader = &websocket.Upgrader{
 		HandshakeTimeout: time.Duration(config.Config.LongConnSvr.WebsocketTimeOut) * time.Second,
 		ReadBufferSize:   config.Config.LongConnSvr.WebsocketMaxMsgLen,
@@ -49,35 +54,39 @@ func (ws *WServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 			//Connection mapping relationship,
 			//userID+" "+platformID->conn
 			SendID := query["sendID"][0] + " " + utils.PlatformIDToName(int32(utils.StringToInt64(query["platformID"][0])))
-			ws.addUserConn(SendID, conn)
-			go ws.readMsg(conn)
+			//Initialize a lock for each user
+			newConn := &UserConn{conn, new(sync.Mutex)}
+			ws.addUserConn(SendID, newConn)
+			go ws.readMsg(newConn)
 		}
 	}
 }
 
-func (ws *WServer) readMsg(conn *websocket.Conn) {
+func (ws *WServer) readMsg(conn *UserConn) {
 	for {
-		msgType, msg, err := conn.ReadMessage()
+		messageType, msg, err := conn.ReadMessage()
+		if messageType == websocket.PingMessage {
+			log.NewInfo("", "this is a  pingMessage")
+		}
 		if err != nil {
 			log.ErrorByKv("WS ReadMsg error", "", "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn), "error", err)
 			ws.delUserConn(conn)
 			return
 		} else {
-			log.ErrorByKv("test", "", "msgType", msgType, "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn))
+			//log.ErrorByKv("test", "", "msgType", msgType, "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn))
 		}
 		ws.msgParse(conn, msg)
 		//ws.writeMsg(conn, 1, chat)
 	}
 
 }
-
-func (ws *WServer) writeMsg(conn *websocket.Conn, a int, msg []byte) error {
-	rwLock.Lock()
-	defer rwLock.Unlock()
+func (ws *WServer) writeMsg(conn *UserConn, a int, msg []byte) error {
+	conn.w.Lock()
+	defer conn.w.Unlock()
 	return conn.WriteMessage(a, msg)
 
 }
-func (ws *WServer) addUserConn(uid string, conn *websocket.Conn) {
+func (ws *WServer) addUserConn(uid string, conn *UserConn) {
 	rwLock.Lock()
 	defer rwLock.Unlock()
 	if oldConn, ok := ws.wsUserToConn[uid]; ok {
@@ -95,7 +104,7 @@ func (ws *WServer) addUserConn(uid string, conn *websocket.Conn) {
 
 }
 
-func (ws *WServer) delUserConn(conn *websocket.Conn) {
+func (ws *WServer) delUserConn(conn *UserConn) {
 	rwLock.Lock()
 	defer rwLock.Unlock()
 	var uidPlatform string
@@ -111,12 +120,12 @@ func (ws *WServer) delUserConn(conn *websocket.Conn) {
 	}
 	err := conn.Close()
 	if err != nil {
-		log.ErrorByKv("close err", "", "uid", uidPlatform, "conn", conn)
+		log.ErrorByKv("close err", "", "uid", uidPlatform)
 	}
 
 }
 
-func (ws *WServer) getUserConn(uid string) *websocket.Conn {
+func (ws *WServer) getUserConn(uid string) *UserConn {
 	rwLock.RLock()
 	defer rwLock.RUnlock()
 	if conn, ok := ws.wsUserToConn[uid]; ok {
@@ -124,7 +133,7 @@ func (ws *WServer) getUserConn(uid string) *websocket.Conn {
 	}
 	return nil
 }
-func (ws *WServer) getUserUid(conn *websocket.Conn) string {
+func (ws *WServer) getUserUid(conn *UserConn) string {
 	rwLock.RLock()
 	defer rwLock.RUnlock()