[嵌入式开发模块]MQTT开源库Paho嵌入式C/C++版本的移植及使用(基于W5500 io库)

小鱼儿 2022-01-13 13:23 2707阅读 0赞

### 文章目录 ###

*  前言
 *   *  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][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;
          }
        }
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpbl9zdHJvbmc_size_16_color_FFFFFF_t_70]: /images/20220113/5cfeb55f97c74e17b0176699e42ac041.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpbl9zdHJvbmc_size_16_color_FFFFFF_t_70 1]: /images/20220113/f09167eb7975425abb4fd5c076378bcd.png
[https_github.com_eclipse_paho.mqtt.embedded-c]: https://github.com/eclipse/paho.mqtt.embedded-c