ignore

gongshangguo

2022-03-08 a32c642ffa2b9f05e3b55fc88b2c2bed3277a080

 client/client.go

@@ -61,8 +61,6 @@
   Writer *bufio.Writer
   // 写入通道
   writeChan chan []byte
   // 退出通道
   exitChan chan int8
   // 连接状态
   state State
   // 报文头
@@ -110,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{}
@@ -213,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)))
@@ -222,6 +219,7 @@

   // 拼装并发送消息
   body := &aiot.Protocol{
      Receiver: aiot.RECEIVER_TO_SAAS,
      SenderId: senderId,
      MsgType: msgType,
      ReqType: reqType,
@@ -241,12 +239,17 @@
      }
   }()

   // 通道已关闭，不能写入
   if c.IsClosed() {
      errMsg := "Can not write msg into closed chain"
      c.Logger.Warn(errMsg, zap.Any("msg",body))
      return errors.New(errMsg)
   }
   body.Receiver = aiot.RECEIVER_TO_SAAS

   // 消息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))
@@ -262,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))
   }
@@ -286,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(""))
      }
   }
}
@@ -418,7 +421,7 @@
}

// 拼装消息ID
func (c *Client) GetMsgProto(msgId string) *aiot.MsgIdProto {
func GetMsgProto(msgId string) *aiot.MsgIdProto {
   // 新消息
   if msgId == "" {
      return &aiot.MsgIdProto{
@@ -436,6 +439,11 @@
// 获取deviceId
func (c *Client) GetDeviceId() string {
   return c.deviceId
}

// 获取连接状态
func (c *Client) GetState() State {
   return c.state
}

// 判断连接是否关闭
@@ -464,6 +472,7 @@
      if c.IsConnected() {
         c.clientCallback.OnClose(c)
      }
      close(c.writeChan)
   }

   // 设置连接属性