STM32移植 MQTT

1万 · 2019-11-12 16:31

STM32移植 MQTT

1万 · 2019-11-12 16:32

下载MQTT的库：Paho https://github.com/eclipse/paho.mqtt.embedded-c

1万 · 2019-11-12 16:33

1万 · 2019-11-12 16:33

1万 · 2019-11-12 16:33

MQTTPacket\samples下的transport.c、transport.h添加到工程。

1万 · 2019-11-12 16:34

//接收数据,recv函数仅仅是copy数据，真正的接收数据是协议来完成的）， recv函数返回其实际copy的字节数。
int transport_getdata(unsigned char* buf, int count)
{
      int rc = 0;
   u16 len = 0;
   u8 *data = 0;
      while(1)
      {
         if(UART7_RX_STA&0X8000)    //接收到一次数据了
         {
            memcpy(buf, (void*)&UART7_RX_BUF[read_buf_len], count);
            read_buf_len += count;
            break;
         }
         delay_ms(10);
      }
      return count;
}

1万 · 2019-11-12 16:36

在需要调用的地方实现订阅发布和读取订阅。
   //打开网络连接
mysock = transport_open(host, port);
if(mysock < 0)
      return mysock;
printf("Sending to hostname %s port %d\n", host, port);
//现在将连接字符串格式化一下，现在还没有发送
len = MQTTSerialize_connect(buf, buflen, &data);
printf("%s",buf);
//现在是数据的发送
rc = transport_sendPacketBuffer(mysock, buf, len);
/* 等待connack *///发送后接收服务器返回的数据，这里使用了一个函数的指针，要定义这个指针
if (MQTTPacket_read(buf, buflen, transport_getdata) == CONNACK)//CONNACK – 确认连接请求
{
      unsigned char sessionPresent, connack_rc
      if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0)
      {
         printf("Unable to connect, return code %d\n", connack_rc);
         goto exit;
      }
}
else
      goto exit;
/* 订阅 */
topicString.cstring = "substopic";
len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos);
rc = transport_sendPacketBuffer(mysock, buf, len);
//等待服务器答复
//SUBACK 订阅确认报文包含一个返回码清单，它们指定了 SUBSCRIBE 请求的每个订阅被授予的最大 QoS 等级。
if (MQTTPacket_read(buf, buflen, transport_getdata) == SUBACK)  /* 等待 suback */
{
      unsigned short submsgid;
      int subcount;
      int granted_qos;
      rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen);
      if (granted_qos != 0)
      {
         printf("granted qos != 0, %d\n", granted_qos);
         goto exit;
      }
}
else
      goto exit;
/* 循环获取消息 on subscribed topic */
//  topicString.cstring = "pubtopic";
//  while (1)
{
      /* transport_getdata() has a built-in 1 second timeout,
      your mileage will vary */
      if (MQTTPacket_read(buf, buflen, transport_getdata) == PUBLISH)//qos 响应报文，当qos>0时才有响应
      {
         unsigned char dup;
         int qos;
         unsigned char retained;
         unsigned short msgid;
         int payloadlen_in;
         unsigned char* payload_in;
         int rc;
         MQTTString receivedTopic;
         rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic,
                  &payload_in, &payloadlen_in, buf, buflen);
         printf("message arrived %.*s\n", payloadlen_in, payload_in); //消息到达
      }
      printf("publishing reading\n");//读取发布
   //下面两行是用来发布消息。这里发布，上面订阅，就形成了一个循环。
      len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, (unsigned char*)payload, payloadlen);
      rc = transport_sendPacketBuffer(mysock, buf, len);
      delay_ms(1000);
}
while(1)
{
      len = MQTTSerialize_pingreq(buf, buflen);//发送心跳
      rc = transport_sendPacketBuffer(mysock, buf, len);//发送
      delay_s(55);
      printf("心跳");
}
exit:
transport_close(mysock);