1

business/partner/purchase/im/im_server.go

@@ -6,6 +6,8 @@ import (
 	"net/http"
 	"time"
 
+	"git.rosy.net.cn/jx-callback/business/jxutils"
+
 	"git.rosy.net.cn/baseapi/utils"
 
 	"git.rosy.net.cn/jx-callback/globals"
@@ -18,16 +20,21 @@ func Init() {
 	//写入全局变量
 	Setup()
 	//建立长链接
-	go Send([]byte(HeartCheckMsg))
+	jxutils.CallMsgHandlerAsync(func() {
+		MtInit()
+	}, "MtInit:"+RandString())
 
 	//启动定时器
 	PingTimer()
 
-	go WriteMessage()
+	jxutils.CallMsgHandlerAsync(func() {
+		WriteMessage()
+	}, "WriteMessage:"+RandString())
 
-	go Manager.Start()
+	jxutils.CallMsgHandlerAsync(func() {
+		Manager.Start()
+	}, "Manager Start:"+RandString())
 
-	//fmt.Printf("服务器启动成功，端口号：%s\n", CommonSetting.HttpPort)
 }
 
 func Run(w http.ResponseWriter, r *http.Request) {
@@ -54,8 +61,8 @@ func Run(w http.ResponseWriter, r *http.Request) {
 	} else {
 		clientID = temp
 	}
-	globals.SugarLogger.Debugf("Run clientID=%s", clientID)
-	clientSocket := NewClient(clientID, conn)
+
+	clientSocket := NewClient(clientID, conn, ClientTypeJx)
 
 	//读取客户端消息
 	clientSocket.Read()
@@ -69,26 +76,6 @@ func Run(w http.ResponseWriter, r *http.Request) {
 	Manager.Connect <- clientSocket
 }
 
-func StartWebSocket(conn *websocket.Conn, clientID string, err error) {
-
-	//设置读取消息大小上线
-	conn.SetReadLimit(maxMessageSize)
-
-	clientSocket := NewClient(clientID, conn)
-
-	//读取客户端消息
-	clientSocket.Read()
-
-	if err = ConnRender(conn, renderData{ClientId: clientID}); err != nil {
-		_ = conn.Close()
-		return
-	}
-
-	// 用户连接事件
-	Manager.Connect <- clientSocket
-
-}
-
 // PingTimer 定时器发送心跳
 func PingTimer() {
 	go func() {
@@ -99,16 +86,20 @@ func PingTimer() {
 		for {
 			i++
 			<-ticker.C
-			//对美团发送心跳
-			Send([]byte(HeartCheckMsg))
-
 			for clientId, conn := range Manager.AllClient() {
-				if err := conn.Socket.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil {
-					Manager.DisConnect <- conn
-					globals.SugarLogger.Debugf("发送心跳失败: %s 总连接数：%d", clientId, Manager.Count())
-				}
-				if err := ConnRender(conn.Socket, renderData{ClientId: clientId}); err != nil {
-					return
+				if conn.ClientType == ClientTypeJx {
+					if err := conn.Socket.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil {
+						Manager.DisConnect <- conn
+						globals.SugarLogger.Debugf("发送心跳失败: %s 总连接数：%d", clientId, Manager.Count())
+					}
+					if err := ConnRender(conn.Socket, renderData{ClientId: clientId}); err != nil {
+						return
+					}
+				} else {
+					if err := conn.Socket.WriteMessage(websocket.TextMessage, []byte(HeartCheckMsg)); err != nil {
+						//对美团重新建立连接
+						MtInit()
+					}
 				}
 				globals.SugarLogger.Debugf("发送心跳 clientId=%s,i=%d", clientId, i)
 			}
@@ -121,14 +112,16 @@ func WriteMessage() {
 	i := 0
 	for {
 		clientInfo := <-ToClientChan
-		//广播发送通知所有客户端
+		//广播发送通知所有京西客户端
 		i++
 		fmt.Printf("WriteMessage clientInfo=%s i=%d", utils.Format4Output(clientInfo, false), i)
 		if Manager.AllClient() != nil {
 			for _, conn := range Manager.AllClient() {
-				globals.SugarLogger.Debugf("WriteMessage conn.ClientId=%s", conn.ClientId)
-				if err := Render(conn.Socket, clientInfo.MessageId, clientInfo.Code, clientInfo.Msg, clientInfo.Data); err != nil {
-					Manager.DisConnect <- conn
+				if conn.ClientType == ClientTypeJx { //只发送给京西
+					globals.SugarLogger.Debugf("WriteMessage conn.ClientId=%s", conn.ClientId)
+					if err := Render(conn.Socket, clientInfo.MessageId, clientInfo.Code, clientInfo.Msg, clientInfo.Data); err != nil {
+						Manager.DisConnect <- conn
+					}
 				}
 			}
 		} else {
@@ -162,6 +155,8 @@ func (c *Client) Read() {
 	go func() {
 		for {
 			messageType, msg, err := c.Socket.ReadMessage()
+			//temp := string(msg)
+			//fmt.Print(temp)
 			if err != nil {
 				if messageType == -1 && websocket.IsCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
 					Manager.DisConnect <- c
@@ -169,10 +164,8 @@ func (c *Client) Read() {
 				} else if messageType != websocket.PingMessage {
 					return
 				}
-			} else {
-				SendToVendor(msg)
-				return
 			}
+			SendToVendor(msg)
 		}
 	}()
 }
@@ -197,10 +190,11 @@ func (manager *ClientManager) EventDisconnect(client *Client) {
 //以下为客户端Client操作*******************************************
 
 // NewClient 初始化Client
-func NewClient(clientId string, socket *websocket.Conn) *Client {
+func NewClient(clientId string, socket *websocket.Conn, clientType string) *Client {
 	return &Client{
 		ClientId:    clientId,
 		Socket:      socket,
+		ClientType:  clientType,
 		ConnectTime: uint64(time.Now().Unix()),
 		IsDeleted:   false,
 	}