写入消息锁

gongshangguo

2022-03-01 45def8688e8caee3707340ac1b52000950df769e

 client/client.go

@@ -197,7 +197,11 @@
      case bodyByte := <- c.writeChan:
         binary.BigEndian.PutUint32(byte4, uint32(len(bodyByte)))
         body = append(byte4, bodyByte...)
         c.closeLock.Lock()
         c.msgLock.Lock()
         _,err = c.Conn.Write(body)
         c.closeLock.Unlock()
         c.msgLock.Unlock()
         if err != nil {
            c.Logger.Error("Fail to write message", zap.Error(err), zap.String("msg", string(bodyByte)))
            c.Close()
@@ -216,13 +220,14 @@
func (c *Client) WriteMsg(senderId string, msgType aiot.MSG_TYPE, reqType aiot.REQ_TYPE, data []byte, msgProto *aiot.MsgIdProto) (*aiot.Protocol, error) {
   // 锁
   c.closeLock.Lock()
   defer c.closeLock.Unlock()

   // 关闭的连接不能写入
   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)))
      c.closeLock.Unlock()
      return nil,nil
   }
   c.closeLock.Unlock()

   // 拼装并发送消息
   body := &aiot.Protocol{
@@ -247,8 +252,10 @@
      return err
   }
   c.msgLock.Lock()
   defer c.msgLock.Unlock()
   c.closeLock.Lock()
   c.writeChan <- msgData
   c.closeLock.Unlock()
   c.msgLock.Unlock()
   return nil
}

@@ -286,6 +293,29 @@
   }
}

// 发送业务包请求
func (c *Client) Request(receiver aiot.RECEIVER, senderId string, msgProto *aiot.MsgIdProto, data []byte) error {
   body := &aiot.Protocol{}
   body.Receiver = receiver
   body.SenderId = senderId
   body.MsgProto = msgProto
   body.MsgType = aiot.MSG_TYPE_BUSINESS
   body.ReqType = aiot.REQ_TYPE_REQUEST
   body.Data = data
   c.Logger.Debug("Send msg...", zap.Any("msg", body), zap.Any("msg", body), zap.Any("reqType", body.ReqType), zap.Any("msgType", body.MsgType))
   msgData, err := json.Marshal(body)
   if err != nil {
      c.Logger.Error("Fail to Marshal send data", zap.Error(err))
      return err
   }
   c.msgLock.Lock()
   c.closeLock.Lock()
   c.writeChan <- msgData
   c.closeLock.Unlock()
   c.msgLock.Unlock()
   return nil
}

// 消息读取通道
func (c *Client) readLoop() {
   var err error