文章目录

前言
- Paho
接口文件
- mqtt_interface.h
- mqtt_interface.c
移植
- 定时器Timer类
- 网络Network类
使用
- 初始化相关实例
- 与MQTT Server创建TCP链接
- 设置MQTT链接选项并创建MQTT链接
- 订阅Topic
- 发布Topic
- 守护
- 断连
- 综合示例

前言

最近成功使用W5500实现了MQTT客户端，进行一个记录。

W5500的io库下载下来后在Internet文件夹下有MQTT文件夹。
watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpbl9zdHJvbmc_size_16_color_FFFFFF_t_70
里头差不多就长这个样子。
其中，mqtt_interface两个文件是移植用的接口文件。而其他的文件实际上就是Paho开源库的Embedded C/C++ 版本的文件。

Paho

Paho是MQTT的官方开源库，其有很多版本，各版本之间的特性比较如下：
watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpbl9zdHJvbmc_size_16_color_FFFFFF_t_70 1
可以看到，我们要讲的嵌入式版本(最后一行)是其中特性最少的；这很正常，受限于设备的能力，肯定要精简掉一些特性。注意，虽然这张图上把非阻塞API那一项勾上了，但实际上嵌入式版本只有阻塞式API，可能这里是个笔误。

在进一步研究之前，请先下载Embedded C的最新版库；即使你用的是W5500的io库，里头已经有相关文件了，也最好更新成最新版的。稍微有点差别。
https://github.com/eclipse/paho.mqtt.embedded-c
点进去上面的链接，下载下来一堆文件，我们只需要MQTTPacket/src这个文件夹以及MQTTClient-C/src里的MQTTClient.h和MQTTClient.c。

记得在MQTTClient.h里略做修改。在
#include “MQTTPacket.h”
后面加上一行：
#include “mqtt_interface.h”

网上对C和C++版本的讲解有很多，也很全面，但是对Embedded版本的讲解却几乎没有。下面由我进行讲解。

接口文件

方便起见，我们这样：先添加进去以下两个文件，这两文件是在我的环境下移植好的，但是在你的环境下会有一点点小错误的；但是不打紧，之后我们对其进行修改以完成移植。

mqtt_interface.h

//*****************************************************************************
//! \file mqtt_interface.h
//! \brief Paho MQTT to WIZnet Chip interface Header file.
//! \details The process of porting an interface to use paho MQTT.
//! \version 1.0.0
//! \date 2016/12/06
//! \par  Revision history
//!       <2016/12/06> 1st Release
//!
//! \author Peter Bang & Justin Kim
//! \copyright
//!
//! Copyright (c)  2016, WIZnet Co., LTD.
//! All rights reserved.
//!
//! Redistribution and use in source and binary forms, with or without
//! modification, are permitted provided that the following conditions
//! are met:
//!
//!     * Redistributions of source code must retain the above copyright
//! notice, this list of conditions and the following disclaimer.
//!     * Redistributions in binary form must reproduce the above copyright
//! notice, this list of conditions and the following disclaimer in the
//! documentation and/or other materials provided with the distribution.
//!     * Neither the name of the <ORGANIZATION> nor the names of its
//! contributors may be used to endorse or promote products derived
//! from this software without specific prior written permission.
//!
//! THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
//! AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
//! IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
//! ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
//! LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
//! CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
//! SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
//! INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
//! CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
//! ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
//! THE POSSIBILITY OF SUCH DAMAGE.
//
//*****************************************************************************
/* MQTT subscribe Example.... W5500 + STM32F103(IoT board)
//Include: Board configuration
#include "IoTEVB.h"
//Include: MCU peripheral Library
#include "stm32f10x_rcc.h"
#include "stm32f10x.h"
//Include: W5500 iolibrary
#include "w5500.h"
#include "wizchip_conf.h"
#include "misc.h"
//Include: Internet iolibrary
#include "MQTTClient.h"
//Include: MCU Specific W5500 driver
#include "W5500HardwareDriver.h"
//Include: Standard IO Library
#include <stdio.h>
//Socket number defines
#define TCP_SOCKET    0
//Receive Buffer Size define
#define BUFFER_SIZE    2048
//Global variables
unsigned char targetIP[4] = {
    }; // mqtt server IP
unsigned int targetPort = 1883; // mqtt server port
uint8_t mac_address[6] = {
    };
wiz_NetInfo gWIZNETINFO = {
     .mac = {
    }, //user MAC
                            .ip = {
    }, //user IP
                            .sn = {
    },
                            .gw = {
    },
                            .dns = {
    },
                            .dhcp = NETINFO_STATIC};
unsigned char tempBuffer[BUFFER_SIZE] = {
    };
struct opts_struct
{
    char* clientid;
    int nodelimiter;
    char* delimiter;
    enum QoS qos;
    char* username;
    char* password;
    char* host;
    int port;
    int showtopics;
} opts ={
     (char*)"stdout-subscriber", 0, (char*)"\n", QOS0, NULL, NULL, targetIP, targetPort, 0 };
// @brief messageArrived callback function
void messageArrived(MessageData* md)
{
    unsigned char testbuffer[100];
    MQTTMessage* message = md->message;
    if (opts.showtopics)
    {
        memcpy(testbuffer,(char*)message->payload,(int)message->payloadlen);
        *(testbuffer + (int)message->payloadlen + 1) = "\n";
        printf("%s\r\n",testbuffer);
    }
    if (opts.nodelimiter)
        printf("%.*s", (int)message->payloadlen, (char*)message->payload);
    else
        printf("%.*s%s", (int)message->payloadlen, (char*)message->payload, opts.delimiter);
}
// @brief 1 millisecond Tick Timer setting
void NVIC_configuration(void)
{
    NVIC_InitTypeDef NVIC_InitStructure;
    SysTick_CLKSourceConfig(SysTick_CLKSource_HCLK);
    SysTick_Config(72000);
    NVIC_InitStructure.NVIC_IRQChannel = SysTick_IRQn;
    NVIC_InitStructure.NVIC_IRQChannelPreemptionPriority = 0; // Highest priority
    NVIC_InitStructure.NVIC_IRQChannelSubPriority = 0;
    NVIC_InitStructure.NVIC_IRQChannelCmd = ENABLE;
    NVIC_Init(&NVIC_InitStructure);
}
// @brief 1 millisecond Tick Timer Handler setting
void SysTick_Handler(void)
{
    MilliTimer_Handler();
}
int main(void)
{
    led_ctrl led1,led2;
    int i;
    int rc = 0;
    unsigned char buf[100];
    //Usart initialization for Debug.
    USART1Initialze();
        printf("USART initialized.\n\r");
    I2C1Initialize();
        printf("I2C initialized.\n\r");
    MACEEP_Read(mac_address,0xfa,6);
    printf("Mac address\n\r");
    for(i = 0 ; i < 6 ; i++)
    {
        printf("%02x ",mac_address[i]);
    }
    printf("\n\r");
    //LED initialization.
    led_initialize();
    led1 = led2 = ON;
    led2Ctrl(led2);
    led1Ctrl(led1);
    //W5500 initialization.
    W5500HardwareInitilize();
        printf("W5500 hardware interface initialized.\n\r");
    W5500Initialze();
        printf("W5500 IC initialized.\n\r");
    //Set network informations
    wizchip_setnetinfo(&gWIZNETINFO);
    setSHAR(mac_address);
    print_network_information();
    Network n;
    MQTTClient c;
    NewNetwork(&n, TCP_SOCKET);
    ConnectNetwork(&n, targetIP, targetPort);
    MQTTClientInit(&c,&n,1000,buf,100,tempBuffer,2048);
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    data.willFlag = 0;
    data.MQTTVersion = 3;
    data.clientID.cstring = opts.clientid;
    data.username.cstring = opts.username;
    data.password.cstring = opts.password;
    data.keepAliveInterval = 60;
    data.cleansession = 1;
    rc = MQTTConnect(&c, &data);
    printf("Connected %d\r\n", rc);
    opts.showtopics = 1;
    printf("Subscribing to %s\r\n", "hello/wiznet");
    rc = MQTTSubscribe(&c, "hello/wiznet", opts.qos, messageArrived);
    printf("Subscribed %d\r\n", rc);
    while(1)
    {
        MQTTYield(&c, data.keepAliveInterval);
    }
}
*/
#ifndef __MQTT_INTERFACE_H_
#define __MQTT_INTERFACE_H_
/*
 * @brief MQTT MilliTimer handler
 * @note MUST BE register to your system 1m Tick timer handler
 */
void MilliTimer_Handler(void);
/*
 * @brief Timer structure
 */
typedef struct Timer Timer;
struct Timer {
    unsigned long systick_period;
    unsigned long end_time;
};
/*
 * @brief Network structure
 */
typedef struct Network Network;
struct Network
{
    int my_socket;
    int (*mqttread) (Network*, unsigned char*, int, int);
    int (*mqttwrite) (Network*, unsigned char*, int, int);
    void (*disconnect) (Network*);
};
/*
 * @brief Timer function
 */
void TimerInit(Timer*);
char TimerIsExpired(Timer*);
void TimerCountdownMS(Timer*, unsigned int);
void TimerCountdown(Timer*, unsigned int);
int TimerLeftMS(Timer*);
/*
 * @brief Network interface porting
 */
int w5x00_read(Network*, unsigned char*, int, int);
int w5x00_write(Network*, unsigned char*, int, int);
void w5x00_disconnect(Network*);
void NewNetwork(Network* n, int sn);
int ConnectNetwork(Network*, char*, int);
#endif //__MQTT_INTERFACE_H_

mqtt_interface.c

//*****************************************************************************
//! \file mqtt_interface.c
//! \brief Paho MQTT to WIZnet Chip interface implement file.
//! \details The process of porting an interface to use paho MQTT.
//! \version 1.0.0
//! \date 2016/12/06
//! \par  Revision history
//!       <2016/12/06> 1st Release
//!
//! \author Peter Bang & Justin Kim
//! \copyright
//!
//! Copyright (c)  2016, WIZnet Co., LTD.
//! All rights reserved.
//!
//! Redistribution and use in source and binary forms, with or without
//! modification, are permitted provided that the following conditions
//! are met:
//!
//!     * Redistributions of source code must retain the above copyright
//! notice, this list of conditions and the following disclaimer.
//!     * Redistributions in binary form must reproduce the above copyright
//! notice, this list of conditions and the following disclaimer in the
//! documentation and/or other materials provided with the distribution.
//!     * Neither the name of the <ORGANIZATION> nor the names of its
//! contributors may be used to endorse or promote products derived
//! from this software without specific prior written permission.
//!
//! THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
//! AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
//! IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
//! ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
//! LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
//! CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
//! SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
//! INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
//! CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
//! ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
//! THE POSSIBILITY OF SUCH DAMAGE.
//
//*****************************************************************************
#include "mqtt_interface.h"
#include "socket.h"
#include "MyOS.h"
unsigned long MilliTimer;
/*
 * @brief MQTT MilliTimer handler
 * @note MUST BE register to your system 1m Tick timer handler.
 */
void MilliTimer_Handler(void) {
  MilliTimer++;
}
/*
 * @brief Timer Initialize
 * @param  timer : pointer to a Timer structure
 *         that contains the configuration information for the Timer.
 */
void TimerInit(Timer* timer) {
  timer->end_time = 0;
}
/*
 * @brief expired Timer
 * @param  timer : pointer to a Timer structure
 *         that contains the configuration information for the Timer.
 */
char TimerIsExpired(Timer* timer) {
  long left = timer->end_time - MilliTimer;
  return (left < 0);
}
/*
 * @brief Countdown millisecond Timer
 * @param  timer : pointer to a Timer structure
 *         that contains the configuration information for the Timer.
 *         timeout : setting timeout millisecond.
 */
void TimerCountdownMS(Timer* timer, unsigned int timeout) {
  timer->end_time = MilliTimer + timeout;
}
/*
 * @brief Countdown second Timer
 * @param  timer : pointer to a Timer structure
 *         that contains the configuration information for the Timer.
 *         timeout : setting timeout millisecond.
 */
void TimerCountdown(Timer* timer, unsigned int timeout) {
  timer->end_time = MilliTimer + (timeout * 1000);
}
/*
 * @brief left millisecond Timer
 * @param  timer : pointer to a Timer structure
 *         that contains the configuration information for the Timer.
 */
int TimerLeftMS(Timer* timer) {
  long left = timer->end_time - MilliTimer;
  return (left < 0) ? 0 : left;
}
/*
 * @brief New network setting
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 *         sn : socket number where x can be (0..7).
 * @retval None
 */
void NewNetwork(Network* n, int sn) {
  n->my_socket = sn;
  n->mqttread = w5x00_read;
  n->mqttwrite = w5x00_write;
  n->disconnect = w5x00_disconnect;
}
/*
 * @brief read function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 *         buffer : pointer to a read buffer.
 *         len : buffer length.
 */
int w5x00_read(Network* n, unsigned char* buffer, int len, int timeout_ms){
  Timer tmr;
  TimerInit(&tmr);
  TimerCountdownMS(&tmr, timeout_ms);
  while(!TimerIsExpired(&tmr)){
    if(getSn_SR(n->my_socket) != SOCK_ESTABLISHED)
      return -1;
    if(getSn_RX_RSR(n->my_socket)>0)
      return recv(n->my_socket, buffer, len);
    MyOS_DlyHMSM(0,0,0,30);
  }
  return 0;
}
/*
 * @brief write function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 *         buffer : pointer to a read buffer.
 *         len : buffer length.
 */
int w5x00_write(Network* n, unsigned char* buffer, int len, int timeout_ms){
    return send(n->my_socket, buffer, len);
}
/*
 * @brief disconnect function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 */
void w5x00_disconnect(Network* n){
  disconnect(n->my_socket);
}
/*
 * @brief connect network function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 *         ip : server iP.
 *         port : server port.
 */
int ConnectNetwork(Network* n, char* ip, int port)
{
  uint8_t myport = 12345;
  socket(n->my_socket,Sn_MR_TCP,myport,0);
  connect(n->my_socket,ip,port);
}

以上两个文件mqtt_interface.h和mqtt_interface.c不完全是io库自带的那个，是基于我的环境进行了一定的修改后移植完成后的版本。原来的版本有一些根本性的问题，所以不要嫌麻烦，动动小手指唱跳rap篮球+cv 一下，先替换掉原来的文件。

移植

下面我们来看看怎么把这个文件移植到你的嵌入式系统中。移植的主要工作就是为MQTT库实现一个定时器Timer类以及一个网络Network类。

定时器Timer类

打开MQTTClient.h可以看到库对Timer类的要求：

/* The Timer structure must be defined in the platform specific header,
 * and have the following functions to operate on it.  */
extern void TimerInit(Timer*);
extern char TimerIsExpired(Timer*);
extern void TimerCountdownMS(Timer*, unsigned int);
extern void TimerCountdown(Timer*, unsigned int);
extern int TimerLeftMS(Timer*);

这几个接口的作用分别是：
TimerInit:初始化一个Timer实例
TimerIsExpired:返回定时器是否超时
TimerCountdownMS:设定过多少ms超时
TimerCountdown:设定过多少s超时
TimerLeftMS:返回还剩多少ms超时

上面的接口文件已经实现了这些接口，其原理是内部用了一个计时用的全局变量，每1ms应该加1，而Timer内部成员则保存计时的起始时间和超时时间。通过内部变量与Timer内部成员的比较，算出是否超时等。这个实现与操作系统等无关，所以应该可以直接用。

但是要记得实现每1ms调用一次以下接口。比如用一个计时中断。

/*
 * @brief MQTT MilliTimer handler
 * @note MUST BE register to your system 1m Tick timer handler
 */
void MilliTimer_Handler(void);

网络Network类

同样在MQTTClient.h中可以看到库对Network类的要求：

/* The Platform specific header must define the Network and Timer structures and functions
 * which operate on them.
 *
typedef struct Network
{
    int (*mqttread)(Network*, unsigned char* read_buffer, int, int);
    int (*mqttwrite)(Network*, unsigned char* send_buffer, int, int);
} Network;*/

简单来说就是MQTT库将网络接口抽象为了Network类。我要好好夸夸这个接口设计，它非常成功地将MQTT实现与网络的具体实现解耦了。网络相关的信息由用户封装到自己定义的Network类之中，库只管在需要的时候回调Network类的几个方法来实现网络功能。如果你用纯C做过面向对象编程的话应该会比较好理解。

我们来看看Network结构体在这个移植版本中的定义：

typedef struct Network Network;
struct Network
{
    int my_socket;
    int (*mqttread) (Network*, unsigned char*, int, int);
    int (*mqttwrite) (Network*, unsigned char*, int, int);
    void (*disconnect) (Network*);
};

可以看到它在要求的结构体上还进行了一定的扩展。
第一个成员my_socket用于存储当前TCP链接的socket号，对应于io库中各个接口的第一个参数。

后面三个则定义了网络的读、写以及断连接口。MQTT库在内部会调用Network实例的对应接口，并把Network实例自身作为第一个参数传入，以实现对应的功能，而接口则可以通过实例的第一个成员来区分要操作的是哪一个socket。

于是我们可以看到NewNetwork的实现是这样的：

void NewNetwork(Network* n, int sn) {
  n->my_socket = sn;
  n->mqttread = w5x00_read;
  n->mqttwrite = w5x00_write;
  n->disconnect = w5x00_disconnect;
}

只是简单的将实例的成员和函数接口进行赋值。这个应该不需要你动。

于是接下来我们就要进行移植的最主要工作—实现w5x00_read、w5x00_write和(其实可以不实现的)w5x00_disconnect了。
我们来看下它们的实现：

/*
 * @brief read function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 *         buffer : pointer to a read buffer.
 *         len : buffer length.
 */
int w5x00_read(Network* n, unsigned char* buffer, int len, int timeout_ms){
  Timer tmr;
  TimerInit(&tmr);
  TimerCountdownMS(&tmr, timeout_ms);
  while(!TimerIsExpired(&tmr)){
    if(getSn_SR(n->my_socket) != SOCK_ESTABLISHED)
      return -1;
    if(getSn_RX_RSR(n->my_socket)>0)
      return recv(n->my_socket, buffer, len);
    MyOS_DlyHMSM(0,0,0,30);
  }
  return 0;
}
/*
 * @brief write function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 *         buffer : pointer to a read buffer.
 *         len : buffer length.
 */
int w5x00_write(Network* n, unsigned char* buffer, int len, int timeout_ms){
    return send(n->my_socket, buffer, len);
}
/*
 * @brief disconnect function
 * @param  n : pointer to a Network structure
 *         that contains the configuration information for the Network.
 */
void w5x00_disconnect(Network* n){
  disconnect(n->my_socket);
}

应该还蛮好理解的，就是要对Network n进行操作，从网络往缓冲区buffer中读入最多len个字符或写buffer中的len个字符到网络中，超时时间timeout_ms微妙。主要就是实现到W5500 io库对应接口的一个映射关系。唯一会出问题的估计就是read函数里头那个MyOS_DlyHMSM，这是我对操作系统的抽象，实现线程sleep的功能，以避免阻塞操作抢占太多系统资源，你应该根据自己的操作系统进行修改。

还有一个ConnectNetwork就不细讲了。

上面的实现默认你用的是W5500的io库，如果不是，可能需要照着进行一定的修改。

使用

使用这个嵌入式版本的MQTT Client有以下几个步骤：

初始化相关实例

static Network _mqttNetwork;
static MQTTClient _mqttClient = DefaultClient;
NewNetwork(&_mqttNetwork, MQTT_SOCKET);
MQTTClientInit(&_mqttClient,&_mqttNetwork,3000,_mqttSendBuf,sizeof(_mqttSendBuf),_mqttRecvBuf,
  sizeof(_mqttRecvBuf));

MQTTClientInit的第三个参数是命令的超时时间(ms)，剩下几个参数是发送及接收缓冲区。

与MQTT Server创建TCP链接

这一步与MQTT库无关，就是使用自己的网络栈创建一条TCP链接，可以使用Network类的方法：

ConnectNetwork(&_mqttNetwork, _serverIP, _serverPort);

设置MQTT链接选项并创建MQTT链接

创建好TCP链接后就要和MQTT服务器创建MQTT的链接了，不然相关功能都无法使用。
首先我们要设置链接选项：

static MQTTPacket_connectData _mqttConnData = MQTTPacket_connectData_initializer;
_mqttConnData.clientID.cstring = _devName;   // 设置客户端的名字
_mqttConnData.cleansession = 1;              // 清会话
// 如不使用用户名密码，直接注释下面即可
_mqttConnData.username.cstring = "用户名"; 
_mqttConnData.password.cstring = "密码";

MQTT有一个遗嘱功能可以在这里设置，如设置：

_mqttConnData.willFlag = TRUE;                      // 启用遗嘱
_mqttConnData.will.topicName.cstring = "feeldead";  // 设置遗嘱的topic名
_mqttConnData.will.qos = QOS1;                      // 遗嘱的QOS
_mqttConnData.will.message.cstring = "我断连了！";   // 遗嘱的内容

这样，当服务器检测到客户端异常断连的话，就会发送遗嘱，这样所有订阅这个主题的客户端就都能知道这个设备掉线了。当然，如果是主动发出断连请求后断掉的话就不会发遗嘱了。

设置好连接选项后就可以调用MQTTConnect来创建MQTT连接了

MQTTConnect(&_mqttClient, &_mqttConnData);

成功则返回 SUCCESS。

订阅Topic

建立连接成功后，可以订阅感兴趣的Topic：

MQTTSubscribe(&_mqttClient, "topicFilter", QOS1, messageArrived);

成功则返回 SUCCESS。最后一个选项用于注册对应Topic的处理函数，原型如下：

void messageArrived(MessageData* md);

发布Topic

建立连接成功后，可以发布Topic：

static char _sData[] = "10C";
static MQTTMessage _sensorData;
_sensorData.qos = QOS0;
_sensorData.retained = 0;  // 是否retained，如是，新发起订阅的客户端会收到最近的一个消息
_sensorData.id = 30;        // 消息id 只需要指定初始的，后面每次会自增
_sensorData.payload = _sData;
_sensorData.payloadlen = sizeof(_sData) - 1;
MQTTPublish(&_mqttClient, "topicName", &_sensorData);

成功则返回 SUCCESS。

守护

建立连接成功后，需要有进程完成例行的通信和接收消息等任务，这是通过调用MQTTYield来阻塞完成的：

MQTTYield(&_mqttClient, 3000);

第二个参数代表这次要守护多久(ms)；
如返回SUCCESS，说明这一次的守护成功；如失败，可能你得考虑重新建立连接。

断连

如需断开连接，先调用MQTTDisconnect：

MQTTDisconnect(&_mqttClient);

然后断开TCP链接。

综合示例

以下示例中的Task主动连接192.168.1.16:1883上的MQTT服务器。

设置遗嘱的TopicName为feeldead，QOS1，消息为自己的设备名。

然后以QOS1订阅主题s/temp，每次收到这个主题的消息时对相关信息进行打印。

通过设置_mqttKeepConn为FALSE来主动断开连接。

如需要发布消息时如上调用MQTTPublish(&_mqttClient, “topicName”, &_sensorData);来发布。

已略去其他关系不是很大的代码。

#include "MQTTClient.h"
……
static const char _devName[] = "device1";
static Network _mqttNetwork;
static MQTTClient _mqttClient = DefaultClient;
static MQTTPacket_connectData _mqttConnData = MQTTPacket_connectData_initializer;
static uint8_t _mqttSendBuf[100];
static uint8_t _mqttRecvBuf[100];
static uint8_t _mqttHost[4] = {
    192, 168, 1, 16};
static uint16_t _mqttPort = 1883;
static const char _sData[] = "34.12,43,1";
static BOOL _mqttKeepConn = TRUE;
static MQTTMessage _sensorData = {
  QOS0,
  0,
  0,
  1,
  _sData,
  sizeof(_sData) - 1,
};
static void _printfMessage(MessageData* md){
  char *buf = (char *)md->message->payload;
  size_t i;
  int index = 0;
  printf("Topic %.*s\n", md->topicName->lenstring.len, md->topicName->lenstring.data);
  printf("QoS: %d\r\nRetained: %d\r\nDup: %d\r\nID: %u\r\n", 
    (int)md->message->qos,
    (int)md->message->retained,
    (int)md->message->dup,
    md->message->id);
  printf("Payload: %.*s\n", md->message->payloadlen, (char *)md->message->payload);
}
void messageArrived(MessageData* md){
  _printfMessage(md);
}
static void MqttSubTask(void *p_arg){
  int rc;
  (void)p_arg;
  _mqttConnData.clientID.cstring = _devName;
  _mqttConnData.cleansession = 1;
  _mqttConnData.willFlag = TRUE;
  _mqttConnData.will.topicName.cstring = "feeldead";
  _mqttConnData.will.qos = QOS1;
  _mqttConnData.will.message.cstring = _devName;
  NewNetwork(&_mqttNetwork, MQTT_SOCKET);
  MQTTClientInit(&_mqttClient,&_mqttNetwork,3000,_mqttSendBuf,sizeof(_mqttSendBuf),_mqttRecvBuf,
    sizeof(_mqttRecvBuf));
  for(;;){
    (void)OSTimeDlyHMSM(0,0,10,0);
    if(!_mqttKeepConn || !Network_Ready())
      continue;
    if(socket(MQTT_SOCKET,Sn_MR_TCP,0 ,0x00) == MQTT_SOCKET){
      (void)printf("MQTTclient: socket inited.\r\n");
    }else{
      (void)printf("MQTTclient: socket init fail.\r\n");
      continue;
    }
    if(connect(MQTT_SOCKET, _mqttHost, _mqttPort) == SOCK_OK){
      (void)printf("MQTTclient: tcp connect bulid.\r\n");
    }else{
      (void)printf("MQTTclient: connect fail.\r\n");
      continue;
    }
    (void)printf("MQTTclient: try connect server.\r\n");
    if((rc = MQTTConnect(&_mqttClient, &_mqttConnData)) == SUCCESS){
      (void)printf("MQTTclient: connect server success.\r\n");
    }else{
      (void)printf("MQTTclient: connect server fail,%d.\r\n", rc);
      continue;
    }
    printf("MQTTclient: Subscribing to %s, Qos: %d\r\n", "s/+/temp", QOS1);
    rc = MQTTSubscribe(&_mqttClient, "s/temp", QOS1, messageArrived);
    if(rc != SUCCESS)
      printf("MQTTclient: Subscribed %d\r\n", rc);
    while(1){
      if((rc = MQTTYield(&_mqttClient, 3000)) != SUCCESS){
        printf("MQTTclient: Yield fail %d\r\n", rc);
        break;
      }
      if(!_mqttKeepConn){
        MQTTDisconnect(&_mqttClient);
        disconnect(MQTT_SOCKET);
        close(MQTT_SOCKET);
        break;
      }
    }
  }
}