From a32c642ffa2b9f05e3b55fc88b2c2bed3277a080 Mon Sep 17 00:00:00 2001
From: gongshangguo <gongshangguo@admin.com>
Date: 星期二, 08 三月 2022 10:00:30 +0800
Subject: [PATCH] ignore
---
client/client.go | 41 +++++++++++++++++++++++++++++------------
1 files changed, 29 insertions(+), 12 deletions(-)
diff --git a/client/client.go b/client/client.go
index 96bfe94..4688786 100644
--- a/client/client.go
+++ b/client/client.go
@@ -6,6 +6,7 @@
"bufio"
"encoding/binary"
"encoding/json"
+ "errors"
uuid "github.com/satori/go.uuid"
"go.uber.org/zap"
"io"
@@ -60,8 +61,6 @@
Writer *bufio.Writer
// 鍐欏叆閫氶亾
writeChan chan []byte
- // 閫�鍑洪�氶亾
- exitChan chan int8
// 杩炴帴鐘舵��
state State
// 鎶ユ枃澶�
@@ -109,7 +108,6 @@
c.readLock = new(sync.Mutex)
c.writeLock = new(sync.Mutex)
c.writeChan = make(chan []byte)
- c.exitChan = make(chan int8)
c.state = StateInit
c.tmpByte4Slice = make([]byte, 4)
c.waitGroup = &util.WaitGroupWrapper{}
@@ -212,7 +210,7 @@
}
// 鍙戦�佹秷鎭�
-func (c *Client) WriteMsg(senderId string, msgType aiot.MSG_TYPE, reqType aiot.REQ_TYPE, data []byte, msgProto *aiot.MsgIdProto) (*aiot.Protocol, error) {
+func (c *Client) writeMsg(senderId string, msgType aiot.MSG_TYPE, reqType aiot.REQ_TYPE, data []byte, msgProto *aiot.MsgIdProto) (*aiot.Protocol, error) {
// 鍏抽棴鐨勮繛鎺ヤ笉鑳藉啓鍏�
if c.IsClosed() {
c.Logger.Error("Can not write msg on the closed chan", zap.Any("msgType", msgType), zap.Any("reqType", reqType), zap.Any("data", string(data)))
@@ -221,6 +219,7 @@
// 鎷艰骞跺彂閫佹秷鎭�
body := &aiot.Protocol{
+ Receiver: aiot.RECEIVER_TO_SAAS,
SenderId: senderId,
MsgType: msgType,
ReqType: reqType,
@@ -239,7 +238,18 @@
c.Logger.Error("Write Body Error:", err)
}
}()
- body.Receiver = aiot.RECEIVER_TO_SAAS
+
+ // 閫氶亾宸插叧闂紝涓嶈兘鍐欏叆
+ if c.IsClosed() {
+ errMsg := "Can not write msg into closed chain"
+ c.Logger.Warn(errMsg, zap.Any("msg",body))
+ return errors.New(errMsg)
+ }
+
+ // 娑堟伅ID榛樿澶勭悊
+ if body.MsgProto == nil {
+ body.MsgProto = GetMsgProto("")
+ }
msgData, err := json.Marshal(body)
if err != nil {
c.Logger.Error("Fail to Marshal send data", zap.Error(err))
@@ -255,7 +265,7 @@
c.Logger.Debug("registering...")
data := c.deviceRegister
msgData, _ := json.Marshal(data)
- _, err := c.WriteMsg(c.deviceId, aiot.MSG_TYPE_REGISTER, aiot.REQ_TYPE_REQUEST, msgData, c.GetMsgProto(""))
+ _, err := c.writeMsg(c.deviceId, aiot.MSG_TYPE_REGISTER, aiot.REQ_TYPE_REQUEST, msgData, GetMsgProto(""))
if err != nil {
c.Logger.Error("Fail to send device register", zap.Any("msg", msgData))
}
@@ -279,7 +289,7 @@
t.Stop()
return
}
- go c.WriteMsg(c.deviceId, aiot.MSG_TYPE_HEART_BEAT, aiot.REQ_TYPE_REQUEST, pingData, c.GetMsgProto(""))
+ go c.writeMsg(c.deviceId, aiot.MSG_TYPE_HEART_BEAT, aiot.REQ_TYPE_REQUEST, pingData, GetMsgProto(""))
}
}
}
@@ -411,7 +421,7 @@
}
// 鎷艰娑堟伅ID
-func (c *Client) GetMsgProto(msgId string) *aiot.MsgIdProto {
+func GetMsgProto(msgId string) *aiot.MsgIdProto {
// 鏂版秷鎭�
if msgId == "" {
return &aiot.MsgIdProto{
@@ -431,6 +441,11 @@
return c.deviceId
}
+// 鑾峰彇杩炴帴鐘舵��
+func (c *Client) GetState() State {
+ return c.state
+}
+
// 鍒ゆ柇杩炴帴鏄惁鍏抽棴
func (c *Client) IsClosed() bool {
return c.state == StateDisconnected
@@ -448,7 +463,7 @@
// 鍏抽棴TCP
func (c *Client) Close() {
- c.Logger.Debug("Closing connect", zap.String("addr", c.addr))
+ c.Logger.Debug("Closing connect...", zap.String("addr", c.addr))
c.closeLock.Lock()
defer c.closeLock.Unlock()
// 鍏抽棴閫氶亾
@@ -457,8 +472,10 @@
if c.IsConnected() {
c.clientCallback.OnClose(c)
}
-
- // 璁剧疆杩炴帴灞炴��
- c.SetState(StateDisconnected)
+ close(c.writeChan)
}
+
+ // 璁剧疆杩炴帴灞炴��
+ c.SetState(StateDisconnected)
+ c.Logger.Debug("Connect closed...", zap.String("addr", c.addr))
}
--
Gitblit v1.8.0