2023_open_source_contest_preliminary_1st_issue2 from Huang_Chen

it is OK

APP_Framework/Applications/app_test/Makefile

@@ -24,7 +24,7 @@ endif
 
 ifeq ($(CONFIG_ADD_XIZI_FEATURES),y)
     SRC_FILES := test_shell.c
-
+    
     ifeq ($(CONFIG_USER_TEST_ADC),y)
         SRC_FILES += test_adc.c
     endif
@@ -130,6 +130,7 @@ ifeq ($(CONFIG_ADD_XIZI_FEATURES),y)
     endif
 
     ifeq ($(CONFIG_USER_TEST_MQTTCLIENT),y)
+        SRC_DIR:= test_mqttclient
         SRC_FILES += 
     endif
 

APP_Framework/Applications/app_test/test_mqttclient/Makefile

@@ -0,0 +1,5 @@
+SRC_FILES += test_mqttclient.c
+			
+
+SRC_DIR += mqtt
+include $(KERNEL_ROOT)/compiler.mk

APP_Framework/Applications/app_test/test_mqttclient/README.md

@@ -0,0 +1,55 @@
+## 1. 简介
+
+本代码实现了MQTT对服务器订阅主体并发送信息功能
+
+## 2. 数据结构设计说明
+
+### 2.1 MQTT数据结构定义
+
+![image-20231007142702891](./imgs/图片4.png)
+
+### 2.2 数据解析与加密
+
+![image-20231007143235579](./imgs/图片5.png)
+
+### 2.3 MQTT连接与订阅
+
+![image-20231007143448685](./imgs/图片6.png)
+
+![image-20231007143607579](./imgs/图片7.png)
+
+## 3. 测试程序说明
+
+MQTT基于TCP/IP协议，分为客户端与服务端。在本任务中，ARM终端作为客户端，与服务端连接并进行主题订阅。
+
+以下代码设置了进行订阅时需要的Client ID、用户名、密码及订阅主题。
+
+![image-20231007144555989](./imgs/图片8.png)
+
+客户端首先需要根据以上信息向服务端发起连接请求，验证用户名及密码
+
+![image-20231007144710185](./imgs/图片9.png)
+
+随后进行主题订阅，接收服务端发布的消息
+
+![image-20231007144823940](./imgs/图片10.png)
+
+在订阅过程中需要对获取的信息进行读取，同时每隔一段时间向服务端发送保持活性ping请求，以维持与服务端的连接
+
+![image-20231007144921430](./imgs/图片11.png)
+
+## 4. 测试流程
+
+测试流程为：
+
+首先执行setip命令，设置设备ip地址
+
+![图片2](./imgs/图片2.png)
+
+随后执行“MqttSocketRecvTest 服务器ip”命令，订阅主题，然后在服务器端发布消息如图所示
+
+![图片1](./imgs/pic1.png)
+
+最终设备端接收到信息
+
+![图片3](./imgs/pic2.png)

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTConnect.h

@@ -0,0 +1,137 @@
+/*******************************************************************************
+ * Copyright (c) 2014, 2017 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Xiang Rong - 442039 Add makefile to Embedded C client
+ *    Ian Craggs - fix for issue #64, bit order in connack response
+ *******************************************************************************/
+
+#ifndef MQTTCONNECT_H_
+#define MQTTCONNECT_H_
+
+#if !defined(DLLImport)
+  #define DLLImport
+#endif
+#if !defined(DLLExport)
+  #define DLLExport
+#endif
+
+
+typedef union
+{
+	unsigned char all;	/**< all connect flags */
+#if defined(REVERSED)
+	struct
+	{
+		unsigned int username : 1;			/**< 3.1 user name */
+		unsigned int password : 1; 			/**< 3.1 password */
+		unsigned int willRetain : 1;		/**< will retain setting */
+		unsigned int willQoS : 2;				/**< will QoS value */
+		unsigned int will : 1;			    /**< will flag */
+		unsigned int cleansession : 1;	  /**< clean session flag */
+		unsigned int : 1;	  	          /**< unused */
+	} bits;
+#else
+	struct
+	{
+		unsigned int : 1;	     					/**< unused */
+		unsigned int cleansession : 1;	  /**< cleansession flag */
+		unsigned int will : 1;			    /**< will flag */
+		unsigned int willQoS : 2;				/**< will QoS value */
+		unsigned int willRetain : 1;		/**< will retain setting */
+		unsigned int password : 1; 			/**< 3.1 password */
+		unsigned int username : 1;			/**< 3.1 user name */
+	} bits;
+#endif
+} MQTTConnectFlags;	/**< connect flags byte */
+
+
+
+/**
+ * Defines the MQTT "Last Will and Testament" (LWT) settings for
+ * the connect packet.
+ */
+typedef struct
+{
+	/** The eyecatcher for this structure.  must be MQTW. */
+	char struct_id[4];
+	/** The version number of this structure.  Must be 0 */
+	int struct_version;
+	/** The LWT topic to which the LWT message will be published. */
+	MQTTString topicName;
+	/** The LWT payload. */
+	MQTTString message;
+	/**
+      * The retained flag for the LWT message (see MQTTAsync_message.retained).
+      */
+	unsigned char retained;
+	/**
+      * The quality of service setting for the LWT message (see
+      * MQTTAsync_message.qos and @ref qos).
+      */
+	char qos;
+} MQTTPacket_willOptions;
+
+
+#define MQTTPacket_willOptions_initializer { {'M', 'Q', 'T', 'W'}, 0, {NULL, {0, NULL}}, {NULL, {0, NULL}}, 0, 0 }
+
+
+typedef struct
+{
+	/** The eyecatcher for this structure.  must be MQTC. */
+	char struct_id[4];
+	/** The version number of this structure.  Must be 0 */
+	int struct_version;
+	/** Version of MQTT to be used.  3 = 3.1 4 = 3.1.1
+	  */
+	unsigned char MQTTVersion;
+	MQTTString clientID;
+	unsigned short keepAliveInterval;
+	unsigned char cleansession;
+	unsigned char willFlag;
+	MQTTPacket_willOptions will;
+	MQTTString username;
+	MQTTString password;
+} MQTTPacket_connectData;
+
+typedef union
+{
+	unsigned char all;	/**< all connack flags */
+#if defined(REVERSED)
+	struct
+	{
+    unsigned int reserved : 7;	  	    /**< unused */
+		unsigned int sessionpresent : 1;    /**< session present flag */
+	} bits;
+#else
+	struct
+	{
+		unsigned int sessionpresent : 1;    /**< session present flag */
+    unsigned int reserved: 7;	     			/**< unused */
+	} bits;
+#endif
+} MQTTConnackFlags;	/**< connack flags byte */
+
+#define MQTTPacket_connectData_initializer { {'M', 'Q', 'T', 'C'}, 0, 4, {NULL, {0, NULL}}, 60, 1, 0, \
+		MQTTPacket_willOptions_initializer, {NULL, {0, NULL}}, {NULL, {0, NULL}} }
+
+DLLExport int MQTTSerialize_connect(unsigned char* buf, int buflen, MQTTPacket_connectData* options);
+DLLExport int MQTTDeserialize_connect(MQTTPacket_connectData* data, unsigned char* buf, int len);
+
+DLLExport int MQTTSerialize_connack(unsigned char* buf, int buflen, unsigned char connack_rc, unsigned char sessionPresent);
+DLLExport int MQTTDeserialize_connack(unsigned char* sessionPresent, unsigned char* connack_rc, unsigned char* buf, int buflen);
+
+DLLExport int MQTTSerialize_disconnect(unsigned char* buf, int buflen);
+DLLExport int MQTTSerialize_pingreq(unsigned char* buf, int buflen);
+
+#endif /* MQTTCONNECT_H_ */

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTConnectClient.c

@@ -0,0 +1,214 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "MQTTPacket.h"
+#include "StackTrace.h"
+
+#include <string.h>
+
+/**
+  * Determines the length of the MQTT connect packet that would be produced using the supplied connect options.
+  * @param options the options to be used to build the connect packet
+  * @return the length of buffer needed to contain the serialized version of the packet
+  */
+int MQTTSerialize_connectLength(MQTTPacket_connectData* options)
+{
+	int len = 0;
+
+	FUNC_ENTRY;
+
+	if (options->MQTTVersion == 3)
+		len = 12; /* variable depending on MQTT or MQIsdp */
+	else if (options->MQTTVersion == 4)
+		len = 10;
+
+	len += MQTTstrlen(options->clientID)+2;
+	if (options->willFlag)
+		len += MQTTstrlen(options->will.topicName)+2 + MQTTstrlen(options->will.message)+2;
+	if (options->username.cstring || options->username.lenstring.data)
+		len += MQTTstrlen(options->username)+2;
+	if (options->password.cstring || options->password.lenstring.data)
+		len += MQTTstrlen(options->password)+2;
+
+	FUNC_EXIT_RC(len);
+	return len;
+}
+
+
+/**
+  * Serializes the connect options into the buffer.
+  * @param buf the buffer into which the packet will be serialized
+  * @param len the length in bytes of the supplied buffer
+  * @param options the options to be used to build the connect packet
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_connect(unsigned char* buf, int buflen, MQTTPacket_connectData* options)
+{
+	unsigned char *ptr = buf;
+	MQTTHeader header = {0};
+	MQTTConnectFlags flags = {0};
+	int len = 0;
+	int rc = -1;
+
+	FUNC_ENTRY;
+	if (MQTTPacket_len(len = MQTTSerialize_connectLength(options)) > buflen)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+
+	header.byte = 0;
+	header.bits.type = CONNECT;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, len); /* write remaining length */
+
+	if (options->MQTTVersion == 4)
+	{
+		writeCString(&ptr, "MQTT");
+		writeChar(&ptr, (char) 4);
+	}
+	else
+	{
+		writeCString(&ptr, "MQIsdp");
+		writeChar(&ptr, (char) 3);
+	}
+
+	flags.all = 0;
+	flags.bits.cleansession = options->cleansession;
+	flags.bits.will = (options->willFlag) ? 1 : 0;
+	if (flags.bits.will)
+	{
+		flags.bits.willQoS = options->will.qos;
+		flags.bits.willRetain = options->will.retained;
+	}
+
+	if (options->username.cstring || options->username.lenstring.data)
+		flags.bits.username = 1;
+	if (options->password.cstring || options->password.lenstring.data)
+		flags.bits.password = 1;
+
+	writeChar(&ptr, flags.all);
+	writeInt(&ptr, options->keepAliveInterval);
+	writeMQTTString(&ptr, options->clientID);
+	if (options->willFlag)
+	{
+		writeMQTTString(&ptr, options->will.topicName);
+		writeMQTTString(&ptr, options->will.message);
+	}
+	if (flags.bits.username)
+		writeMQTTString(&ptr, options->username);
+	if (flags.bits.password)
+		writeMQTTString(&ptr, options->password);
+
+	rc = ptr - buf;
+
+	exit: FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Deserializes the supplied (wire) buffer into connack data - return code
+  * @param sessionPresent the session present flag returned (only for MQTT 3.1.1)
+  * @param connack_rc returned integer value of the connack return code
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param len the length in bytes of the data in the supplied buffer
+  * @return error code.  1 is success, 0 is failure
+  */
+int MQTTDeserialize_connack(unsigned char* sessionPresent, unsigned char* connack_rc, unsigned char* buf, int buflen)
+{
+	MQTTHeader header = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = NULL;
+	int rc = 0;
+	int mylen;
+	MQTTConnackFlags flags = {0};
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	if (header.bits.type != CONNACK)
+		goto exit;
+
+	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */
+	enddata = curdata + mylen;
+	if (enddata - curdata < 2)
+		goto exit;
+
+	flags.all = readChar(&curdata);
+	*sessionPresent = flags.bits.sessionpresent;
+	*connack_rc = readChar(&curdata);
+
+	rc = 1;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Serializes a 0-length packet into the supplied buffer, ready for writing to a socket
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer, to avoid overruns
+  * @param packettype the message type
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_zero(unsigned char* buf, int buflen, unsigned char packettype)
+{
+	MQTTHeader header = {0};
+	int rc = -1;
+	unsigned char *ptr = buf;
+
+	FUNC_ENTRY;
+	if (buflen < 2)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+	header.byte = 0;
+	header.bits.type = packettype;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, 0); /* write remaining length */
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Serializes a disconnect packet into the supplied buffer, ready for writing to a socket
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer, to avoid overruns
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_disconnect(unsigned char* buf, int buflen)
+{
+	return MQTTSerialize_zero(buf, buflen, DISCONNECT);
+}
+
+
+/**
+  * Serializes a disconnect packet into the supplied buffer, ready for writing to a socket
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer, to avoid overruns
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_pingreq(unsigned char* buf, int buflen)
+{
+	return MQTTSerialize_zero(buf, buflen, PINGREQ);
+}

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTConnectServer.c

@@ -0,0 +1,148 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "StackTrace.h"
+#include "MQTTPacket.h"
+#include <string.h>
+
+#define min(a, b) ((a < b) ? a : b)
+
+
+/**
+  * Validates MQTT protocol name and version combinations
+  * @param protocol the MQTT protocol name as an MQTTString
+  * @param version the MQTT protocol version number, as in the connect packet
+  * @return correct MQTT combination?  1 is true, 0 is false
+  */
+int MQTTPacket_checkVersion(MQTTString* protocol, int version)
+{
+	int rc = 0;
+
+	if (version == 3 && memcmp(protocol->lenstring.data, "MQIsdp",
+			min(6, protocol->lenstring.len)) == 0)
+		rc = 1;
+	else if (version == 4 && memcmp(protocol->lenstring.data, "MQTT",
+			min(4, protocol->lenstring.len)) == 0)
+		rc = 1;
+	return rc;
+}
+
+
+/**
+  * Deserializes the supplied (wire) buffer into connect data structure
+  * @param data the connect data structure to be filled out
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param len the length in bytes of the data in the supplied buffer
+  * @return error code.  1 is success, 0 is failure
+  */
+int MQTTDeserialize_connect(MQTTPacket_connectData* data, unsigned char* buf, int len)
+{
+	MQTTHeader header = {0};
+	MQTTConnectFlags flags = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = &buf[len];
+	int rc = 0;
+	MQTTString Protocol;
+	int version;
+	int mylen = 0;
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	if (header.bits.type != CONNECT)
+		goto exit;
+
+	curdata += MQTTPacket_decodeBuf(curdata, &mylen); /* read remaining length */
+
+	if (!readMQTTLenString(&Protocol, &curdata, enddata) ||
+		enddata - curdata < 0) /* do we have enough data to read the protocol version byte? */
+		goto exit;
+
+	version = (int)readChar(&curdata); /* Protocol version */
+	/* If we don't recognize the protocol version, we don't parse the connect packet on the
+	 * basis that we don't know what the format will be.
+	 */
+	if (MQTTPacket_checkVersion(&Protocol, version))
+	{
+		flags.all = readChar(&curdata);
+		data->cleansession = flags.bits.cleansession;
+		data->keepAliveInterval = readInt(&curdata);
+		if (!readMQTTLenString(&data->clientID, &curdata, enddata))
+			goto exit;
+		data->willFlag = flags.bits.will;
+		if (flags.bits.will)
+		{
+			data->will.qos = flags.bits.willQoS;
+			data->will.retained = flags.bits.willRetain;
+			if (!readMQTTLenString(&data->will.topicName, &curdata, enddata) ||
+				  !readMQTTLenString(&data->will.message, &curdata, enddata))
+				goto exit;
+		}
+		if (flags.bits.username)
+		{
+			if (enddata - curdata < 3 || !readMQTTLenString(&data->username, &curdata, enddata))
+				goto exit; /* username flag set, but no username supplied - invalid */
+			if (flags.bits.password &&
+				(enddata - curdata < 3 || !readMQTTLenString(&data->password, &curdata, enddata)))
+				goto exit; /* password flag set, but no password supplied - invalid */
+		}
+		else if (flags.bits.password)
+			goto exit; /* password flag set without username - invalid */
+		rc = 1;
+	}
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Serializes the connack packet into the supplied buffer.
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param connack_rc the integer connack return code to be used 
+  * @param sessionPresent the MQTT 3.1.1 sessionPresent flag
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_connack(unsigned char* buf, int buflen, unsigned char connack_rc, unsigned char sessionPresent)
+{
+	MQTTHeader header = {0};
+	int rc = 0;
+	unsigned char *ptr = buf;
+	MQTTConnackFlags flags = {0};
+
+	FUNC_ENTRY;
+	if (buflen < 2)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+	header.byte = 0;
+	header.bits.type = CONNACK;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */
+
+	flags.all = 0;
+	flags.bits.sessionpresent = sessionPresent;
+	writeChar(&ptr, flags.all); 
+	writeChar(&ptr, connack_rc);
+
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTDeserializePublish.c

@@ -0,0 +1,107 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "StackTrace.h"
+#include "MQTTPacket.h"
+#include <string.h>
+
+#define min(a, b) ((a < b) ? 1 : 0)
+
+/**
+  * Deserializes the supplied (wire) buffer into publish data
+  * @param dup returned integer - the MQTT dup flag
+  * @param qos returned integer - the MQTT QoS value
+  * @param retained returned integer - the MQTT retained flag
+  * @param packetid returned integer - the MQTT packet identifier
+  * @param topicName returned MQTTString - the MQTT topic in the publish
+  * @param payload returned byte buffer - the MQTT publish payload
+  * @param payloadlen returned integer - the length of the MQTT payload
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @return error code.  1 is success
+  */
+int MQTTDeserialize_publish(unsigned char* dup, int* qos, unsigned char* retained, unsigned short* packetid, MQTTString* topicName,
+		unsigned char** payload, int32_t* payloadlen, unsigned char* buf, int buflen)
+{
+	MQTTHeader header = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = NULL;
+	int rc = 0;
+	int mylen = 0;
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	if (header.bits.type != PUBLISH)
+		goto exit;
+	*dup = header.bits.dup;
+	*qos = header.bits.qos;
+	*retained = header.bits.retain;
+
+	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */
+	enddata = curdata + mylen;
+
+	if (!readMQTTLenString(topicName, &curdata, enddata) ||
+		enddata - curdata < 0) /* do we have enough data to read the protocol version byte? */
+		goto exit;
+
+	if (*qos > 0)
+		*packetid = readInt(&curdata);
+
+	*payloadlen = enddata - curdata;
+	*payload = curdata;
+	rc = 1;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+
+/**
+  * Deserializes the supplied (wire) buffer into an ack
+  * @param packettype returned integer - the MQTT packet type
+  * @param dup returned integer - the MQTT dup flag
+  * @param packetid returned integer - the MQTT packet identifier
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @return error code.  1 is success, 0 is failure
+  */
+int MQTTDeserialize_ack(unsigned char* packettype, unsigned char* dup, unsigned short* packetid, unsigned char* buf, int buflen)
+{
+	MQTTHeader header = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = NULL;
+	int rc = 0;
+	int mylen;
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	*dup = header.bits.dup;
+	*packettype = header.bits.type;
+
+	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */
+	enddata = curdata + mylen;
+
+	if (enddata - curdata < 2)
+		goto exit;
+	*packetid = readInt(&curdata);
+
+	rc = 1;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTFormat.c

@@ -0,0 +1,262 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "StackTrace.h"
+#include "MQTTPacket.h"
+
+#include <string.h>
+
+
+const char* MQTTPacket_names[] =
+{
+	"RESERVED", "CONNECT", "CONNACK", "PUBLISH", "PUBACK", "PUBREC", "PUBREL",
+	"PUBCOMP", "SUBSCRIBE", "SUBACK", "UNSUBSCRIBE", "UNSUBACK",
+	"PINGREQ", "PINGRESP", "DISCONNECT"
+};
+
+
+const char* MQTTPacket_getName(unsigned short packetid)
+{
+	return MQTTPacket_names[packetid];
+}
+
+
+int MQTTStringFormat_connect(char* strbuf, int strbuflen, MQTTPacket_connectData* data)
+{
+	int strindex = 0;
+
+	strindex = snprintf(strbuf, strbuflen,
+			"CONNECT MQTT version %d, client id %.*s, clean session %d, keep alive %d",
+			(int)data->MQTTVersion, data->clientID.lenstring.len, data->clientID.lenstring.data,
+			(int)data->cleansession, data->keepAliveInterval);
+	if (data->willFlag)
+		strindex += snprintf(&strbuf[strindex], strbuflen - strindex,
+				", will QoS %d, will retain %d, will topic %.*s, will message %.*s",
+				data->will.qos, data->will.retained,
+				data->will.topicName.lenstring.len, data->will.topicName.lenstring.data,
+				data->will.message.lenstring.len, data->will.message.lenstring.data);
+	if (data->username.lenstring.data && data->username.lenstring.len > 0)
+		strindex += snprintf(&strbuf[strindex], strbuflen - strindex,
+				", user name %.*s", data->username.lenstring.len, data->username.lenstring.data);
+	if (data->password.lenstring.data && data->password.lenstring.len > 0)
+		strindex += snprintf(&strbuf[strindex], strbuflen - strindex,
+				", password %.*s", data->password.lenstring.len, data->password.lenstring.data);
+	return strindex;
+}
+
+
+int MQTTStringFormat_connack(char* strbuf, int strbuflen, unsigned char connack_rc, unsigned char sessionPresent)
+{
+	int strindex = snprintf(strbuf, strbuflen, "CONNACK session present %d, rc %d", sessionPresent, connack_rc);
+	return strindex;
+}
+
+
+int MQTTStringFormat_publish(char* strbuf, int strbuflen, unsigned char dup, int qos, unsigned char retained,
+		unsigned short packetid, MQTTString topicName, unsigned char* payload, int payloadlen)
+{
+	int strindex = snprintf(strbuf, strbuflen,
+				"PUBLISH dup %d, QoS %d, retained %d, packet id %d, topic %.*s, payload length %d, payload %.*s",
+				dup, qos, retained, packetid,
+				(topicName.lenstring.len < 20) ? topicName.lenstring.len : 20, topicName.lenstring.data,
+				payloadlen, (payloadlen < 20) ? payloadlen : 20, payload);
+	return strindex;
+}
+
+
+int MQTTStringFormat_ack(char* strbuf, int strbuflen, unsigned char packettype, unsigned char dup, unsigned short packetid)
+{
+	int strindex = snprintf(strbuf, strbuflen, "%s, packet id %d", MQTTPacket_names[packettype], packetid);
+	if (dup)
+		strindex += snprintf(strbuf + strindex, strbuflen - strindex, ", dup %d", dup);
+	return strindex;
+}
+
+
+int MQTTStringFormat_subscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid, int count,
+		MQTTString topicFilters[], int requestedQoSs[])
+{
+	return snprintf(strbuf, strbuflen,
+		"SUBSCRIBE dup %d, packet id %d count %d topic %.*s qos %d",
+		dup, packetid, count,
+		topicFilters[0].lenstring.len, topicFilters[0].lenstring.data,
+		requestedQoSs[0]);
+}
+
+
+int MQTTStringFormat_suback(char* strbuf, int strbuflen, unsigned short packetid, int count, int* grantedQoSs)
+{
+	return snprintf(strbuf, strbuflen,
+		"SUBACK packet id %d count %d granted qos %d", packetid, count, grantedQoSs[0]);
+}
+
+
+int MQTTStringFormat_unsubscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid,
+		int count, MQTTString topicFilters[])
+{
+	return snprintf(strbuf, strbuflen,
+					"UNSUBSCRIBE dup %d, packet id %d count %d topic %.*s",
+					dup, packetid, count,
+					topicFilters[0].lenstring.len, topicFilters[0].lenstring.data);
+}
+
+
+#if defined(MQTT_CLIENT)
+char* MQTTFormat_toClientString(char* strbuf, int strbuflen, unsigned char* buf, int buflen)
+{
+	int index = 0;
+	int rem_length = 0;
+	MQTTHeader header = {0};
+	int strindex = 0;
+
+	header.byte = buf[index++];
+	index += MQTTPacket_decodeBuf(&buf[index], &rem_length);
+
+	switch (header.bits.type)
+	{
+
+	case CONNACK:
+	{
+		unsigned char sessionPresent, connack_rc;
+		if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) == 1)
+			strindex = MQTTStringFormat_connack(strbuf, strbuflen, connack_rc, sessionPresent);
+	}
+	break;
+	case PUBLISH:
+	{
+		unsigned char dup, retained, *payload;
+		unsigned short packetid;
+		int qos, payloadlen;
+		MQTTString topicName = MQTTString_initializer;
+		if (MQTTDeserialize_publish(&dup, &qos, &retained, &packetid, &topicName,
+				&payload, &payloadlen, buf, buflen) == 1)
+			strindex = MQTTStringFormat_publish(strbuf, strbuflen, dup, qos, retained, packetid,
+					topicName, payload, payloadlen);
+	}
+	break;
+	case PUBACK:
+	case PUBREC:
+	case PUBREL:
+	case PUBCOMP:
+	{
+		unsigned char packettype, dup;
+		unsigned short packetid;
+		if (MQTTDeserialize_ack(&packettype, &dup, &packetid, buf, buflen) == 1)
+			strindex = MQTTStringFormat_ack(strbuf, strbuflen, packettype, dup, packetid);
+	}
+	break;
+	case SUBACK:
+	{
+		unsigned short packetid;
+		int maxcount = 1, count = 0;
+		int grantedQoSs[1];
+		if (MQTTDeserialize_suback(&packetid, maxcount, &count, grantedQoSs, buf, buflen) == 1)
+			strindex = MQTTStringFormat_suback(strbuf, strbuflen, packetid, count, grantedQoSs);
+	}
+	break;
+	case UNSUBACK:
+	{
+		unsigned short packetid;
+		if (MQTTDeserialize_unsuback(&packetid, buf, buflen) == 1)
+			strindex = MQTTStringFormat_ack(strbuf, strbuflen, UNSUBACK, 0, packetid);
+	}
+	break;
+	case PINGREQ:
+	case PINGRESP:
+	case DISCONNECT:
+		strindex = snprintf(strbuf, strbuflen, "%s", MQTTPacket_names[header.bits.type]);
+		break;
+	}
+	return strbuf;
+}
+#endif
+
+#if defined(MQTT_SERVER)
+char* MQTTFormat_toServerString(char* strbuf, int strbuflen, unsigned char* buf, int buflen)
+{
+	int index = 0;
+	int rem_length = 0;
+	MQTTHeader header = {0};
+	int strindex = 0;
+
+	header.byte = buf[index++];
+	index += MQTTPacket_decodeBuf(&buf[index], &rem_length);
+
+	switch (header.bits.type)
+	{
+	case CONNECT:
+	{
+		MQTTPacket_connectData data;
+		int rc;
+		if ((rc = MQTTDeserialize_connect(&data, buf, buflen)) == 1)
+			strindex = MQTTStringFormat_connect(strbuf, strbuflen, &data);
+	}
+	break;
+	case PUBLISH:
+	{
+		unsigned char dup, retained, *payload;
+		unsigned short packetid;
+		int qos, payloadlen;
+		MQTTString topicName = MQTTString_initializer;
+		if (MQTTDeserialize_publish(&dup, &qos, &retained, &packetid, &topicName,
+				&payload, &payloadlen, buf, buflen) == 1)
+			strindex = MQTTStringFormat_publish(strbuf, strbuflen, dup, qos, retained, packetid,
+					topicName, payload, payloadlen);
+	}
+	break;
+	case PUBACK:
+	case PUBREC:
+	case PUBREL:
+	case PUBCOMP:
+	{
+		unsigned char packettype, dup;
+		unsigned short packetid;
+		if (MQTTDeserialize_ack(&packettype, &dup, &packetid, buf, buflen) == 1)
+			strindex = MQTTStringFormat_ack(strbuf, strbuflen, packettype, dup, packetid);
+	}
+	break;
+	case SUBSCRIBE:
+	{
+		unsigned char dup;
+		unsigned short packetid;
+		int maxcount = 1, count = 0;
+		MQTTString topicFilters[1];
+		int requestedQoSs[1];
+		if (MQTTDeserialize_subscribe(&dup, &packetid, maxcount, &count,
+				topicFilters, requestedQoSs, buf, buflen) == 1)
+			strindex = MQTTStringFormat_subscribe(strbuf, strbuflen, dup, packetid, count, topicFilters, requestedQoSs);;
+	}
+	break;
+	case UNSUBSCRIBE:
+	{
+		unsigned char dup;
+		unsigned short packetid;
+		int maxcount = 1, count = 0;
+		MQTTString topicFilters[1];
+		if (MQTTDeserialize_unsubscribe(&dup, &packetid, maxcount, &count, topicFilters, buf, buflen) == 1)
+			strindex =  MQTTStringFormat_unsubscribe(strbuf, strbuflen, dup, packetid, count, topicFilters);
+	}
+	break;
+	case PINGREQ:
+	case PINGRESP:
+	case DISCONNECT:
+		strindex = snprintf(strbuf, strbuflen, "%s", MQTTPacket_names[header.bits.type]);
+		break;
+	}
+	strbuf[strbuflen] = '\0';
+	return strbuf;
+}
+#endif

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTFormat.h

@@ -0,0 +1,37 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#if !defined(MQTTFORMAT_H)
+#define MQTTFORMAT_H
+
+#include "StackTrace.h"
+#include "MQTTPacket.h"
+
+const char* MQTTPacket_getName(unsigned short packetid);
+int MQTTStringFormat_connect(char* strbuf, int strbuflen, MQTTPacket_connectData* data);
+int MQTTStringFormat_connack(char* strbuf, int strbuflen, unsigned char connack_rc, unsigned char sessionPresent);
+int MQTTStringFormat_publish(char* strbuf, int strbuflen, unsigned char dup, int qos, unsigned char retained,
+		unsigned short packetid, MQTTString topicName, unsigned char* payload, int payloadlen);
+int MQTTStringFormat_ack(char* strbuf, int strbuflen, unsigned char packettype, unsigned char dup, unsigned short packetid);
+int MQTTStringFormat_subscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid, int count,
+		MQTTString topicFilters[], int requestedQoSs[]);
+int MQTTStringFormat_suback(char* strbuf, int strbuflen, unsigned short packetid, int count, int* grantedQoSs);
+int MQTTStringFormat_unsubscribe(char* strbuf, int strbuflen, unsigned char dup, unsigned short packetid,
+		int count, MQTTString topicFilters[]);
+char* MQTTFormat_toClientString(char* strbuf, int strbuflen, unsigned char* buf, int buflen);
+char* MQTTFormat_toServerString(char* strbuf, int strbuflen, unsigned char* buf, int buflen);
+
+#endif

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTPacket.c

@@ -0,0 +1,412 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Sergio R. Caprile - non-blocking packet read functions for stream transport
+ *******************************************************************************/
+
+#include "StackTrace.h"
+#include "MQTTPacket.h"
+
+#include <string.h>
+
+/**
+ * Encodes the message length according to the MQTT algorithm
+ * @param buf the buffer into which the encoded data is written
+ * @param length the length to be encoded
+ * @return the number of bytes written to buffer
+ */
+int MQTTPacket_encode(unsigned char* buf, int length)
+{
+	int rc = 0;
+
+	FUNC_ENTRY;
+	do
+	{
+		char d = length % 128;
+		length /= 128;
+		/* if there are more digits to encode, set the top bit of this digit */
+		if (length > 0)
+			d |= 0x80;
+		buf[rc++] = d;
+	} while (length > 0);
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+ * Decodes the message length according to the MQTT algorithm
+ * @param getcharfn pointer to function to read the next character from the data source
+ * @param value the decoded length returned
+ * @return the number of bytes read from the socket
+ */
+int MQTTPacket_decode(int (*getcharfn)(unsigned char*, int), int* value)
+{
+	unsigned char c;
+	int multiplier = 1;
+	int len = 0;
+#define MAX_NO_OF_REMAINING_LENGTH_BYTES 4
+
+	FUNC_ENTRY;
+	*value = 0;
+	do
+	{
+		int rc = MQTTPACKET_READ_ERROR;
+
+		if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
+		{
+			rc = MQTTPACKET_READ_ERROR;	/* bad data */
+			goto exit;
+		}
+		rc = (*getcharfn)(&c, 1);
+		if (rc != 1)
+			goto exit;
+		*value += (c & 127) * multiplier;
+		multiplier *= 128;
+	} while ((c & 128) != 0);
+exit:
+	FUNC_EXIT_RC(len);
+	return len;
+}
+
+
+int MQTTPacket_len(int rem_len)
+{
+	rem_len += 1; /* header byte */
+
+	/* now remaining_length field */
+	if (rem_len < 128)
+		rem_len += 1;
+	else if (rem_len < 16384)
+		rem_len += 2;
+	else if (rem_len < 2097151)
+		rem_len += 3;
+	else
+		rem_len += 4;
+	return rem_len;
+}
+
+
+static unsigned char* bufptr;
+
+int bufchar(unsigned char* c, int count)
+{
+	int i;
+
+	for (i = 0; i < count; ++i)
+		*c = *bufptr++;
+	return count;
+}
+
+
+int MQTTPacket_decodeBuf(unsigned char* buf, int* value)
+{
+	bufptr = buf;
+	return MQTTPacket_decode(bufchar, value);
+}
+
+
+/**
+ * Calculates an integer from two bytes read from the input buffer
+ * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned
+ * @return the integer value calculated
+ */
+int readInt(unsigned char** pptr)
+{
+	unsigned char* ptr = *pptr;
+	int len = 256*(*ptr) + (*(ptr+1));
+	*pptr += 2;
+	return len;
+}
+
+
+/**
+ * Reads one character from the input buffer.
+ * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned
+ * @return the character read
+ */
+char readChar(unsigned char** pptr)
+{
+	char c = **pptr;
+	(*pptr)++;
+	return c;
+}
+
+
+/**
+ * Writes one character to an output buffer.
+ * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
+ * @param c the character to write
+ */
+void writeChar(unsigned char** pptr, char c)
+{
+	**pptr = c;
+	(*pptr)++;
+}
+
+
+/**
+ * Writes an integer as 2 bytes to an output buffer.
+ * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
+ * @param anInt the integer to write
+ */
+void writeInt(unsigned char** pptr, int anInt)
+{
+	**pptr = (unsigned char)(anInt / 256);
+	(*pptr)++;
+	**pptr = (unsigned char)(anInt % 256);
+	(*pptr)++;
+}
+
+
+/**
+ * Writes a "UTF" string to an output buffer.  Converts C string to length-delimited.
+ * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
+ * @param string the C string to write
+ */
+void writeCString(unsigned char** pptr, const char* string)
+{
+	int len = strlen(string);
+	writeInt(pptr, len);
+	memcpy(*pptr, string, len);
+	*pptr += len;
+}
+
+
+int getLenStringLen(char* ptr)
+{
+	int len = 256*((unsigned char)(*ptr)) + (unsigned char)(*(ptr+1));
+	return len;
+}
+
+
+void writeMQTTString(unsigned char** pptr, MQTTString mqttstring)
+{
+	if (mqttstring.lenstring.len > 0)
+	{
+		writeInt(pptr, mqttstring.lenstring.len);
+		memcpy(*pptr, mqttstring.lenstring.data, mqttstring.lenstring.len);
+		*pptr += mqttstring.lenstring.len;
+	}
+	else if (mqttstring.cstring)
+		writeCString(pptr, mqttstring.cstring);
+	else
+		writeInt(pptr, 0);
+}
+
+
+/**
+ * @param mqttstring the MQTTString structure into which the data is to be read
+ * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
+ * @param enddata pointer to the end of the data: do not read beyond
+ * @return 1 if successful, 0 if not
+ */
+int readMQTTLenString(MQTTString* mqttstring, unsigned char** pptr, unsigned char* enddata)
+{
+	int rc = 0;
+
+	FUNC_ENTRY;
+	/* the first two bytes are the length of the string */
+	if (enddata - (*pptr) > 1) /* enough length to read the integer? */
+	{
+		mqttstring->lenstring.len = readInt(pptr); /* increments pptr to point past length */
+		if (&(*pptr)[mqttstring->lenstring.len] <= enddata)
+		{
+			mqttstring->lenstring.data = (char*)*pptr;
+			*pptr += mqttstring->lenstring.len;
+			rc = 1;
+		}
+	}
+	mqttstring->cstring = NULL;
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+ * Return the length of the MQTTstring - C string if there is one, otherwise the length delimited string
+ * @param mqttstring the string to return the length of
+ * @return the length of the string
+ */
+int MQTTstrlen(MQTTString mqttstring)
+{
+	int rc = 0;
+
+	if (mqttstring.cstring)
+		rc = strlen(mqttstring.cstring);
+	else
+		rc = mqttstring.lenstring.len;
+	return rc;
+}
+
+
+/**
+ * Compares an MQTTString to a C string
+ * @param a the MQTTString to compare
+ * @param bptr the C string to compare
+ * @return boolean - equal or not
+ */
+int MQTTPacket_equals(MQTTString* a, char* bptr)
+{
+	int alen = 0,
+		blen = 0;
+	char *aptr;
+	
+	if (a->cstring)
+	{
+		aptr = a->cstring;
+		alen = strlen(a->cstring);
+	}
+	else
+	{
+		aptr = a->lenstring.data;
+		alen = a->lenstring.len;
+	}
+	blen = strlen(bptr);
+	
+	return (alen == blen) && (strncmp(aptr, bptr, alen) == 0);
+}
+
+
+/**
+ * Helper function to read packet data from some source into a buffer
+ * @param buf the buffer into which the packet will be serialized
+ * @param buflen the length in bytes of the supplied buffer
+ * @param getfn pointer to a function which will read any number of bytes from the needed source
+ * @return integer MQTT packet type, or -1 on error
+ * @note  the whole message must fit into the caller's buffer
+ */
+int MQTTPacket_read(unsigned char* buf, int buflen, int (*getfn)(unsigned char*, int))
+{
+	int rc = -1;
+	MQTTHeader header = {0};
+	int len = 0;
+	int rem_len = 0;
+
+	/* 1. read the header byte.  This has the packet type in it */
+	if ((*getfn)(buf, 1) != 1)
+		goto exit;
+
+	len = 1;
+	/* 2. read the remaining length.  This is variable in itself */
+	MQTTPacket_decode(getfn, &rem_len);
+	len += MQTTPacket_encode(buf + 1, rem_len); /* put the original remaining length back into the buffer */
+
+	/* 3. read the rest of the buffer using a callback to supply the rest of the data */
+	if((rem_len + len) > buflen)
+		goto exit;
+	if (rem_len && ((*getfn)(buf + len, rem_len) != rem_len))
+		goto exit;
+
+	header.byte = buf[0];
+	rc = header.bits.type;
+exit:
+	return rc;
+}
+
+/**
+ * Decodes the message length according to the MQTT algorithm, non-blocking
+ * @param trp pointer to a transport structure holding what is needed to solve getting data from it
+ * @param value the decoded length returned
+ * @return integer the number of bytes read from the socket, 0 for call again, or -1 on error
+ */
+static int MQTTPacket_decodenb(MQTTTransport *trp)
+{
+	unsigned char c;
+	int rc = MQTTPACKET_READ_ERROR;
+
+	FUNC_ENTRY;
+	if(trp->len == 0){		/* initialize on first call */
+		trp->multiplier = 1;
+		trp->rem_len = 0;
+	}
+	do {
+		int frc;
+		if (trp->len >= MAX_NO_OF_REMAINING_LENGTH_BYTES)
+			goto exit;
+		if ((frc=(*trp->getfn)(trp->sck, &c, 1)) == -1)
+			goto exit;
+		if (frc == 0){
+			rc = 0;
+			goto exit;
+		}
+		++(trp->len);
+		trp->rem_len += (c & 127) * trp->multiplier;
+		trp->multiplier *= 128;
+	} while ((c & 128) != 0);
+	rc = trp->len;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+/**
+ * Helper function to read packet data from some source into a buffer, non-blocking
+ * @param buf the buffer into which the packet will be serialized
+ * @param buflen the length in bytes of the supplied buffer
+ * @param trp pointer to a transport structure holding what is needed to solve getting data from it
+ * @return integer MQTT packet type, 0 for call again, or -1 on error
+ * @note  the whole message must fit into the caller's buffer
+ */
+int MQTTPacket_readnb(unsigned char* buf, int buflen, MQTTTransport *trp)
+{
+	int rc = -1, frc;
+	MQTTHeader header = {0};
+
+	switch(trp->state){
+	default:
+		trp->state = 0;
+		/*FALLTHROUGH*/
+	case 0:
+		/* read the header byte.  This has the packet type in it */
+		if ((frc=(*trp->getfn)(trp->sck, buf, 1)) == -1)
+			goto exit;
+		if (frc == 0)
+			return 0;
+		trp->len = 0;
+		++trp->state;
+		/*FALLTHROUGH*/
+		/* read the remaining length.  This is variable in itself */
+	case 1:
+		if((frc=MQTTPacket_decodenb(trp)) == MQTTPACKET_READ_ERROR)
+			goto exit;
+		if(frc == 0)
+			return 0;
+		trp->len = 1 + MQTTPacket_encode(buf + 1, trp->rem_len); /* put the original remaining length back into the buffer */
+		if((trp->rem_len + trp->len) > buflen)
+			goto exit;
+		++trp->state;
+		/*FALLTHROUGH*/
+	case 2:
+		if(trp->rem_len){
+			/* read the rest of the buffer using a callback to supply the rest of the data */
+			if ((frc=(*trp->getfn)(trp->sck, buf + trp->len, trp->rem_len)) == -1)
+				goto exit;
+			if (frc == 0)
+				return 0;
+			trp->rem_len -= frc;
+			trp->len += frc;
+			if(trp->rem_len)
+				return 0;
+		}
+		header.byte = buf[0];
+		rc = header.bits.type;
+		break;
+	}
+
+exit:
+	trp->state = 0;
+	return rc;
+}
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTPacket.h

@@ -0,0 +1,134 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Xiang Rong - 442039 Add makefile to Embedded C client
+ *******************************************************************************/
+#include <stdint.h>
+
+#ifndef MQTTPACKET_H_
+#define MQTTPACKET_H_
+
+#if defined(__cplusplus) /* If this is a C++ compiler, use C linkage */
+extern "C" {
+#endif
+
+#if defined(WIN32_DLL) || defined(WIN64_DLL)
+  #define DLLImport __declspec(dllimport)
+  #define DLLExport __declspec(dllexport)
+#elif defined(LINUX_SO)
+  #define DLLImport extern
+  #define DLLExport  __attribute__ ((visibility ("default")))
+#else
+  #define DLLImport
+  #define DLLExport  
+#endif
+
+enum errors
+{
+	MQTTPACKET_BUFFER_TOO_SHORT = -2,
+	MQTTPACKET_READ_ERROR = -1,
+	MQTTPACKET_READ_COMPLETE
+};
+
+enum msgTypes
+{
+	CONNECT = 1, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL,
+	PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK,
+	PINGREQ, PINGRESP, DISCONNECT
+};
+
+/**
+ * Bitfields for the MQTT header byte.
+ */
+typedef union
+{
+	unsigned char byte;	                /**< the whole byte */
+#if defined(REVERSED)
+	struct
+	{
+		unsigned int type : 4;			/**< message type nibble */
+		unsigned int dup : 1;				/**< DUP flag bit */
+		unsigned int qos : 2;				/**< QoS value, 0, 1 or 2 */
+		unsigned int retain : 1;		/**< retained flag bit */
+	} bits;
+#else
+	struct
+	{
+		unsigned int retain : 1;		/**< retained flag bit */
+		unsigned int qos : 2;				/**< QoS value, 0, 1 or 2 */
+		unsigned int dup : 1;				/**< DUP flag bit */
+		unsigned int type : 4;			/**< message type nibble */
+	} bits;
+#endif
+} MQTTHeader;
+
+typedef struct
+{
+	int len;
+	char* data;
+} MQTTLenString;
+
+typedef struct
+{
+	char* cstring;
+	MQTTLenString lenstring;
+} MQTTString;
+
+#define MQTTString_initializer {NULL, {0, NULL}}
+
+int MQTTstrlen(MQTTString mqttstring);
+
+#include "MQTTConnect.h"
+#include "MQTTPublish.h"
+#include "MQTTSubscribe.h"
+#include "MQTTUnsubscribe.h"
+#include "MQTTFormat.h"
+
+DLLExport int MQTTSerialize_ack(unsigned char* buf, int buflen, unsigned char type, unsigned char dup, unsigned short packetid);
+DLLExport int MQTTDeserialize_ack(unsigned char* packettype, unsigned char* dup, unsigned short* packetid, unsigned char* buf, int buflen);
+
+int MQTTPacket_len(int rem_len);
+DLLExport int MQTTPacket_equals(MQTTString* a, char* b);
+
+DLLExport int MQTTPacket_encode(unsigned char* buf, int length);
+int MQTTPacket_decode(int (*getcharfn)(unsigned char*, int), int* value);
+int MQTTPacket_decodeBuf(unsigned char* buf, int* value);
+
+int readInt(unsigned char** pptr);
+char readChar(unsigned char** pptr);
+void writeChar(unsigned char** pptr, char c);
+void writeInt(unsigned char** pptr, int anInt);
+int readMQTTLenString(MQTTString* mqttstring, unsigned char** pptr, unsigned char* enddata);
+void writeCString(unsigned char** pptr, const char* string);
+void writeMQTTString(unsigned char** pptr, MQTTString mqttstring);
+
+DLLExport int MQTTPacket_read(unsigned char* buf, int buflen, int (*getfn)(unsigned char*, int));
+
+typedef struct {
+	int (*getfn)(void *, unsigned char*, int); /* must return -1 for error, 0 for call again, or the number of bytes read */
+	void *sck;	/* pointer to whatever the system may use to identify the transport */
+	int multiplier;
+	int rem_len;
+	int len;
+	char state;
+}MQTTTransport;
+
+int MQTTPacket_readnb(unsigned char* buf, int buflen, MQTTTransport *trp);
+
+#ifdef __cplusplus /* If this is a C++ compiler, use C linkage */
+}
+#endif
+
+
+#endif /* MQTTPACKET_H_ */

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTPublish.h

@@ -0,0 +1,38 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Xiang Rong - 442039 Add makefile to Embedded C client
+ *******************************************************************************/
+
+#ifndef MQTTPUBLISH_H_
+#define MQTTPUBLISH_H_
+
+#if !defined(DLLImport)
+  #define DLLImport 
+#endif
+#if !defined(DLLExport)
+  #define DLLExport
+#endif
+
+DLLExport int MQTTSerialize_publish(unsigned char* buf, int buflen, unsigned char dup, int qos, unsigned char retained, unsigned short packetid,
+		MQTTString topicName, unsigned char* payload, int payloadlen);
+
+DLLExport int MQTTDeserialize_publish(unsigned char* dup, int* qos, unsigned char* retained, unsigned short* packetid, MQTTString* topicName,
+		unsigned char** payload, int32_t* payloadlen, unsigned char* buf, int len);
+
+DLLExport int MQTTSerialize_puback(unsigned char* buf, int buflen, unsigned short packetid);
+DLLExport int MQTTSerialize_pubrel(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid);
+DLLExport int MQTTSerialize_pubcomp(unsigned char* buf, int buflen, unsigned short packetid);
+
+#endif /* MQTTPUBLISH_H_ */

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTSerializePublish.c

@@ -0,0 +1,169 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Ian Craggs - fix for https://bugs.eclipse.org/bugs/show_bug.cgi?id=453144
+ *******************************************************************************/
+
+#include "MQTTPacket.h"
+#include "StackTrace.h"
+
+#include <string.h>
+
+
+/**
+  * Determines the length of the MQTT publish packet that would be produced using the supplied parameters
+  * @param qos the MQTT QoS of the publish (packetid is omitted for QoS 0)
+  * @param topicName the topic name to be used in the publish  
+  * @param payloadlen the length of the payload to be sent
+  * @return the length of buffer needed to contain the serialized version of the packet
+  */
+int MQTTSerialize_publishLength(int qos, MQTTString topicName, int payloadlen)
+{
+	int len = 0;
+
+	len += 2 + MQTTstrlen(topicName) + payloadlen;
+	if (qos > 0)
+		len += 2; /* packetid */
+	return len;
+}
+
+
+/**
+  * Serializes the supplied publish data into the supplied buffer, ready for sending
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param dup integer - the MQTT dup flag
+  * @param qos integer - the MQTT QoS value
+  * @param retained integer - the MQTT retained flag
+  * @param packetid integer - the MQTT packet identifier
+  * @param topicName MQTTString - the MQTT topic in the publish
+  * @param payload byte buffer - the MQTT publish payload
+  * @param payloadlen integer - the length of the MQTT payload
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTSerialize_publish(unsigned char* buf, int buflen, unsigned char dup, int qos, unsigned char retained, unsigned short packetid,
+		MQTTString topicName, unsigned char* payload, int payloadlen)
+{
+	unsigned char *ptr = buf;
+	MQTTHeader header = {0};
+	int rem_len = 0;
+	int rc = 0;
+
+	FUNC_ENTRY;
+	if (MQTTPacket_len(rem_len = MQTTSerialize_publishLength(qos, topicName, payloadlen)) > buflen)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+
+	header.bits.type = PUBLISH;
+	header.bits.dup = dup;
+	header.bits.qos = qos;
+	header.bits.retain = retained;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */;
+
+	writeMQTTString(&ptr, topicName);
+
+	if (qos > 0)
+		writeInt(&ptr, packetid);
+
+	memcpy(ptr, payload, payloadlen);
+	ptr += payloadlen;
+
+	rc = ptr - buf;
+
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+
+/**
+  * Serializes the ack packet into the supplied buffer.
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param type the MQTT packet type
+  * @param dup the MQTT dup flag
+  * @param packetid the MQTT packet identifier
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_ack(unsigned char* buf, int buflen, unsigned char packettype, unsigned char dup, unsigned short packetid)
+{
+	MQTTHeader header = {0};
+	int rc = 0;
+	unsigned char *ptr = buf;
+
+	FUNC_ENTRY;
+	if (buflen < 4)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+	header.bits.type = packettype;
+	header.bits.dup = dup;
+	header.bits.qos = (packettype == PUBREL) ? 1 : 0;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */
+	writeInt(&ptr, packetid);
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Serializes a puback packet into the supplied buffer.
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param packetid integer - the MQTT packet identifier
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_puback(unsigned char* buf, int buflen, unsigned short packetid)
+{
+	return MQTTSerialize_ack(buf, buflen, PUBACK, 0, packetid);
+}
+
+
+/**
+  * Serializes a pubrel packet into the supplied buffer.
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param dup integer - the MQTT dup flag
+  * @param packetid integer - the MQTT packet identifier
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_pubrel(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid)
+{
+	return MQTTSerialize_ack(buf, buflen, PUBREL, dup, packetid);
+}
+
+
+/**
+  * Serializes a pubrel packet into the supplied buffer.
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param packetid integer - the MQTT packet identifier
+  * @return serialized length, or error if 0
+  */
+int MQTTSerialize_pubcomp(unsigned char* buf, int buflen, unsigned short packetid)
+{
+	return MQTTSerialize_ack(buf, buflen, PUBCOMP, 0, packetid);
+}
+
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTSubscribe.h

@@ -0,0 +1,39 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Xiang Rong - 442039 Add makefile to Embedded C client
+ *******************************************************************************/
+
+#ifndef MQTTSUBSCRIBE_H_
+#define MQTTSUBSCRIBE_H_
+
+#if !defined(DLLImport)
+  #define DLLImport 
+#endif
+#if !defined(DLLExport)
+  #define DLLExport
+#endif
+
+DLLExport int MQTTSerialize_subscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid,
+		int count, MQTTString topicFilters[], int32_t requestedQoSs[]);
+
+DLLExport int MQTTDeserialize_subscribe(unsigned char* dup, unsigned short* packetid,
+		int maxcount, int* count, MQTTString topicFilters[], int requestedQoSs[], unsigned char* buf, int len);
+
+DLLExport int MQTTSerialize_suback(unsigned char* buf, int buflen, unsigned short packetid, int count, int* grantedQoSs);
+
+DLLExport int MQTTDeserialize_suback(unsigned short* packetid, int maxcount, int32_t* count, int32_t grantedQoSs[], unsigned char* buf, int len);
+
+
+#endif /* MQTTSUBSCRIBE_H_ */

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTSubscribeClient.c

@@ -0,0 +1,137 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "MQTTPacket.h"
+#include "StackTrace.h"
+
+#include <string.h>
+
+/**
+  * Determines the length of the MQTT subscribe packet that would be produced using the supplied parameters
+  * @param count the number of topic filter strings in topicFilters
+  * @param topicFilters the array of topic filter strings to be used in the publish
+  * @return the length of buffer needed to contain the serialized version of the packet
+  */
+int MQTTSerialize_subscribeLength(int count, MQTTString topicFilters[])
+{
+	int i;
+	int len = 2; /* packetid */
+
+	for (i = 0; i < count; ++i)
+		len += 2 + MQTTstrlen(topicFilters[i]) + 1; /* length + topic + req_qos */
+	return len;
+}
+
+
+/**
+  * Serializes the supplied subscribe data into the supplied buffer, ready for sending
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied bufferr
+  * @param dup integer - the MQTT dup flag
+  * @param packetid integer - the MQTT packet identifier
+  * @param count - number of members in the topicFilters and reqQos arrays
+  * @param topicFilters - array of topic filter names
+  * @param requestedQoSs - array of requested QoS
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTSerialize_subscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid, int count,
+		MQTTString topicFilters[], int32_t requestedQoSs[])
+{
+	unsigned char *ptr = buf;
+	MQTTHeader header = {0};
+	int rem_len = 0;
+	int rc = 0;
+	int i = 0;
+
+	FUNC_ENTRY;
+	if (MQTTPacket_len(rem_len = MQTTSerialize_subscribeLength(count, topicFilters)) > buflen)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+
+	header.byte = 0;
+	header.bits.type = SUBSCRIBE;
+	header.bits.dup = dup;
+	header.bits.qos = 1;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */;
+
+	writeInt(&ptr, packetid);
+
+	for (i = 0; i < count; ++i)
+	{
+		writeMQTTString(&ptr, topicFilters[i]);
+		writeChar(&ptr, requestedQoSs[i]);
+	}
+
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+
+/**
+  * Deserializes the supplied (wire) buffer into suback data
+  * @param packetid returned integer - the MQTT packet identifier
+  * @param maxcount - the maximum number of members allowed in the grantedQoSs array
+  * @param count returned integer - number of members in the grantedQoSs array
+  * @param grantedQoSs returned array of integers - the granted qualities of service
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @return error code.  1 is success, 0 is failure
+  */
+int MQTTDeserialize_suback(unsigned short* packetid, int maxcount, int32_t* count, int32_t grantedQoSs[], unsigned char* buf, int buflen)
+{
+	MQTTHeader header = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = NULL;
+	int rc = 0;
+	int mylen;
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	if (header.bits.type != SUBACK)
+		goto exit;
+
+	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */
+	enddata = curdata + mylen;
+	if (enddata - curdata < 2)
+		goto exit;
+
+	*packetid = readInt(&curdata);
+
+	*count = 0;
+	while (curdata < enddata)
+	{
+		if (*count > maxcount)
+		{
+			rc = -1;
+			goto exit;
+		}
+		grantedQoSs[(*count)++] = readChar(&curdata);
+	}
+
+	rc = 1;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTSubscribeServer.c

@@ -0,0 +1,112 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "MQTTPacket.h"
+#include "StackTrace.h"
+
+#include <string.h>
+
+
+/**
+  * Deserializes the supplied (wire) buffer into subscribe data
+  * @param dup integer returned - the MQTT dup flag
+  * @param packetid integer returned - the MQTT packet identifier
+  * @param maxcount - the maximum number of members allowed in the topicFilters and requestedQoSs arrays
+  * @param count - number of members in the topicFilters and requestedQoSs arrays
+  * @param topicFilters - array of topic filter names
+  * @param requestedQoSs - array of requested QoS
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTDeserialize_subscribe(unsigned char* dup, unsigned short* packetid, int maxcount, int* count, MQTTString topicFilters[],
+	int requestedQoSs[], unsigned char* buf, int buflen)
+{
+	MQTTHeader header = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = NULL;
+	int rc = -1;
+	int mylen = 0;
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	if (header.bits.type != SUBSCRIBE)
+		goto exit;
+	*dup = header.bits.dup;
+
+	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */
+	enddata = curdata + mylen;
+
+	*packetid = readInt(&curdata);
+
+	*count = 0;
+	while (curdata < enddata)
+	{
+		if (!readMQTTLenString(&topicFilters[*count], &curdata, enddata))
+			goto exit;
+		if (curdata >= enddata) /* do we have enough data to read the req_qos version byte? */
+			goto exit;
+		requestedQoSs[*count] = readChar(&curdata);
+		(*count)++;
+	}
+
+	rc = 1;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Serializes the supplied suback data into the supplied buffer, ready for sending
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param packetid integer - the MQTT packet identifier
+  * @param count - number of members in the grantedQoSs array
+  * @param grantedQoSs - array of granted QoS
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTSerialize_suback(unsigned char* buf, int buflen, unsigned short packetid, int count, int* grantedQoSs)
+{
+	MQTTHeader header = {0};
+	int rc = -1;
+	unsigned char *ptr = buf;
+	int i;
+
+	FUNC_ENTRY;
+	if (buflen < 2 + count)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+	header.byte = 0;
+	header.bits.type = SUBACK;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, 2 + count); /* write remaining length */
+
+	writeInt(&ptr, packetid);
+
+	for (i = 0; i < count; ++i)
+		writeChar(&ptr, grantedQoSs[i]);
+
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTUnsubscribe.h

@@ -0,0 +1,38 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Xiang Rong - 442039 Add makefile to Embedded C client
+ *******************************************************************************/
+
+#ifndef MQTTUNSUBSCRIBE_H_
+#define MQTTUNSUBSCRIBE_H_
+
+#if !defined(DLLImport)
+  #define DLLImport 
+#endif
+#if !defined(DLLExport)
+  #define DLLExport
+#endif
+
+DLLExport int MQTTSerialize_unsubscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid,
+		int count, MQTTString topicFilters[]);
+
+DLLExport int MQTTDeserialize_unsubscribe(unsigned char* dup, unsigned short* packetid, int max_count, int* count, MQTTString topicFilters[],
+		unsigned char* buf, int len);
+
+DLLExport int MQTTSerialize_unsuback(unsigned char* buf, int buflen, unsigned short packetid);
+
+DLLExport int MQTTDeserialize_unsuback(unsigned short* packetid, unsigned char* buf, int len);
+
+#endif /* MQTTUNSUBSCRIBE_H_ */

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTUnsubscribeClient.c

@@ -0,0 +1,106 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "MQTTPacket.h"
+#include "StackTrace.h"
+
+#include <string.h>
+
+/**
+  * Determines the length of the MQTT unsubscribe packet that would be produced using the supplied parameters
+  * @param count the number of topic filter strings in topicFilters
+  * @param topicFilters the array of topic filter strings to be used in the publish
+  * @return the length of buffer needed to contain the serialized version of the packet
+  */
+int MQTTSerialize_unsubscribeLength(int count, MQTTString topicFilters[])
+{
+	int i;
+	int len = 2; /* packetid */
+
+	for (i = 0; i < count; ++i)
+		len += 2 + MQTTstrlen(topicFilters[i]); /* length + topic*/
+	return len;
+}
+
+
+/**
+  * Serializes the supplied unsubscribe data into the supplied buffer, ready for sending
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @param dup integer - the MQTT dup flag
+  * @param packetid integer - the MQTT packet identifier
+  * @param count - number of members in the topicFilters array
+  * @param topicFilters - array of topic filter names
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTSerialize_unsubscribe(unsigned char* buf, int buflen, unsigned char dup, unsigned short packetid,
+		int count, MQTTString topicFilters[])
+{
+	unsigned char *ptr = buf;
+	MQTTHeader header = {0};
+	int rem_len = 0;
+	int rc = -1;
+	int i = 0;
+
+	FUNC_ENTRY;
+	if (MQTTPacket_len(rem_len = MQTTSerialize_unsubscribeLength(count, topicFilters)) > buflen)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+
+	header.byte = 0;
+	header.bits.type = UNSUBSCRIBE;
+	header.bits.dup = dup;
+	header.bits.qos = 1;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, rem_len); /* write remaining length */;
+
+	writeInt(&ptr, packetid);
+
+	for (i = 0; i < count; ++i)
+		writeMQTTString(&ptr, topicFilters[i]);
+
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Deserializes the supplied (wire) buffer into unsuback data
+  * @param packetid returned integer - the MQTT packet identifier
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @return error code.  1 is success, 0 is failure
+  */
+int MQTTDeserialize_unsuback(unsigned short* packetid, unsigned char* buf, int buflen)
+{
+	unsigned char type = 0;
+	unsigned char dup = 0;
+	int rc = 0;
+
+	FUNC_ENTRY;
+	rc = MQTTDeserialize_ack(&type, &dup, packetid, buf, buflen);
+	if (type == UNSUBACK)
+		rc = 1;
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/MQTTUnsubscribeServer.c

@@ -0,0 +1,102 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *******************************************************************************/
+
+#include "MQTTPacket.h"
+#include "StackTrace.h"
+
+#include <string.h>
+
+
+/**
+  * Deserializes the supplied (wire) buffer into unsubscribe data
+  * @param dup integer returned - the MQTT dup flag
+  * @param packetid integer returned - the MQTT packet identifier
+  * @param maxcount - the maximum number of members allowed in the topicFilters and requestedQoSs arrays
+  * @param count - number of members in the topicFilters and requestedQoSs arrays
+  * @param topicFilters - array of topic filter names
+  * @param buf the raw buffer data, of the correct length determined by the remaining length field
+  * @param buflen the length in bytes of the data in the supplied buffer
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTDeserialize_unsubscribe(unsigned char* dup, unsigned short* packetid, int maxcount, int* count, MQTTString topicFilters[],
+		unsigned char* buf, int len)
+{
+	MQTTHeader header = {0};
+	unsigned char* curdata = buf;
+	unsigned char* enddata = NULL;
+	int rc = 0;
+	int mylen = 0;
+
+	FUNC_ENTRY;
+	header.byte = readChar(&curdata);
+	if (header.bits.type != UNSUBSCRIBE)
+		goto exit;
+	*dup = header.bits.dup;
+
+	curdata += (rc = MQTTPacket_decodeBuf(curdata, &mylen)); /* read remaining length */
+	enddata = curdata + mylen;
+
+	*packetid = readInt(&curdata);
+
+	*count = 0;
+	while (curdata < enddata)
+	{
+		if (!readMQTTLenString(&topicFilters[*count], &curdata, enddata))
+			goto exit;
+		(*count)++;
+	}
+
+	rc = 1;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+
+/**
+  * Serializes the supplied unsuback data into the supplied buffer, ready for sending
+  * @param buf the buffer into which the packet will be serialized
+  * @param buflen the length in bytes of the supplied buffer
+  * @param packetid integer - the MQTT packet identifier
+  * @return the length of the serialized data.  <= 0 indicates error
+  */
+int MQTTSerialize_unsuback(unsigned char* buf, int buflen, unsigned short packetid)
+{
+	MQTTHeader header = {0};
+	int rc = 0;
+	unsigned char *ptr = buf;
+
+	FUNC_ENTRY;
+	if (buflen < 2)
+	{
+		rc = MQTTPACKET_BUFFER_TOO_SHORT;
+		goto exit;
+	}
+	header.byte = 0;
+	header.bits.type = UNSUBACK;
+	writeChar(&ptr, header.byte); /* write header */
+
+	ptr += MQTTPacket_encode(ptr, 2); /* write remaining length */
+
+	writeInt(&ptr, packetid);
+
+	rc = ptr - buf;
+exit:
+	FUNC_EXIT_RC(rc);
+	return rc;
+}
+
+

APP_Framework/Applications/app_test/test_mqttclient/mqtt/Makefile

@@ -0,0 +1,13 @@
+ SRC_FILES +=MQTTPacket.c\
+			 MQTTConnectClient.c \
+		     MQTTConnectServer.c \
+			 MQTTDeserializePublish.c \
+			 MQTTFormat.c \
+			 MQTTSerializePublish.c \
+			 MQTTSubscribeClient.c \
+			 MQTTSubscribeServer.c \
+			 MQTTUnsubscribeClient.c \
+			 MQTTUnsubscribeServer.c \
+			 transport.c
+
+include $(KERNEL_ROOT)/compiler.mk

APP_Framework/Applications/app_test/test_mqttclient/mqtt/StackTrace.h

@@ -0,0 +1,78 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    Ian Craggs - initial API and implementation and/or initial documentation
+ *    Ian Craggs - fix for bug #434081
+ *******************************************************************************/
+
+#ifndef STACKTRACE_H_
+#define STACKTRACE_H_
+
+#include <stdio.h>
+#define NOSTACKTRACE 1
+
+#if defined(NOSTACKTRACE)
+#define FUNC_ENTRY
+#define FUNC_ENTRY_NOLOG
+#define FUNC_ENTRY_MED
+#define FUNC_ENTRY_MAX
+#define FUNC_EXIT
+#define FUNC_EXIT_NOLOG
+#define FUNC_EXIT_MED
+#define FUNC_EXIT_MAX
+#define FUNC_EXIT_RC(x)
+#define FUNC_EXIT_MED_RC(x)
+#define FUNC_EXIT_MAX_RC(x)
+
+#else
+
+#if defined(WIN32)
+#define inline __inline
+#define FUNC_ENTRY StackTrace_entry(__FUNCTION__, __LINE__, TRACE_MINIMUM)
+#define FUNC_ENTRY_NOLOG StackTrace_entry(__FUNCTION__, __LINE__, -1)
+#define FUNC_ENTRY_MED StackTrace_entry(__FUNCTION__, __LINE__, TRACE_MEDIUM)
+#define FUNC_ENTRY_MAX StackTrace_entry(__FUNCTION__, __LINE__, TRACE_MAXIMUM)
+#define FUNC_EXIT StackTrace_exit(__FUNCTION__, __LINE__, NULL, TRACE_MINIMUM)
+#define FUNC_EXIT_NOLOG StackTrace_exit(__FUNCTION__, __LINE__, -1)
+#define FUNC_EXIT_MED StackTrace_exit(__FUNCTION__, __LINE__, NULL, TRACE_MEDIUM)
+#define FUNC_EXIT_MAX StackTrace_exit(__FUNCTION__, __LINE__, NULL, TRACE_MAXIMUM)
+#define FUNC_EXIT_RC(x) StackTrace_exit(__FUNCTION__, __LINE__, &x, TRACE_MINIMUM)
+#define FUNC_EXIT_MED_RC(x) StackTrace_exit(__FUNCTION__, __LINE__, &x, TRACE_MEDIUM)
+#define FUNC_EXIT_MAX_RC(x) StackTrace_exit(__FUNCTION__, __LINE__, &x, TRACE_MAXIMUM)
+#else
+#define FUNC_ENTRY StackTrace_entry(__func__, __LINE__, TRACE_MINIMUM)
+#define FUNC_ENTRY_NOLOG StackTrace_entry(__func__, __LINE__, -1)
+#define FUNC_ENTRY_MED StackTrace_entry(__func__, __LINE__, TRACE_MEDIUM)
+#define FUNC_ENTRY_MAX StackTrace_entry(__func__, __LINE__, TRACE_MAXIMUM)
+#define FUNC_EXIT StackTrace_exit(__func__, __LINE__, NULL, TRACE_MINIMUM)
+#define FUNC_EXIT_NOLOG StackTrace_exit(__func__, __LINE__, NULL, -1)
+#define FUNC_EXIT_MED StackTrace_exit(__func__, __LINE__, NULL, TRACE_MEDIUM)
+#define FUNC_EXIT_MAX StackTrace_exit(__func__, __LINE__, NULL, TRACE_MAXIMUM)
+#define FUNC_EXIT_RC(x) StackTrace_exit(__func__, __LINE__, &x, TRACE_MINIMUM)
+#define FUNC_EXIT_MED_RC(x) StackTrace_exit(__func__, __LINE__, &x, TRACE_MEDIUM)
+#define FUNC_EXIT_MAX_RC(x) StackTrace_exit(__func__, __LINE__, &x, TRACE_MAXIMUM)
+
+void StackTrace_entry(const char* name, int line, int trace);
+void StackTrace_exit(const char* name, int line, void* return_value, int trace);
+
+void StackTrace_printStack(FILE* dest);
+char* StackTrace_get(unsigned long);
+
+#endif
+
+#endif
+
+
+
+
+#endif /* STACKTRACE_H_ */

APP_Framework/Applications/app_test/test_mqttclient/mqtt/transport.c

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2022 AIIT XUOS Lab
+ * XiUOS is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *        http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+/**
+ * @file transport.c
+ * @brief mqtt transport function
+ * @version 3.0
+ * @author AIIT XUOS Lab
+ * @date 2023.8.10
+ */
+
+#include "transport.h"
+#include "lwip/opt.h"
+#include "lwip/arch.h"
+#include "lwip/api.h"
+#include "lwip/inet.h"
+#include "lwip/sockets.h"
+#include "string.h"
+
+static int mysock;
+
+
+int32_t transport_sendPacketBuffer( uint8_t* buf, int32_t buflen)
+{
+	int32_t rc;
+	rc = write(mysock, buf, buflen);
+	return rc;
+}
+
+
+int transport_getdata(unsigned char* buf, int count)
+{
+	int32_t rc;
+	
+  rc = recv(mysock, buf, count, 0);
+  lw_print("get data : %lx\n",rc);
+	return rc;
+}
+
+
+int32_t transport_close(void)
+{
+  
+	int rc;
+//	rc = close(mysock);
+  rc = shutdown(mysock, SHUT_WR);
+	rc = recv(mysock, NULL, (size_t)0, 0);
+	rc = close(mysock);
+	return rc;
+}

APP_Framework/Applications/app_test/test_mqttclient/mqtt/transport.h

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2022 AIIT XUOS Lab
+ * XiUOS is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *        http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+/**
+ * @file transport.h
+ * @brief mqtt transport function
+ * @version 3.0
+ * @author AIIT XUOS Lab
+ * @date 2023.8.10
+ */
+
+#ifndef __TRANSPORT_H
+#define __TRANSPORT_H
+
+#include <stdint.h>
+
+
+int32_t transport_sendPacketBuffer( uint8_t* buf, int32_t buflen);
+
+
+int transport_getdata(unsigned char* buf, int count);
+
+
+int32_t transport_open(int8_t* servip, int32_t port);
+
+
+int32_t transport_close(void);
+
+
+
+#endif

APP_Framework/Applications/app_test/test_mqttclient/test_mqttclient.c

@@ -0,0 +1,748 @@
+/*
+ * Copyright (c) 2022 AIIT XUOS Lab
+ * XiUOS is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *        http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+/**
+ * @file test_mqttclient.c
+ * @brief mqtt subscribe fuction test
+ * @version 3.0
+ * @author AIIT XUOS Lab
+ * @date 2023.8.10
+ */
+
+#include <transform.h>
+#include "test_mqttclient.h"
+#include <stdio.h>
+#ifdef ADD_XIZI_FEATURES
+#include <sys_arch.h>
+#include <lwip/sockets.h>
+#include "lwip/sys.h"
+#include "lwip/api.h"
+
+#include "mqtt/MQTTPacket.h"
+#include "mqtt/MQTTSubscribe.h"
+#include "mqtt/transport.h"
+#include <cJSON_Process.h>
+
+#endif
+
+#ifdef ADD_NUTTX_FEATURES
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include "stdio.h"
+#endif
+
+
+#define MQTT_DEMO_BUF_SIZE                    65535
+#define MQTT_DEMO_SEND_TIMES                  20
+#define LWIP_MQTT_DEMO_TASK_STACK_SIZE        4096
+#define LWIP_MQTT_DEMO_TASK_PRIO              20
+
+static char mqtt_demo_ipaddr[] = {192, 168, 130, 77};
+static char mqtt_demo_netmask[] = {255, 255, 254, 0};
+static char mqtt_demo_gwaddr[] = {192, 168, 130, 1};
+
+static pthread_t mqtt_client_task;
+static pthread_t mqtt_server_task;
+
+static uint16_t mqtt_socket_port = 1883;
+static char mqtt_ip_str[128] = {192,168,100,1};
+
+void MqttSocketConfigParam(char *ip_str)
+{
+    int ip1, ip2, ip3, ip4, port = 0;
+
+    if(ip_str == NULL)
+        return;
+
+    if(sscanf(ip_str, "%d.%d.%d.%d:%d", &ip1, &ip2, &ip3, &ip4, &port)) {
+        printf("config ip %s port %d\n", ip_str, port);
+        strcpy(mqtt_ip_str, ip_str);
+        if(port)
+            mqtt_socket_port = port;
+        return;
+    }
+
+    if(sscanf(ip_str, "%d.%d.%d.%d", &ip1, &ip2, &ip3, &ip4)) {
+        printf("config ip %s\n", ip_str);
+        strcpy(mqtt_ip_str, ip_str);
+    }
+}
+
+MQTT_USER_MSG  mqtt_user_msg;
+
+uint8_t MQTT_Connect(void)
+{
+    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
+    uint8_t buf[200];
+    int buflen = sizeof(buf);
+    int len = 0;
+    data.clientID.cstring = CLIENT_ID;                   //随机
+    data.keepAliveInterval = KEEPLIVE_TIME;         //保持活跃
+    data.username.cstring = USER_NAME;              //用户名
+    data.password.cstring = PASSWORD;               //密钥
+    data.MQTTVersion = MQTT_VERSION;                //3表示3.1版本，4表示3.11版本
+    data.cleansession = 1;
+    //组装消息
+    len = MQTTSerialize_connect((unsigned char *)buf, buflen, &data);
+    //发送消息
+    transport_sendPacketBuffer(buf, len);
+    
+    /* 等待连接响应 */
+    if (MQTTPacket_read(buf, buflen, transport_getdata) == CONNACK)
+    {
+        unsigned char sessionPresent, connack_rc;
+        if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0)
+        {
+          lw_print("无法连接，错误代码是: %d！\n", connack_rc);
+            return Connect_NOK;
+        }
+        else 
+        {
+            lw_print("用户名与密钥验证成功，MQTT连接成功！\n");
+            return Connect_OK;
+        }
+    }
+    else
+        lw_print("MQTT连接无响应！\n");
+        return Connect_NOTACK;
+}
+
+
+int32_t MQTT_PingReq(int32_t sock)
+{
+	  int32_t len;
+		uint8_t buf[200];
+		int32_t buflen = sizeof(buf);	 
+		fd_set readfd;
+	  struct timeval tv;
+	  tv.tv_sec = 5;
+	  tv.tv_usec = 0;
+	
+	  FD_ZERO(&readfd);
+	  FD_SET(sock,&readfd);			
+	
+		len = MQTTSerialize_pingreq(buf, buflen);
+		transport_sendPacketBuffer(buf, len);
+	
+		//等待可读事件
+		if(select(sock+1,&readfd,NULL,NULL,&tv) == 0)
+			return -1;
+		
+	  //有可读事件
+		if(FD_ISSET(sock,&readfd) == 0)
+			return -2;
+		
+		if(MQTTPacket_read(buf, buflen, transport_getdata) != PINGRESP)
+			return -3;
+		
+		return 0;
+	
+}
+
+
+/************************************************************************
+** 函数名称: MQTTSubscribe								
+** 函数功能: 订阅消息
+** 入口参数: int32_t sock：套接字
+**           int8_t *topic：主题
+**           enum QoS pos：消息质量
+** 出口参数: >=0:发送成功 <0:发送失败
+** 备    注: 
+************************************************************************/
+int32_t MQTTSubscribe(int32_t sock,char *topic,enum QoS pos)
+{
+	  static uint32_t PacketID = 0;
+	  uint16_t packetidbk = 0;
+	  int32_t conutbk = 0;
+		uint8_t buf[100];
+		int32_t buflen = sizeof(buf);
+	  MQTTString topicString = MQTTString_initializer;  
+		int32_t len;
+	  int32_t req_qos,qosbk;
+
+		fd_set readfd;
+	  struct timeval tv;
+	  tv.tv_sec = 2;
+	  tv.tv_usec = 0;
+	
+	  FD_ZERO(&readfd);
+	  FD_SET(sock,&readfd);		
+	
+	  //复制主题
+    topicString.cstring = (char *)topic;
+		//订阅质量
+	  req_qos = pos;
+	
+	  //串行化订阅消息
+    len = MQTTSerialize_subscribe(buf, buflen, 0, PacketID++, 1, &topicString, &req_qos);
+		//发送TCP数据
+	  if(transport_sendPacketBuffer(buf, len) < 0)
+				return -1;
+	  
+    //等待可读事件--等待超时
+		if(select(sock+1,&readfd,NULL,NULL,&tv) == 0)
+				return -2;
+		//有可读事件--没有可读事件
+		if(FD_ISSET(sock,&readfd) == 0)
+				return -3;
+
+		//等待订阅返回--未收到订阅返回
+		if(MQTTPacket_read(buf, buflen, transport_getdata) != SUBACK)
+				return -4;	
+		
+		//拆订阅回应包
+		if(MQTTDeserialize_suback(&packetidbk,1, &conutbk, &qosbk, buf, buflen) != 1)
+				return -5;
+		
+		//检测返回数据的正确性
+		if((qosbk == 0x80)||(packetidbk != (PacketID-1)))
+				return -6;
+		
+    //订阅成功
+		return 0;
+}
+
+int32_t ReadPacketTimeout(int32_t sock,uint8_t *buf,int32_t buflen,uint32_t timeout)
+{
+		fd_set readfd;
+	  struct timeval tv;
+	  if(timeout != 0)
+		{
+				tv.tv_sec = timeout;
+				tv.tv_usec = 0;
+				FD_ZERO(&readfd);
+				FD_SET(sock,&readfd); 
+
+				
+				if(select(sock+1,&readfd,NULL,NULL,&tv) == 0)
+						return -1;
+				
+				if(FD_ISSET(sock,&readfd) == 0)
+						return -1;
+	  }
+		
+		return MQTTPacket_read(buf, buflen, transport_getdata);
+}
+
+void deliverMessage(MQTTString  *TopicName,MQTTMessage *msg,MQTT_USER_MSG *mqtt_user_msg)
+{
+		//消息质量
+		mqtt_user_msg->msgqos = msg->qos;
+		//保存消息
+		memcpy(mqtt_user_msg->msg,msg->payload,msg->payloadlen);
+		mqtt_user_msg->msg[msg->payloadlen] = 0;
+		//保存消息长度
+		mqtt_user_msg->msglenth = msg->payloadlen;
+		//消息主题
+		memcpy((char *)mqtt_user_msg->topic,TopicName->lenstring.data,TopicName->lenstring.len);
+		mqtt_user_msg->topic[TopicName->lenstring.len] = 0;
+		//消息ID
+		mqtt_user_msg->packetid = msg->id;
+		//标明消息合法
+		mqtt_user_msg->valid = 1;		
+}
+
+void UserMsgCtl(MQTT_USER_MSG  *msg)
+{
+		//这里处理数据只是打印，用户可以在这里添加自己的处理方式
+	  lw_print("****收到订阅的消息******\n");
+		//<2F><><EFBFBD>غ<EFBFBD><D8BA><EFBFBD><EFBFBD><EFBFBD>Ϣ
+	  switch(msg->msgqos)
+		{
+			case 0:
+				    lw_print("MQTT>>消息质量QoS0\n");
+				    break;
+			case 1:
+				    lw_print("MQTT>>消息质量QoS1\n");
+				    break;
+			case 2:
+				    lw_print("MQTT>>消息质量QoS2\n");
+				    break;
+			default:
+				    lw_print("MQTT>>错误的消息质量\n");
+				    break;
+		}
+		lw_print("MQTT>>消息主题：%s\n",msg->topic);	
+		lw_print("MQTT>>消息内容：%s\n",msg->msg);	
+		lw_print("MQTT>>消息长度：%d\n",msg->msglenth);	 
+    Proscess(msg->msg);
+	  //处理完后销毁数据
+	  msg->valid  = 0;
+}
+
+
+void mqtt_pktype_ctl(uint8_t packtype,uint8_t *buf,uint32_t buflen)
+{
+	  MQTTMessage msg;
+		int32_t rc;
+	  MQTTString receivedTopic;
+	  uint32_t len;
+    lw_print("packtype:%d\n",packtype);
+		switch(packtype)
+		{
+			case PUBLISH:
+        
+        if(MQTTDeserialize_publish(&msg.dup,(int*)&msg.qos, &msg.retained, &msg.id, &receivedTopic,
+          (unsigned char **)&msg.payload, &msg.payloadlen, buf, buflen) != 1)
+            return;	
+        
+        deliverMessage(&receivedTopic,&msg,&mqtt_user_msg);
+        
+      
+        if(msg.qos == QOS0)
+        {
+            //QOS0-不需要ACK
+           //直接处理数据
+           UserMsgCtl(&mqtt_user_msg);
+           return;
+        }
+        //发送PUBACK消息
+        if(msg.qos == QOS1)
+        {
+            len =MQTTSerialize_puback(buf,buflen,mqtt_user_msg.packetid);
+            if(len == 0)
+              return;
+             //发送返回
+            if(transport_sendPacketBuffer(buf,len)<0)
+               return;	
+             //返回后处理消息
+            UserMsgCtl(&mqtt_user_msg); 
+            return;												
+        }
+
+        //对于质量2,只需要发送PUBREC就可以了
+        if(msg.qos == QOS2)
+        {
+           len = MQTTSerialize_ack(buf, buflen, PUBREC, 0, mqtt_user_msg.packetid);			                
+           if(len == 0)
+             return;
+            //发送返回
+           transport_sendPacketBuffer(buf,len);	
+        }		
+        break;
+			case  PUBREL:				           
+          //解析包数据，必须包ID相同才可以
+        rc = MQTTDeserialize_ack(&msg.type,&msg.dup, &msg.id, buf,buflen);
+        if((rc != 1)||(msg.type != PUBREL)||(msg.id != mqtt_user_msg.packetid))
+          return ;
+       //收到PUBREL，需要处理并抛弃数据
+        if(mqtt_user_msg.valid == 1)
+        {
+            //返回后处理消息
+           UserMsgCtl(&mqtt_user_msg);
+        }      
+        //串行化PUBCMP消息
+        len = MQTTSerialize_pubcomp(buf,buflen,msg.id);	                   	
+        if(len == 0)
+          return;									
+       //发送返回--PUBCOMP
+        transport_sendPacketBuffer(buf,len);										
+        break;
+			case   PUBACK://等级1客户端推送数据后，服务器返回
+				break;
+			case   PUBREC://等级2客户端推送数据后，服务器返回
+				break;
+			case   PUBCOMP://等级2客户端推送PUBREL后，服务器返回
+        break;
+			default:
+				break;
+		}
+}
+
+
+static void *MqttSocketRecvTask(void *arg)
+{
+MQTT_START: 
+    lw_print("Recv begin**********\n");
+    int fd = -1, clientfd;
+    int recv_len;
+    int ret;
+    char *recv_buf;
+    struct sockaddr_in mqtt_addr;
+    socklen_t addr_len;
+
+     fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (fd < 0) {
+        lw_print("Socket error\n");
+        return NULL;
+    }
+
+    struct sockaddr_in mqtt_sock;
+    mqtt_sock.sin_family = AF_INET;
+    mqtt_sock.sin_port = htons(mqtt_socket_port);
+    mqtt_sock.sin_addr.s_addr = inet_addr(mqtt_ip_str);
+
+    memset(&(mqtt_sock.sin_zero), 0, sizeof(mqtt_sock.sin_zero));
+
+    ret = connect(fd, (struct sockaddr *)&mqtt_sock, sizeof(struct sockaddr));
+
+    if (ret < 0) {
+        lw_print("Unable to connect %s:%d = %d\n", mqtt_ip_str, mqtt_socket_port, ret);
+        close(fd);
+        return NULL;
+    }
+
+    lw_print("MQTT connect %s:%d success, begin to verify username and password.\n", mqtt_ip_str, mqtt_socket_port);
+    
+    if(MQTT_Connect() != Connect_OK)
+    {
+        lw_print("MQTT verify failed.\n");
+        shutdown(fd, SHUT_WR);
+        recv(fd, NULL, (size_t)0, 0);
+        close(fd);
+        PrivTaskDelay(1000);
+        goto MQTT_START;
+    }
+
+    lw_print("MQTT subscribe begin.\n");
+    if(MQTTSubscribe(fd,(char *)TOPIC,QOS1) < 0)
+    {
+        lw_print("MQTT subscribe failed.\n");
+        shutdown(fd, SHUT_WR);
+        recv(fd, NULL, (size_t)0, 0);
+        close(fd);
+        return NULL;
+    }
+
+    lw_print("subscribe success.\n");
+
+    fd_set readfd;
+    uint8_t no_mqtt_msg_exchange = 1;
+    uint8_t buf[MSG_MAX_LEN];
+	int32_t buflen = sizeof(buf);
+    int32_t type;
+    struct timeval tv;    
+	tv.tv_sec = 0;
+	tv.tv_usec = 10;
+
+  int32_t curtick=0;
+
+    while(1)
+    {
+      // lw_print("waiting********\n");
+      curtick +=1;
+      no_mqtt_msg_exchange = 1;
+			
+		  FD_ZERO(&readfd);
+		  FD_SET(fd,&readfd);
+      select(fd+1,&readfd,NULL,NULL,&tv);
+
+      if(FD_ISSET(fd,&readfd) != 0)
+      {
+          
+          type = ReadPacketTimeout(fd,buf,buflen,0);
+          if(type != -1)
+          {
+            lw_print("ctl***********\n");
+              mqtt_pktype_ctl(type,buf,buflen);
+  
+              no_mqtt_msg_exchange = 0;
+								
+					}
+			}
+
+      if( curtick >(2*10000))
+        {
+            curtick =0;
+            //判断是否有数据交换
+            if(no_mqtt_msg_exchange == 0)
+            {
+               //如果有数据交换，这次就不需要发送PING消息
+               continue;
+            }
+            
+            if(MQTT_PingReq(fd) < 0)
+            {
+               //重连服务器
+               lw_print("发送保持活性ping失败....\n");
+               goto CLOSE;	 
+            }
+            
+
+            lw_print("发送保持活性ping作为心跳成功....\n");
+
+            no_mqtt_msg_exchange = 0;
+        } 
+    }
+
+CLOSE:
+	  lw_print("MQTT subscribe failed.\n");
+    shutdown(fd, SHUT_WR);
+    recv(fd, NULL, (size_t)0, 0);
+    close(fd);
+    return NULL;
+
+
+}
+
+
+
+void MqttSocketRecvTest(int argc, char *argv[])
+{
+    if(argc >= 2) {
+        lw_print("lw: [%s] target ip %s\n", __func__, argv[1]);
+        MqttSocketConfigParam(argv[1]);
+    }
+
+
+    // ip4_addr_t dns_ip;
+    // netconn_gethostbyname(HOST_NAME, &dns_ip);
+    // char* host_ip = ip_ntoa(&dns_ip);
+    // lw_print("host name : %s , host_ip : %s\n",HOST_NAME,host_ip);
+    // MqttSocketConfigParam(host_ip);
+
+
+#ifdef ADD_XIZI_FEATURES
+    lwip_config_tcp(0, mqtt_demo_ipaddr, mqtt_demo_netmask, mqtt_demo_gwaddr);
+
+    pthread_attr_t attr;
+    attr.schedparam.sched_priority = LWIP_MQTT_DEMO_TASK_PRIO;
+    attr.stacksize = LWIP_MQTT_DEMO_TASK_STACK_SIZE;
+#endif
+
+#ifdef ADD_NUTTX_FEATURES
+    pthread_attr_t attr = PTHREAD_ATTR_INITIALIZER;
+    attr.priority = LWIP_mqtt_DEMO_TASK_PRIO;
+    attr.stacksize = LWIP_mqtt_DEMO_TASK_STACK_SIZE;
+#endif
+
+    PrivTaskCreate(&mqtt_server_task, &attr, &MqttSocketRecvTask, NULL);
+    PrivTaskStartup(&mqtt_server_task);
+}
+
+PRIV_SHELL_CMD_FUNCTION(MqttSocketRecvTest, a tcp send sample, PRIV_SHELL_CMD_MAIN_ATTR);
+
+
+typedef struct
+{
+  uint8_t  humi_high8bit;                //ԭʼ<D4AD><CABC><EFBFBD>ݣ<EFBFBD>ʪ<EFBFBD>ȸ<EFBFBD>8λ
+  uint8_t  humi_low8bit;                 //ԭʼ<D4AD><CABC><EFBFBD>ݣ<EFBFBD>ʪ<EFBFBD>ȵ<EFBFBD>8λ
+  uint8_t  temp_high8bit;                 //ԭʼ<D4AD><CABC><EFBFBD>ݣ<EFBFBD><DDA3>¶ȸ<C2B6>8λ
+  uint8_t  temp_low8bit;                 //ԭʼ<D4AD><CABC><EFBFBD>ݣ<EFBFBD><DDA3>¶ȸ<C2B6>8λ
+  uint8_t  check_sum;                     //У<><D0A3><EFBFBD>
+  double    humidity;        //ʵ<><CAB5>ʪ<EFBFBD><CAAA>
+  double    temperature;     //ʵ<><CAB5><EFBFBD>¶<EFBFBD>  
+} DHT11_Data_TypeDef;
+
+
+uint16_t GetNextPackID(void)
+{
+	 static uint16_t pubpacketid = 0;
+	 return pubpacketid++;
+}
+
+int32_t WaitForPacket(int32_t sock,uint8_t packettype,uint8_t times)
+{
+	  int32_t type;
+		uint8_t buf[MSG_MAX_LEN];
+	  uint8_t n = 0;
+		int32_t buflen = sizeof(buf);
+		do
+		{
+				//读取数据包
+				type = ReadPacketTimeout(sock,buf,buflen,2);
+			  if(type != -1)
+					mqtt_pktype_ctl(type,buf,buflen);
+				n++;
+		}while((type != packettype)&&(n < times));
+		//收到期望的包
+		if(type == packettype)
+			 return 0;
+		else 
+			 return -1;		
+}
+
+int32_t MQTTMsgPublish(int32_t sock, char *topic, int8_t qos, uint8_t* msg)
+{
+    int8_t retained = 0;      //保留标志位
+    uint32_t msg_len;         //数据长度
+		uint8_t buf[MSG_MAX_LEN];
+		int32_t buflen = sizeof(buf),len;
+		MQTTString topicString = MQTTString_initializer;
+	  uint16_t packid = 0,packetidbk;
+	
+		//填充主题
+	  topicString.cstring = (char *)topic;
+
+	  //填充数据包ID
+	  if((qos == QOS1)||(qos == QOS2))
+		{ 
+			packid = GetNextPackID();
+		}
+		else
+		{
+			  qos = QOS0;
+			  retained = 0;
+			  packid = 0;
+		}
+     
+    msg_len = strlen((char *)msg);
+    
+		//推送消息
+		len = MQTTSerialize_publish(buf, buflen, 0, qos, retained, packid, topicString, (unsigned char*)msg, msg_len);
+		if(len <= 0)
+				return -1;
+		if(transport_sendPacketBuffer(buf, len) < 0)	
+				return -2;	
+		
+		//质量等级0，不需要返回
+		if(qos == QOS0)
+		{
+				return 0;
+		}
+		
+		//等级1
+		if(qos == QOS1)
+		{
+				//等待PUBACK
+			  if(WaitForPacket(sock,PUBACK,5) < 0)
+					 return -3;
+				return 1;
+			  
+		}
+		//等级2
+		if(qos == QOS2)	
+		{
+			  //等待PUBREC
+			  if(WaitForPacket(sock,PUBREC,5) < 0)
+					 return -3;
+			  //发送PUBREL
+        len = MQTTSerialize_pubrel(buf, buflen,0, packetidbk);
+				if(len == 0)
+					return -4;
+				if(transport_sendPacketBuffer(buf, len) < 0)	
+					return -6;			
+			  //等待PUBCOMP
+			  if(WaitForPacket(sock,PUBREC,5) < 0)
+					 return -7;
+				return 2;
+		}
+		//等级错误
+		return -8;
+}
+
+
+static void *MqttSocketSendTask(void *arg)
+{
+
+  int fd = -1, clientfd;
+    int recv_len;
+    int ret;
+    char *recv_buf;
+    struct sockaddr_in mqtt_addr;
+    socklen_t addr_len;
+
+     fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (fd < 0) {
+        lw_print("Socket error\n");
+        return NULL;
+    }
+
+    struct sockaddr_in mqtt_sock;
+    mqtt_sock.sin_family = AF_INET;
+    mqtt_sock.sin_port = htons(mqtt_socket_port);
+    mqtt_sock.sin_addr.s_addr = inet_addr(mqtt_ip_str);
+
+    memset(&(mqtt_sock.sin_zero), 0, sizeof(mqtt_sock.sin_zero));
+
+    ret = connect(fd, (struct sockaddr *)&mqtt_sock, sizeof(struct sockaddr));
+
+    if (ret < 0) {
+        lw_print("Unable to connect %s:%d = %d\n", mqtt_ip_str, mqtt_socket_port, ret);
+        close(fd);
+        return NULL;
+    }
+
+    lw_print("MQTT connect %s:%d success, begin to verify hostname and password.\n", mqtt_ip_str, mqtt_socket_port);
+    
+    if(MQTT_Connect() != Connect_OK)
+    {
+        lw_print("MQTT verify failed.\n");
+        shutdown(fd, SHUT_WR);
+        recv(fd, NULL, (size_t)0, 0);
+        close(fd);
+        return NULL;
+    }
+
+    lw_print("MQTT subscribe begin.\n");
+    if(MQTTSubscribe(fd,(char *)TOPIC,QOS1) < 0)
+    {
+        lw_print("MQTT subscribe failed.\n");
+        shutdown(fd, SHUT_WR);
+        recv(fd, NULL, (size_t)0, 0);
+        close(fd);
+        return NULL;
+    }
+
+    lw_print("subscribe success.\n");
+
+
+
+    uint8_t no_mqtt_msg_exchange = 1;
+    uint32_t curtick=0;
+    uint8_t res;
+
+    cJSON* cJSON_Data = NULL;
+    cJSON_Data = cJSON_Data_Init();
+    DHT11_Data_TypeDef* recv_data;
+
+    double a,b;
+    while(1)
+    {
+      curtick+=1;
+      char* p ="Hello,here is hc";
+      ret = MQTTMsgPublish(fd,(char*)TOPIC,QOS0,(uint8_t*)p);
+      if(ret >= 0)
+      {
+          no_mqtt_msg_exchange = 0;
+          PrivTaskDelay(1000);
+      }
+    }
+}
+
+
+void MqttSocketSendTest(int argc, char *argv[])
+{
+    if(argc >= 2) {
+        lw_print("lw: [%s] target ip %s\n", __func__, argv[1]);
+        MqttSocketConfigParam(argv[1]);
+    }
+
+
+    // ip4_addr_t dns_ip;
+    // netconn_gethostbyname(HOST_NAME, &dns_ip);
+    // char* host_ip = ip_ntoa(&dns_ip);
+    // lw_print("host name : %s , host_ip : %s\n",HOST_NAME,host_ip);
+    // MqttSocketConfigParam(host_ip);
+
+
+#ifdef ADD_XIZI_FEATURES
+    lwip_config_tcp(0, mqtt_demo_ipaddr, mqtt_demo_netmask, mqtt_demo_gwaddr);
+
+    pthread_attr_t attr;
+    attr.schedparam.sched_priority = LWIP_MQTT_DEMO_TASK_PRIO;
+    attr.stacksize = LWIP_MQTT_DEMO_TASK_STACK_SIZE;
+#endif
+
+#ifdef ADD_NUTTX_FEATURES
+    pthread_attr_t attr = PTHREAD_ATTR_INITIALIZER;
+    attr.priority = LWIP_mqtt_DEMO_TASK_PRIO;
+    attr.stacksize = LWIP_mqtt_DEMO_TASK_STACK_SIZE;
+#endif
+
+    PrivTaskCreate(&mqtt_client_task, &attr, &MqttSocketSendTask, NULL);
+    PrivTaskStartup(&mqtt_client_task);
+}
+
+PRIV_SHELL_CMD_FUNCTION(MqttSocketSendTest, a tcp send sample, PRIV_SHELL_CMD_MAIN_ATTR);
+

APP_Framework/Applications/app_test/test_mqttclient/test_mqttclient.h

@@ -0,0 +1,99 @@
+/*
+ * Copyright (c) 2022 AIIT XUOS Lab
+ * XiUOS is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *        http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+/**
+ * @file test_mqttclient.h
+ * @brief mqtt subscribe fuction test
+ * @version 3.0
+ * @author AIIT XUOS Lab
+ * @date 2023.8.10
+ */
+
+#include <stdint.h>
+
+#define   MSG_MAX_LEN     1024
+#define   MSG_TOPIC_LEN   50
+#define   KEEPLIVE_TIME   650
+#define   MQTT_VERSION    4
+
+#ifdef    LWIP_DNS
+#define   HOST_NAME       "iot-06z00im0uwa0ki2.mqtt.iothub.aliyuncs.com"     //服务器域名
+#else
+#define   HOST_NAME       "iot-06z00im0uwa0ki2.mqtt.iothub.aliyuncs.com"     //服务器IP地址
+#endif
+
+
+//#define   HOST_IP       "129.204.201.235"
+#define   HOST_PORT     1883    //由于是TCP连接，端口必须是1883
+
+#define   CLIENT_ID     "iw3rn3pa11K.test|securemode=2,signmethod=hmacsha256,timestamp=1689296035604|"         //随机的id
+#define   USER_NAME     "test&iw3rn3pa11K"     //用户名
+#define   PASSWORD      "7b948d22fe46f0f63d1a403376d26e7cb298abc227d29e44311d7040307a71f8"  //秘钥
+
+
+
+#define   TOPIC         "/iw3rn3pa11K/test/user/Test"      //订阅的主题
+
+#define   TEST_MESSAGE  "test_message"  //发送测试消息
+
+
+// #define   CLIENT_ID     "hc123456789"         //随机的id
+// #define   USER_NAME     "xiuos"     //用户名
+// #define   PASSWORD      "xiuos"  //秘钥
+
+
+enum QoS 
+{ QOS0 = 0, 
+  QOS1, 
+  QOS2 
+};
+
+enum MQTT_Connect
+{
+  Connect_OK = 0,
+  Connect_NOK,
+  Connect_NOTACK
+};
+
+//数据交互结构体
+typedef struct __MQTTMessage
+{
+    uint32_t qos;
+    uint8_t retained;
+    uint8_t dup;
+    uint16_t id;
+	  uint8_t type;
+    void *payload;
+    int32_t payloadlen;
+}MQTTMessage;
+
+//用户接收消息结构体
+typedef struct __MQTT_MSG
+{
+	  uint8_t  msgqos;                 //消息质量
+		uint8_t  msg[MSG_MAX_LEN];       //消息
+	  uint32_t msglenth;               //消息长度
+	  uint8_t  topic[MSG_TOPIC_LEN];   //主题    
+	  uint16_t packetid;               //消息ID
+	  uint8_t  valid;                  //标明消息是否有效
+}MQTT_USER_MSG;
+
+//发送消息结构体
+typedef struct
+{
+    int8_t topic[MSG_TOPIC_LEN];
+    int8_t qos;
+    int8_t retained;
+
+    uint8_t msg[MSG_MAX_LEN];
+    uint8_t msglen;
+} mqtt_recv_msg_t, *p_mqtt_recv_msg_t, mqtt_send_msg_t, *p_mqtt_send_msg_t;

APP_Framework/lib/cJSON/Makefile

@@ -1,3 +1,4 @@
 SRC_FILES :=  cJSON.c
+SRC_FILES +=  cJSON_Process.c
 
 include $(KERNEL_ROOT)/compiler.mk

APP_Framework/lib/cJSON/cJSON_Process.c

@@ -0,0 +1,85 @@
+#include "cJSON_Process.h"
+#include <lwip/sockets.h>
+
+
+
+
+
+cJSON* cJSON_Data_Init(void)
+{
+  cJSON* cJSON_Root = NULL;    //json根节点
+  
+  
+  cJSON_Root = cJSON_CreateObject();  /*创建项目*/
+  if(NULL == cJSON_Root)
+  {
+      return NULL;
+  }
+  cJSON_AddStringToObject(cJSON_Root, NAME, DEFAULT_NAME);  /*添加元素  键值对*/
+  cJSON_AddNumberToObject(cJSON_Root, TEMP_NUM, DEFAULT_TEMP_NUM);
+  cJSON_AddNumberToObject(cJSON_Root, HUM_NUM, DEFAULT_HUM_NUM);
+  
+  char* p = cJSON_Print(cJSON_Root);  /*p 指向的字符串是json格式的*/
+
+  p = NULL;
+  
+  return cJSON_Root;
+  
+}
+uint8_t cJSON_Update(const cJSON * const object,const char * const string,void *d)
+{
+  cJSON* node = NULL;    //json根节点
+  node = cJSON_GetObjectItem(object,string);
+  if(node == NULL)
+    return 0;
+  if(cJSON_IsBool(node))
+  {
+    int *b = (int*)d;
+
+    cJSON_GetObjectItem(object,string)->type = *b ? cJSON_True : cJSON_False;
+
+    return 1;
+  }
+  else if(cJSON_IsString(node))
+  {
+    cJSON_GetObjectItem(object,string)->valuestring = (char*)d;
+
+    return 1;
+  }
+  else if(cJSON_IsNumber(node))
+  {
+    double *num = (double*)d;
+
+    cJSON_GetObjectItem(object,string)->valuedouble = (double)*num;
+
+    return 1;
+  }
+  else
+    return 1;
+}
+
+void Proscess(void* data)
+{
+  
+  cJSON *root,*json_name,*json_temp_num,*json_hum_num;
+  root = cJSON_Parse((char*)data); //解析成json形式
+
+  json_name = cJSON_GetObjectItem( root , NAME);   //获取键值内容
+  json_temp_num = cJSON_GetObjectItem( root , TEMP_NUM );
+  json_hum_num = cJSON_GetObjectItem( root , HUM_NUM );
+
+  lw_print("name:%s\n temp_num:%f\n hum_num:%f\n",
+              json_name->valuestring,
+              json_temp_num->valuedouble,
+              json_hum_num->valuedouble);
+
+  cJSON_Delete(root); //释放内存 
+}
+
+
+
+
+
+
+
+

APP_Framework/lib/cJSON/cJSON_Process.h

@@ -0,0 +1,23 @@
+#ifndef _CJSON_PROCESS_H_
+#define _CJSON_PROCESS_H_
+#include "cJSON.h"
+#include "stdint.h"
+
+
+#define   NAME            "name"     
+#define   TEMP_NUM        "temp"  
+#define   HUM_NUM         "hum" 
+
+#define   DEFAULT_NAME          "fire"     
+#define   DEFAULT_TEMP_NUM       25.0 
+#define   DEFAULT_HUM_NUM        50.0 
+
+
+#define   UPDATE_SUCCESS       1 
+#define   UPDATE_FAIL          0
+
+cJSON* cJSON_Data_Init(void);
+uint8_t cJSON_Update(const cJSON * const object,const char * const string,void * d);
+void Proscess(void* data);
+#endif
+