IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-04 09:05:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

change send message by ws and optimization

Browse Source
This commit is contained in:
Gordon
2021-10-21 19:10:55 +08:00
parent c8155553b8
commit 098c7cfa9e
24 changed files with 603 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/msg_gateway/gate/logic.go
+83 -25
View File
@@ -9,11 +9,13 @@ import (
"Open_IM/src/utils"
"context"
"encoding/json"
"fmt"
"github.com/gorilla/websocket"
"runtime"
"strings"
)
func (ws *WServer) msgParse(conn *websocket.Conn, jsonMsg []byte) {
func (ws *WServer) msgParse(conn *UserConn, jsonMsg []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}
@@ -23,32 +25,43 @@ func (ws *WServer) msgParse(conn *websocket.Conn, jsonMsg []byte) {
m := Req{}
if err := json.Unmarshal(jsonMsg, &m); 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)
return
}
if !utils.VerifyToken(m.Token, m.SendID) {
ws.sendErrMsg(conn, 202, "token validate err")
ws.sendErrMsg(conn, 202, "token validate err", m.ReqIdentifier, m.MsgIncr)
return
}
fmt.Println("test fmt Basic Info Authentication Success", m.OperationID, "reqIdentifier", m.ReqIdentifier, "sendID", m.SendID)
log.InfoByKv("Basic Info Authentication Success", m.OperationID, "reqIdentifier", m.ReqIdentifier, "sendID", m.SendID)
switch m.ReqIdentifier {
case constant.WSGetNewestSeq:
ws.newestSeqReq(conn, &m)
go ws.newestSeqReq(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) {
func (ws *WServer) newestSeqResp(conn *UserConn, m *Req, pb *pbChat.GetNewSeqResp) {
mReply := make(map[string]interface{})
mData := make(map[string]interface{})
mReply["reqIdentifier"] = m.ReqIdentifier
@@ -59,7 +72,7 @@ func (ws *WServer) newestSeqResp(conn *websocket.Conn, m *Req, pb *pbChat.GetNew
mReply["data"] = mData
ws.sendMsg(conn, mReply)
}
func (ws *WServer) newestSeqReq(conn *websocket.Conn, m *Req) {
func (ws *WServer) newestSeqReq(conn *UserConn, m *Req) {
log.InfoByKv("Ws call success to getNewSeq", m.OperationID, "Parameters", m)
pbData := pbChat.GetNewSeqReq{}
pbData.UserID = m.SendID
@@ -79,7 +92,7 @@ func (ws *WServer) newestSeqReq(conn *websocket.Conn, m *Req) {
}
func (ws *WServer) pullMsgResp(conn *websocket.Conn, m *Req, pb *pbChat.PullMessageResp) {
func (ws *WServer) pullMsgResp(conn *UserConn, m *Req, pb *pbChat.PullMessageResp) {
mReply := make(map[string]interface{})
msg := make(map[string]interface{})
mReply["reqIdentifier"] = m.ReqIdentifier
@@ -104,7 +117,7 @@ func (ws *WServer) pullMsgResp(conn *websocket.Conn, m *Req, pb *pbChat.PullMess
}
func (ws *WServer) pullMsgReq(conn *websocket.Conn, m *Req) {
func (ws *WServer) pullMsgReq(conn *UserConn, m *Req) {
log.InfoByKv("Ws call success to pullMsgReq", m.OperationID, "Parameters", m)
reply := new(pbChat.PullMessageResp)
isPass, errCode, errMsg, data := ws.argsValidate(m, constant.WSPullMsg)
@@ -130,21 +143,55 @@ func (ws *WServer) pullMsgReq(conn *websocket.Conn, m *Req) {
ws.pullMsgResp(conn, m, reply)
}
}
func (ws *WServer) pullMsgBySeqListReq(conn *UserConn, m *Req) {
log.NewError(m.OperationID, "Ws call success to pullMsgBySeqListReq", m)
reply := new(pbChat.PullMessageResp)
isPass, errCode, errMsg, data := ws.argsValidate(m, constant.WSPullMsgBySeqList)
if isPass {
pbData := pbChat.PullMessageBySeqListReq{}
pbData.SeqList = data.(SeqListData).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())
return
}
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 {
reply.ErrCode = errCode
reply.ErrMsg = errMsg
ws.pullMsgResp(conn, m, reply)
}
}
func (ws *WServer) sendMsgResp(conn *websocket.Conn, m *Req, pb *pbChat.UserSendMsgResp) {
mReply := make(map[string]interface{})
func (ws *WServer) sendMsgResp(conn *UserConn, m *Req, pb *pbChat.UserSendMsgResp, sendTime int64) {
// := 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()
//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
mReplyData["sendTime"] = utils.Int64ToString(sendTime)
//mReply["data"] = mReplyData
mReply := Resp{
ReqIdentifier: m.ReqIdentifier,
MsgIncr: m.MsgIncr,
ErrCode: pb.GetErrCode(),
ErrMsg: pb.GetErrMsg(),
OperationID: m.OperationID,
Data: mReplyData,
}
fmt.Println("test fmt send msg resp", m.OperationID, "reqIdentifier", m.ReqIdentifier, "sendID", m.SendID)
ws.sendMsg(conn, mReply)
}
func (ws *WServer) sendMsgReq(conn *websocket.Conn, m *Req) {
func (ws *WServer) sendMsgReq(conn *UserConn, m *Req, sendTime int64) {
log.InfoByKv("Ws call success to sendMsgReq", m.OperationID, "Parameters", m)
reply := new(pbChat.UserSendMsgResp)
isPass, errCode, errMsg, pData := ws.argsValidate(m, constant.WSSendMsg)
@@ -165,32 +212,43 @@ func (ws *WServer) sendMsgReq(conn *websocket.Conn, m *Req) {
Options: utils.MapToJsonString(data.Options),
ClientMsgID: data.ClientMsgID,
OffLineInfo: utils.MapToJsonString(data.OfflineInfo),
SendTime: sendTime,
}
log.InfoByKv("Ws call success to sendMsgReq", m.OperationID, "Parameters", m)
time := utils.GetCurrentTimestampBySecond()
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("", "", "ws UserSendMsg call, api call rpc...")
reply, err := client.UserSendMsg(context.Background(), &pbData)
if err != nil {
log.NewError(pbData.OperationID, "UserSendMsg err", err.Error())
reply.ErrCode = 100
reply.ErrMsg = "rpc err"
}
log.NewInfo(pbData.OperationID, "sendMsgReq call rpc cost time ", utils.GetCurrentTimestampBySecond()-time)
log.Info("", "", "api UserSendMsg call end..., [data: %s] [reply: %s]", pbData.String(), reply.String())
ws.sendMsgResp(conn, m, reply)
ws.sendMsgResp(conn, m, reply, sendTime)
log.NewInfo(pbData.OperationID, "sendMsgResp end cost time ", utils.GetCurrentTimestampBySecond()-time)
} else {
reply.ErrCode = errCode
reply.ErrMsg = errMsg
ws.sendMsgResp(conn, m, reply)
ws.sendMsgResp(conn, m, reply, sendTime)
}
}
func (ws *WServer) sendMsg(conn *websocket.Conn, mReply map[string]interface{}) {
func (ws *WServer) sendMsg(conn *UserConn, mReply 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)
}
}
func (ws *WServer) sendErrMsg(conn *websocket.Conn, errCode int32, errMsg string) {
func (ws *WServer) sendErrMsg(conn *UserConn, errCode int32, errMsg string, reqIdentifier int32, msgIncr string) {
mReply := make(map[string]interface{})
mReply["errCode"] = errCode
mReply["errMsg"] = errMsg
mReply["reqIdentifier"] = reqIdentifier
mReply["msgIncr"] = msgIncr
ws.sendMsg(conn, mReply)
}
src/msg_gateway/gate/rpc_server.go
+32 -5
View File
@@ -81,10 +81,12 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
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 {
var tag bool
a := genUidPlatformArray(RecvID)
for _, v := range a {
if conn := ws.getUserConn(v); conn != nil {
UIDAndPID := strings.Split(v, " ")
tag = true
resultCode := sendMsgToUser(conn, bMsg, in, UIDAndPID[1], UIDAndPID[0])
temp := &pbRelay.SingleMsgToUser{
ResultCode: resultCode,
@@ -94,6 +96,25 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
resp = append(resp, temp)
}
}
if !tag {
log.NewError(in.OperationID, "push err ,ws conn not in map", in.String())
}
//for key, conn := range ws.wsUserToConn {
// UIDAndPID := strings.Split(key, " ")
// if UIDAndPID[0] == RecvID {
// tag = true
// 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 ,ws conn not in map", in.String())
//}
//switch in.GetContentType() {
//case constant.SyncSenderMsg:
// log.InfoByKv("come sync", in.OperationID, "args", in.String())
@@ -143,7 +164,7 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
}, nil
}
func sendMsgToUser(conn *websocket.Conn, bMsg []byte, in *pbRelay.MsgToUserReq, RecvPlatForm, RecvID string) (ResultCode int64) {
func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.MsgToUserReq, RecvPlatForm, RecvID string) (ResultCode int64) {
err := ws.writeMsg(conn, websocket.TextMessage, bMsg)
if err != nil {
log.ErrorByKv("PushMsgToUser is failed By Ws", "", "Addr", conn.RemoteAddr().String(),
@@ -157,3 +178,9 @@ func sendMsgToUser(conn *websocket.Conn, bMsg []byte, in *pbRelay.MsgToUserReq,
}
}
func genUidPlatformArray(uid string) (array []string) {
for i := 1; i <= utils.LinuxPlatformID; i++ {
array = append(array, uid+" "+utils.PlatformIDToName(int32(i)))
}
return array
}
src/msg_gateway/gate/validate.go
+21 -7
View File
@@ -13,13 +13,22 @@ import (
)
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" validate:"required"`
SendID string `json:"sendID" validate:"required"`
OperationID string `json:"operationID" validate:"required"`
MsgIncr string `json:"msgIncr" validate:"required"`
Data interface{} `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 interface{} `json:"data"`
}
type SeqData struct {
SeqBegin int64 `mapstructure:"seqBegin" validate:"required"`
SeqEnd int64 `mapstructure:"seqEnd" validate:"required"`
@@ -30,13 +39,16 @@ 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 SeqListData struct {
SeqList []int64 `mapstructure:"seqList" validate:"required"`
}
func (ws *WServer) argsValidate(m *Req, r int32) (isPass bool, errCode int32, errMsg string, data interface{}) {
switch r {
@@ -44,6 +56,8 @@ func (ws *WServer) argsValidate(m *Req, r int32) (isPass bool, errCode int32, er
data = SeqData{}
case constant.WSSendMsg:
data = MsgData{}
case constant.WSPullMsgBySeqList:
data = SeqListData{}
default:
}
if err := mapstructure.WeakDecode(m.Data, &data); err != nil {
src/msg_gateway/gate/ws_server.go
+24 -18
View File
@@ -6,23 +6,28 @@ import (
"Open_IM/src/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,36 @@ 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()
_, msg, err := conn.ReadMessage()
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 +101,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 +117,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 +130,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()