diff --git a/Makefile b/Makefile
index fc401239..6cb0d501 100755
--- a/Makefile
+++ b/Makefile
@@ -96,7 +96,7 @@ SYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_C}}
TEST_FILES_CS = test3
SYNC_SSL_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_CS}}
-TEST_FILES_A = test4 test6 test9 test_mqtt4async
+TEST_FILES_A = test4 test45 test6 test9 test_mqtt4async
ASYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_A}}
TEST_FILES_AS = test5
diff --git a/src/Clients.h b/src/Clients.h
index cdeba0c4..1bd038d0 100644
--- a/src/Clients.h
+++ b/src/Clients.h
@@ -32,19 +32,8 @@
#include "MQTTClient.h"
#include "LinkedList.h"
#include "MQTTClientPersistence.h"
-/*BE
-include "LinkedList"
-BE*/
-/*BE
-def PUBLICATIONS
-{
- n32 ptr STRING open "topic"
- n32 ptr DATA "payload"
- n32 dec "payloadlen"
- n32 dec "refcount"
-}
-BE*/
+
/**
* Stored publication data to minimize copying
*/
@@ -57,28 +46,6 @@ typedef struct
int refcount;
} Publications;
-/*BE
-// This should get moved to MQTTProtocol, but the includes don't quite work yet
-map MESSAGE_TYPES
-{
- "PUBREC" 5
- "PUBREL" .
- "PUBCOMP" .
-}
-
-
-def MESSAGES
-{
- n32 dec "qos"
- n32 map bool "retain"
- n32 dec "msgid"
- n32 ptr PUBLICATIONS "publish"
- n32 time "lastTouch"
- n8 map MESSAGE_TYPES "nextMessageType"
- n32 dec "len"
-}
-defList(MESSAGES)
-BE*/
/**
* Client publication message data
*/
@@ -95,17 +62,6 @@ typedef struct
int len; /**> length of the whole structure+data */
} Messages;
-
-/*BE
-def WILLMESSAGES
-{
- n32 ptr STRING open "topic"
- n32 ptr DATA open "msg"
- n32 dec "retained"
- n32 dec "qos"
-}
-BE*/
-
/**
* Client will message data
*/
@@ -118,40 +74,6 @@ typedef struct
int qos;
} willMessages;
-/*BE
-map CLIENT_BITS
-{
- "cleansession" 1 : .
- "connected" 2 : .
- "good" 4 : .
- "ping_outstanding" 8 : .
-}
-def CLIENTS
-{
- n32 ptr STRING open "clientID"
- n32 ptr STRING open "username"
- n32 ptr STRING open "password"
- n32 map CLIENT_BITS "bits"
- at 4 n8 bits 7:6 dec "connect_state"
- at 8
- n32 dec "socket"
- n32 ptr "SSL"
- n32 dec "msgID"
- n32 dec "keepAliveInterval"
- n32 dec "maxInflightMessages"
- n32 ptr BRIDGECONNECTIONS "bridge_context"
- n32 time "lastContact"
- n32 ptr WILLMESSAGES "will"
- n32 ptr MESSAGESList open "inboundMsgs"
- n32 ptr MESSAGESList open "outboundMsgs"
- n32 ptr MESSAGESList open "messageQueue"
- n32 dec "discardedMsgs"
-}
-
-defList(CLIENTS)
-
-BE*/
-
typedef struct
{
int socket;
diff --git a/src/Heap.c b/src/Heap.c
index 70468cea..6945a0e1 100644
--- a/src/Heap.c
+++ b/src/Heap.c
@@ -56,6 +56,8 @@ static mutex_type heap_mutex = &heap_mutex_store;
static heap_info state = {0, 0}; /**< global heap state information */
static int eyecatcher = 0x88888888;
+/*#define HEAP_STACK 1 */
+
/**
* Each item on the heap is recorded with this structure.
*/
@@ -65,6 +67,9 @@ typedef struct
int line; /**< the line no in the source file where it was allocated */
void* ptr; /**< pointer to the allocated storage */
size_t size; /**< size of the allocated storage */
+#if defined(HEAP_STACK)
+ char* stack;
+#endif
} storageElement;
static Tree heap; /**< Tree that holds the allocation records */
@@ -168,6 +173,17 @@ void* mymalloc(char* file, int line, size_t size)
}
space += filenamelen;
strcpy(s->file, file);
+#if defined(HEAP_STACK)
+#define STACK_LEN 300
+ if ((s->stack = malloc(STACK_LEN)) == NULL)
+ {
+ Log(LOG_ERROR, 13, errmsg);
+ free(s->file);
+ free(s);
+ return NULL;
+ }
+ StackTrace_get(Thread_getid(), s->stack, STACK_LEN);
+#endif
s->line = line;
/* Add space for eyecatcher at each end */
if ((s->ptr = malloc(size + 2*sizeof(int))) == NULL)
@@ -361,6 +377,9 @@ static void HeapScan(enum LOG_LEVELS log_level)
storageElement* s = (storageElement*)(current->content);
Log(log_level, -1, "Heap element size %d, line %d, file %s, ptr %p", s->size, s->line, s->file, s->ptr);
Log(log_level, -1, " Content %*.s", (10 > current->size) ? s->size : 10, (char*)(((int*)s->ptr) + 1));
+#if defined(HEAP_STACK)
+ Log(log_level, -1, " Stack:\n%s", s->stack);
+#endif
}
Log(log_level, -1, "Heap scan end");
Thread_unlock_mutex(heap_mutex);
diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c
index 9b3d9532..3f6a455b 100644
--- a/src/MQTTAsync.c
+++ b/src/MQTTAsync.c
@@ -253,9 +253,12 @@ typedef struct
int type;
MQTTAsync_onSuccess* onSuccess;
MQTTAsync_onFailure* onFailure;
+ MQTTAsync_onSuccess5* onSuccess5;
+ MQTTAsync_onFailure5* onFailure5;
MQTTAsync_token token;
void* context;
START_TIME_TYPE start_time;
+ MQTTProperties properties;
union
{
struct
@@ -263,6 +266,7 @@ typedef struct
int count;
char** topics;
int* qoss;
+ MQTTSubscribe_options opts;
} sub;
struct
{
@@ -281,6 +285,7 @@ typedef struct
{
int internal;
int timeout;
+ enum MQTTReasonCodes reasonCode;
} dis;
struct
{
@@ -334,6 +339,10 @@ typedef struct MQTTAsync_struct
int retrying;
int reconnectNow;
+ /* MQTT V5 properties */
+ MQTTProperties* connectProps;
+ MQTTProperties* willProps;
+
} MQTTAsyncs;
@@ -373,8 +382,8 @@ static void MQTTAsync_removeResponsesAndCommands(MQTTAsyncs* m);
static int MQTTAsync_completeConnection(MQTTAsyncs* m, MQTTPacket* pack);
static thread_return_type WINAPI MQTTAsync_receiveThread(void* n);
static void MQTTAsync_stop(void);
-static void MQTTAsync_closeOnly(Clients* client);
-static void MQTTAsync_closeSession(Clients* client);
+static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props);
+static void MQTTAsync_closeSession(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props);
static int clientStructCompare(void* a, void* b);
static int MQTTAsync_cleanSession(Clients* client);
static int MQTTAsync_deliverMessage(MQTTAsyncs* m, char* topicName, size_t topicLen, MQTTAsync_message* mm);
@@ -983,7 +992,7 @@ static void MQTTAsync_checkDisconnect(MQTTAsync handle, MQTTAsync_command* comma
if (m->c->outboundMsgs->count == 0 || MQTTAsync_elapsed(command->start_time) >= command->details.dis.timeout)
{
int was_connected = m->c->connected;
- MQTTAsync_closeSession(m->c);
+ MQTTAsync_closeSession(m->c, command->details.dis.reasonCode, &command->properties);
if (command->details.dis.internal)
{
if (m->cl && was_connected)
@@ -1088,6 +1097,7 @@ static void MQTTAsync_freeCommand1(MQTTAsync_queuedCommand *command)
free(command->command.details.pub.payload);
command->command.details.pub.payload = NULL;
}
+ MQTTProperties_free(&command->command.properties);
}
static void MQTTAsync_freeCommand(MQTTAsync_queuedCommand *command)
@@ -1248,7 +1258,7 @@ static int MQTTAsync_processCommand(void)
if (command->client->c->MQTTVersion == MQTTVERSION_DEFAULT)
{
- if (command->command.details.conn.MQTTVersion == 0)
+ if (command->command.details.conn.MQTTVersion == MQTTVERSION_DEFAULT)
command->command.details.conn.MQTTVersion = MQTTVERSION_3_1_1;
else if (command->command.details.conn.MQTTVersion == MQTTVERSION_3_1_1)
command->command.details.conn.MQTTVersion = MQTTVERSION_3_1;
@@ -1256,7 +1266,7 @@ static int MQTTAsync_processCommand(void)
else
command->command.details.conn.MQTTVersion = command->client->c->MQTTVersion;
- Log(TRACE_MIN, -1, "Connecting to serverURI %s with MQTT version %d", serverURI, command->command.details.conn.MQTTVersion);
+ Log(TRACE_PROTOCOL, -1, "Connecting to serverURI %s with MQTT version %d", serverURI, command->command.details.conn.MQTTVersion);
#if defined(OPENSSL)
rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->ssl, command->command.details.conn.MQTTVersion,
NULL, NULL);
@@ -1279,6 +1289,8 @@ static int MQTTAsync_processCommand(void)
{
List* topics = ListInitialize();
List* qoss = ListInitialize();
+ MQTTProperties* props = NULL;
+ MQTTSubscribe_options* subopts = NULL;
int i;
for (i = 0; i < command->command.details.sub.count; i++)
@@ -1286,25 +1298,35 @@ static int MQTTAsync_processCommand(void)
ListAppend(topics, command->command.details.sub.topics[i], strlen(command->command.details.sub.topics[i]));
ListAppend(qoss, &command->command.details.sub.qoss[i], sizeof(int));
}
- rc = MQTTProtocol_subscribe(command->client->c, topics, qoss, command->command.token, NULL, NULL);
+ if (command->client->c->MQTTVersion >= MQTTVERSION_5)
+ {
+ props = &command->command.properties;
+ subopts = &command->command.details.sub.opts;
+ }
+ rc = MQTTProtocol_subscribe(command->client->c, topics, qoss, command->command.token, subopts, props);
ListFreeNoContent(topics);
ListFreeNoContent(qoss);
}
else if (command->command.type == UNSUBSCRIBE)
{
List* topics = ListInitialize();
+ MQTTProperties* props = NULL;
int i;
for (i = 0; i < command->command.details.unsub.count; i++)
ListAppend(topics, command->command.details.unsub.topics[i], strlen(command->command.details.unsub.topics[i]));
- rc = MQTTProtocol_unsubscribe(command->client->c, topics, command->command.token, NULL);
+ if (command->client->c->MQTTVersion >= MQTTVERSION_5)
+ props = &command->command.properties;
+
+ rc = MQTTProtocol_unsubscribe(command->client->c, topics, command->command.token, props);
ListFreeNoContent(topics);
}
else if (command->command.type == PUBLISH)
{
Messages* msg = NULL;
Publish* p = NULL;
+ MQTTProperties initialized = MQTTProperties_initializer;
p = malloc(sizeof(Publish));
@@ -1312,7 +1334,10 @@ static int MQTTAsync_processCommand(void)
p->payloadlen = command->command.details.pub.payloadlen;
p->topic = command->command.details.pub.destinationName;
p->msgId = command->command.token;
- p->MQTTVersion = MQTTVERSION_DEFAULT;
+ p->MQTTVersion = command->client->c->MQTTVersion;
+ p->properties = initialized;
+ if (p->MQTTVersion >= MQTTVERSION_5)
+ p->properties = command->command.properties;
rc = MQTTProtocol_startPublish(command->client->c, p, command->command.details.pub.qos, command->command.details.pub.retained, &msg);
@@ -1446,7 +1471,7 @@ static void nextOrClose(MQTTAsyncs* m, int rc, char* message)
{
MQTTAsync_queuedCommand* conn;
- MQTTAsync_closeOnly(m->c);
+ MQTTAsync_closeOnly(m->c, SUCCESS, NULL);
/* put the connect command back to the head of the command queue, using the next serverURI */
conn = malloc(sizeof(MQTTAsync_queuedCommand));
memset(conn, '\0', sizeof(MQTTAsync_queuedCommand));
@@ -1469,7 +1494,7 @@ static void nextOrClose(MQTTAsyncs* m, int rc, char* message)
}
else
{
- MQTTAsync_closeSession(m->c);
+ MQTTAsync_closeSession(m->c, SUCCESS, NULL);
if (m->connect.onFailure)
{
MQTTAsync_failureData data;
@@ -1700,7 +1725,7 @@ void MQTTAsync_destroy(MQTTAsync* handle)
if (m == NULL)
goto exit;
- MQTTAsync_closeSession(m->c);
+ MQTTAsync_closeSession(m->c, SUCCESS, NULL);
MQTTAsync_removeResponsesAndCommands(m);
ListFree(m->responses);
@@ -1726,6 +1751,18 @@ void MQTTAsync_destroy(MQTTAsync* handle)
if (m->createOptions)
free(m->createOptions);
MQTTAsync_freeServerURIs(m);
+ if (m->connectProps)
+ {
+ MQTTProperties_free(m->connectProps);
+ free(m->connectProps);
+ m->connectProps = NULL;
+ }
+ if (m->willProps)
+ {
+ MQTTProperties_free(m->willProps);
+ free(m->willProps);
+ m->willProps = NULL;
+ }
if (!ListRemove(handles, m))
Log(LOG_ERROR, -1, "free error");
*handle = NULL;
@@ -1741,6 +1778,7 @@ exit:
void MQTTAsync_freeMessage(MQTTAsync_message** message)
{
FUNC_ENTRY;
+ MQTTProperties_free(&(*message)->properties);
free((*message)->payload);
free(*message);
*message = NULL;
@@ -1787,7 +1825,6 @@ static int MQTTAsync_completeConnection(MQTTAsyncs* m, MQTTPacket* pack)
rc = MQTTASYNC_DISCONNECTED;
}
}
- free(connack);
m->pack = NULL;
#if !defined(WIN32) && !defined(WIN64)
Thread_signal_cond(send_cond);
@@ -1852,7 +1889,7 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
else if (m->c->connect_state != 0)
nextOrClose(m, rc, "socket error");
else /* calling disconnect_internal won't have any effect if we're already disconnected */
- MQTTAsync_closeOnly(m->c);
+ MQTTAsync_closeOnly(m->c, SUCCESS, NULL);
}
else
{
@@ -1885,7 +1922,8 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
{
if (pack->header.bits.type == CONNACK)
{
- int sessionPresent = ((Connack*)pack)->flags.bits.sessionPresent;
+ Connack* connack = (Connack*)pack;
+ int sessionPresent = connack->flags.bits.sessionPresent;
int rc = MQTTAsync_completeConnection(m, pack);
if (rc == MQTTASYNC_SUCCESS)
@@ -1894,7 +1932,8 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
if (m->serverURIcount > 0)
Log(TRACE_MIN, -1, "Connect succeeded to %s",
m->serverURIs[m->connect.details.conn.currentURI]);
- onSuccess = (m->connect.onSuccess != NULL); /* save setting of onSuccess callback */
+ onSuccess = (m->connect.onSuccess != NULL ||
+ m->connect.onSuccess5 != NULL); /* save setting of onSuccess callback */
if (m->connect.onSuccess)
{
MQTTAsync_successData data;
@@ -1909,12 +1948,28 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
(*(m->connect.onSuccess))(m->connect.context, &data);
m->connect.onSuccess = NULL; /* don't accidentally call it again */
}
+ else if (m->connect.onSuccess5)
+ {
+ MQTTAsync_successData5 data = MQTTAsync_successData5_initializer;
+ Log(TRACE_MIN, -1, "Calling connect success for client %s", m->c->clientID);
+ if (m->serverURIcount > 0)
+ data.alt.connect.serverURI = m->serverURIs[m->connect.details.conn.currentURI];
+ else
+ data.alt.connect.serverURI = m->serverURI;
+ data.alt.connect.MQTTVersion = m->connect.details.conn.MQTTVersion;
+ data.alt.connect.sessionPresent = sessionPresent;
+ data.props = connack->properties;
+ data.reasonCode = connack->rc;
+ (*(m->connect.onSuccess5))(m->connect.context, &data);
+ m->connect.onSuccess5 = NULL; /* don't accidentally call it again */
+ }
if (m->connected)
{
char* reason = (onSuccess) ? "connect onSuccess called" : "automatic reconnect";
Log(TRACE_MIN, -1, "Calling connected for client %s", m->c->clientID);
(*(m->connected))(m->connected_context, reason);
}
+ MQTTPacket_freeConnack(connack);
}
else
nextOrClose(m, rc, "CONNACK return code");
@@ -1938,7 +1993,45 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
* request, then we call onSuccess with the list of returned QoSs, which inelegantly,
* could include some failures, or worse, the whole list could have failed.
*/
- if (sub->qoss->count == 1 && *(int*)(sub->qoss->first->content) == MQTT_BAD_SUBSCRIBE)
+ if (m->c->MQTTVersion >= MQTTVERSION_5)
+ {
+ if (sub->qoss->count == 1 && *(int*)(sub->qoss->first->content) >= UNSPECIFIED_ERROR)
+ {
+ if (command->command.onFailure5)
+ {
+ MQTTAsync_failureData5 data = MQTTAsync_failureData5_initializer;
+
+ data.token = command->command.token;
+ data.reasonCode = *(int*)(sub->qoss->first->content);
+ data.message = NULL;
+ data.properties = sub->properties;
+ Log(TRACE_MIN, -1, "Calling subscribe failure for client %s", m->c->clientID);
+ (*(command->command.onFailure5))(command->command.context, &data);
+ }
+ }
+ else if (command->command.onSuccess5)
+ {
+ MQTTAsync_successData5 data;
+ int* array = NULL;
+
+ if (sub->qoss->count == 1)
+ data.alt.qos = *(int*)(sub->qoss->first->content);
+ else if (sub->qoss->count > 1)
+ {
+ ListElement* cur_qos = NULL;
+ int* element = array = data.alt.qosList = malloc(sub->qoss->count * sizeof(int));
+ while (ListNextElement(sub->qoss, &cur_qos))
+ *element++ = *(int*)(cur_qos->content);
+ }
+ data.token = command->command.token;
+ data.props = sub->properties;
+ Log(TRACE_MIN, -1, "Calling subscribe success for client %s", m->c->clientID);
+ (*(command->command.onSuccess5))(command->command.context, &data);
+ if (array)
+ free(array);
+ }
+ }
+ else if (sub->qoss->count == 1 && *(int*)(sub->qoss->first->content) == MQTT_BAD_SUBSCRIBE)
{
if (command->command.onFailure)
{
@@ -1980,7 +2073,7 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
else if (pack->header.bits.type == UNSUBACK)
{
ListElement* current = NULL;
- int handleCalled = 0;
+ Unsuback* unsub = (Unsuback*)pack;
/* use the msgid to find the callback to be called */
while (ListNextElement(m->responses, ¤t))
@@ -1990,19 +2083,38 @@ static thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
{
if (!ListDetach(m->responses, command)) /* remove the response from the list */
Log(LOG_ERROR, -1, "Unsubscribe command not removed from command list");
- if (command->command.onSuccess)
+ if (command->command.onSuccess || command->command.onSuccess5)
{
- rc = MQTTProtocol_handleUnsubacks(pack, m->c->net.socket);
- handleCalled = 1;
Log(TRACE_MIN, -1, "Calling unsubscribe success for client %s", m->c->clientID);
- (*(command->command.onSuccess))(command->command.context, NULL);
+ if (command->command.onSuccess)
+ (*(command->command.onSuccess))(command->command.context, NULL);
+ else
+ {
+ MQTTAsync_successData5 data;
+ enum MQTTReasonCodes* array = NULL;
+
+ if (unsub->reasonCodes->count == 1)
+ data.alt.unsub.reasonCode = *(enum MQTTReasonCodes*)(unsub->reasonCodes->first->content);
+ else if (unsub->reasonCodes->count > 1)
+ {
+ ListElement* cur_rc = NULL;
+ enum MQTTReasonCodes* element = array = data.alt.unsub.reasonCodes = malloc(unsub->reasonCodes->count * sizeof(enum MQTTReasonCodes));
+ while (ListNextElement(unsub->reasonCodes, &cur_rc))
+ *element++ = *(enum MQTTReasonCodes*)(cur_rc->content);
+ }
+ data.token = command->command.token;
+ data.props = unsub->properties;
+ Log(TRACE_MIN, -1, "Calling unsubscribe success for client %s", m->c->clientID);
+ (*(command->command.onSuccess5))(command->command.context, &data);
+ if (array)
+ free(array);
+ }
}
MQTTAsync_freeCommand(command);
break;
}
}
- if (!handleCalled)
- rc = MQTTProtocol_handleUnsubacks(pack, m->c->net.socket);
+ rc = MQTTProtocol_handleUnsubacks(pack, m->c->net.socket);
}
}
}
@@ -2112,7 +2224,7 @@ int MQTTAsync_setConnected(MQTTAsync handle, void* context, MQTTAsync_connected*
}
-static void MQTTAsync_closeOnly(Clients* client)
+static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props)
{
FUNC_ENTRY;
client->good = 0;
@@ -2121,7 +2233,7 @@ static void MQTTAsync_closeOnly(Clients* client)
{
MQTTProtocol_checkPendingWrites();
if (client->connected && Socket_noPendingWrites(client->net.socket))
- MQTTPacket_send_disconnect(client, SUCCESS, NULL);
+ MQTTPacket_send_disconnect(client, reasonCode, props);
Thread_lock_mutex(socket_mutex);
#if defined(OPENSSL)
SSLSocket_close(&client->net);
@@ -2139,10 +2251,10 @@ static void MQTTAsync_closeOnly(Clients* client)
}
-static void MQTTAsync_closeSession(Clients* client)
+static void MQTTAsync_closeSession(Clients* client, enum MQTTReasonCodes reasonCode, MQTTProperties* props)
{
FUNC_ENTRY;
- MQTTAsync_closeOnly(client);
+ MQTTAsync_closeOnly(client, reasonCode, props);
if (client->cleansession)
MQTTAsync_cleanSession(client);
@@ -2208,10 +2320,12 @@ static int MQTTAsync_deliverMessage(MQTTAsyncs* m, char* topicName, size_t topic
void Protocol_processPublication(Publish* publish, Clients* client)
{
MQTTAsync_message* mm = NULL;
+ MQTTAsync_message initialized = MQTTAsync_message_initializer;
int rc = 0;
FUNC_ENTRY;
mm = malloc(sizeof(MQTTAsync_message));
+ memcpy(mm, &initialized, sizeof(MQTTAsync_message));
/* If the message is QoS 2, then we have already stored the incoming payload
* in an allocated buffer, so we don't need to copy again.
@@ -2233,6 +2347,9 @@ void Protocol_processPublication(Publish* publish, Clients* client)
mm->dup = publish->header.bits.dup;
mm->msgid = publish->msgId;
+ if (publish->MQTTVersion >= MQTTVERSION_5)
+ mm->properties = MQTTProperties_copy(&publish->properties);
+
if (client->messageQueue->count == 0 && client->connected)
{
ListElement* found = NULL;
@@ -2293,7 +2410,7 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
goto exit;
}
- if (strncmp(options->struct_id, "MQTC", 4) != 0 || options->struct_version < 0 || options->struct_version > 5)
+ if (strncmp(options->struct_id, "MQTC", 4) != 0 || options->struct_version < 0 || options->struct_version > 6)
{
rc = MQTTASYNC_BAD_STRUCTURE;
goto exit;
@@ -2337,6 +2454,8 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
m->connect.onSuccess = options->onSuccess;
m->connect.onFailure = options->onFailure;
+ m->connect.onSuccess5 = options->onSuccess5;
+ m->connect.onFailure5 = options->onFailure5;
m->connect.context = options->context;
m->connectTimeout = options->connectTimeout;
@@ -2363,7 +2482,7 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
if (options->struct_version >= 3)
m->c->MQTTVersion = options->MQTTVersion;
else
- m->c->MQTTVersion = 0;
+ m->c->MQTTVersion = MQTTVERSION_DEFAULT;
if (options->struct_version >= 4)
{
m->automaticReconnect = options->automaticReconnect;
@@ -2499,6 +2618,36 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
m->serverURIs[i] = MQTTStrdup(options->serverURIs[i]);
}
+ if (m->connectProps)
+ {
+ MQTTProperties_free(m->connectProps);
+ free(m->connectProps);
+ m->connectProps = NULL;
+ }
+ if (options->struct_version >=6 && options->connectProperties)
+ {
+ MQTTProperties initialized = MQTTProperties_initializer;
+
+ m->connectProps = malloc(sizeof(MQTTProperties));
+ *m->connectProps = initialized;
+ *m->connectProps = MQTTProperties_copy(options->connectProperties);
+ }
+
+ if (m->willProps)
+ {
+ MQTTProperties_free(m->willProps);
+ free(m->willProps);
+ m->willProps = NULL;
+ }
+ if (options->struct_version >=6 && options->willProperties)
+ {
+ MQTTProperties initialized = MQTTProperties_initializer;
+
+ m->willProps = malloc(sizeof(MQTTProperties));
+ *m->willProps = initialized;
+ *m->willProps = MQTTProperties_copy(options->willProperties);
+ }
+
/* Add connect request to operation queue */
conn = malloc(sizeof(MQTTAsync_queuedCommand));
memset(conn, '\0', sizeof(MQTTAsync_queuedCommand));
@@ -2507,6 +2656,8 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
{
conn->command.onSuccess = options->onSuccess;
conn->command.onFailure = options->onFailure;
+ conn->command.onSuccess5 = options->onSuccess5;
+ conn->command.onFailure5 = options->onFailure5;
conn->command.context = options->context;
}
conn->command.type = CONNECT;
@@ -2547,8 +2698,16 @@ static int MQTTAsync_disconnect1(MQTTAsync handle, const MQTTAsync_disconnectOpt
{
dis->command.onSuccess = options->onSuccess;
dis->command.onFailure = options->onFailure;
+ dis->command.onSuccess5 = options->onSuccess5;
+ dis->command.onFailure5 = options->onFailure5;
dis->command.context = options->context;
dis->command.details.dis.timeout = options->timeout;
+ if (m->c->MQTTVersion >= MQTTVERSION_5 && options->struct_version >= 1)
+ {
+ if (options != NULL)
+ dis->command.properties = MQTTProperties_copy(&options->properties);
+ dis->command.details.dis.reasonCode = options->reasonCode;
+ }
}
dis->command.type = DISCONNECT;
dis->command.details.dis.internal = internal;
@@ -2577,7 +2736,10 @@ void MQTTProtocol_closeSession(Clients* c, int sendwill)
int MQTTAsync_disconnect(MQTTAsync handle, const MQTTAsync_disconnectOptions* options)
{
- return MQTTAsync_disconnect1(handle, options, 0);
+ if (options != NULL && (strncmp(options->struct_id, "MQTD", 4) != 0 || options->struct_version < 0 || options->struct_version > 1))
+ return MQTTASYNC_BAD_STRUCTURE;
+ else
+ return MQTTAsync_disconnect1(handle, options, 0);
}
@@ -2693,8 +2855,15 @@ int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, int
{
sub->command.onSuccess = response->onSuccess;
sub->command.onFailure = response->onFailure;
+ sub->command.onSuccess5 = response->onSuccess5;
+ sub->command.onFailure5 = response->onFailure5;
sub->command.context = response->context;
response->token = sub->command.token;
+ if (m->c->MQTTVersion >= MQTTVERSION_5)
+ {
+ sub->command.properties = MQTTProperties_copy(&response->properties);
+ sub->command.details.sub.opts = response->subscribe_options;
+ }
}
sub->command.type = SUBSCRIBE;
sub->command.details.sub.count = count;
@@ -2767,8 +2936,12 @@ int MQTTAsync_unsubscribeMany(MQTTAsync handle, int count, char* const* topic, M
{
unsub->command.onSuccess = response->onSuccess;
unsub->command.onFailure = response->onFailure;
+ unsub->command.onSuccess5 = response->onSuccess5;
+ unsub->command.onFailure5 = response->onFailure5;
unsub->command.context = response->context;
response->token = unsub->command.token;
+ if (m->c->MQTTVersion >= MQTTVERSION_5)
+ unsub->command.properties = MQTTProperties_copy(&response->properties);
}
unsub->command.details.unsub.count = count;
unsub->command.details.unsub.topics = malloc(sizeof(char*) * count);
@@ -2845,8 +3018,12 @@ int MQTTAsync_send(MQTTAsync handle, const char* destinationName, int payloadlen
{
pub->command.onSuccess = response->onSuccess;
pub->command.onFailure = response->onFailure;
+ pub->command.onSuccess5 = response->onSuccess5;
+ pub->command.onFailure5 = response->onFailure5;
pub->command.context = response->context;
response->token = pub->command.token;
+ if (m->c->MQTTVersion >= MQTTVERSION_5)
+ pub->command.properties = MQTTProperties_copy(&response->properties);
}
pub->command.details.pub.destinationName = MQTTStrdup(destinationName);
pub->command.details.pub.payloadlen = payloadlen;
@@ -2874,7 +3051,8 @@ int MQTTAsync_sendMessage(MQTTAsync handle, const char* destinationName, const M
rc = MQTTASYNC_NULL_PARAMETER;
goto exit;
}
- if (strncmp(message->struct_id, "MQTM", 4) != 0 || message->struct_version != 0)
+ if (strncmp(message->struct_id, "MQTM", 4) != 0 ||
+ (message->struct_version != 0 && message->struct_version != 1))
{
rc = MQTTASYNC_BAD_STRUCTURE;
goto exit;
@@ -2958,7 +3136,8 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
{
rc = MQTTCLIENT_SUCCESS;
m->c->connect_state = 3;
- if (MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion, NULL, NULL) == SOCKET_ERROR)
+ if (MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion,
+ m->connectProps, m->willProps) == SOCKET_ERROR)
{
rc = SOCKET_ERROR;
goto exit;
@@ -2977,7 +3156,8 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
{
#endif
m->c->connect_state = 3; /* TCP/SSL connect completed, in which case send the MQTT connect packet */
- if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion, NULL, NULL)) == SOCKET_ERROR)
+ if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion,
+ m->connectProps, m->willProps)) == SOCKET_ERROR)
goto exit;
#if defined(OPENSSL)
}
@@ -2993,7 +3173,8 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
if(!m->c->cleansession && m->c->session == NULL)
m->c->session = SSL_get1_session(m->c->net.ssl);
m->c->connect_state = 3; /* SSL connect completed, in which case send the MQTT connect packet */
- if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion, NULL, NULL)) == SOCKET_ERROR)
+ if ((rc = MQTTPacket_send_connect(m->c, m->connect.details.conn.MQTTVersion,
+ m->connectProps, m->willProps)) == SOCKET_ERROR)
goto exit;
}
#endif
@@ -3043,7 +3224,7 @@ static MQTTPacket* MQTTAsync_cycle(int* sock, unsigned long timeout, int* rc)
if (m->c->connect_state == 1 || m->c->connect_state == 2)
*rc = MQTTAsync_connecting(m);
else
- pack = MQTTPacket_Factory(MQTTVERSION_DEFAULT, &m->c->net, rc);
+ pack = MQTTPacket_Factory(m->c->MQTTVersion, &m->c->net, rc);
if (m->c->connect_state == 3 && *rc == SOCKET_ERROR)
{
Log(TRACE_MINIMUM, -1, "CONNECT sent but MQTTPacket_Factory has returned SOCKET_ERROR");
diff --git a/src/MQTTAsync.h b/src/MQTTAsync.h
index c83f8ec4..d8abc0a5 100644
--- a/src/MQTTAsync.h
+++ b/src/MQTTAsync.h
@@ -20,6 +20,7 @@
* Ian Craggs - binary will message
* Ian Craggs - binary password
* Ian Craggs - remove const on eyecatchers #168
+ * Ian Craggs - MQTT 5.0
*******************************************************************************/
/********************************************************************/
@@ -28,7 +29,7 @@
* @cond MQTTAsync_main
* @mainpage Asynchronous MQTT client library for C
*
- * © Copyright IBM Corp. 2009, 2017
+ * © Copyright IBM Corp. 2009, 2018
*
* @brief An Asynchronous MQTT client library for C.
*
@@ -105,6 +106,9 @@
/// @endcond
*/
+#include "MQTTProperties.h"
+#include "MQTTReasonCodes.h"
+#include "MQTTSubscribeOpts.h"
#if !defined(NO_PERSISTENCE)
#include "MQTTClientPersistence.h"
#endif
@@ -190,6 +194,10 @@
* MQTT version to connect with: 3.1.1
*/
#define MQTTVERSION_3_1_1 4
+/**
+ * MQTT version to connect with: 5
+ */
+#define MQTTVERSION_5 5
/**
* Bad return code from subscribe, as defined in the 3.1.1 specification
*/
@@ -243,7 +251,8 @@ typedef struct
{
/** The eyecatcher for this structure. must be MQTM. */
char struct_id[4];
- /** The version number of this structure. Must be 0 */
+ /** The version number of this structure. Must be 0 or 1.
+ * 0 indicates no message properties */
int struct_version;
/** The length of the MQTT message payload in bytes. */
int payloadlen;
@@ -293,9 +302,13 @@ typedef struct
* MQTT client and server.
*/
int msgid;
+ /**
+ * The MQTT V5 properties associated with the message.
+ */
+ MQTTProperties properties;
} MQTTAsync_message;
-#define MQTTAsync_message_initializer { {'M', 'Q', 'T', 'M'}, 0, 0, NULL, 0, 0, 0, 0 }
+#define MQTTAsync_message_initializer { {'M', 'Q', 'T', 'M'}, 1, 0, NULL, 0, 0, 0, 0, MQTTProperties_initializer }
/**
* This is a callback function. The client application
@@ -378,7 +391,6 @@ typedef void MQTTAsync_connectionLost(void* context, char* cause);
typedef void MQTTAsync_connected(void* context, char* cause);
-
/** The data returned on completion of an unsuccessful API call in the response callback onFailure. */
typedef struct
{
@@ -390,6 +402,26 @@ typedef struct
const char *message;
} MQTTAsync_failureData;
+
+/** The data returned on completion of an unsuccessful API call in the response callback onFailure. */
+typedef struct
+{
+ /** The eyecatcher for this structure. Will be MQFD. */
+ char struct_id[4];
+ /** The version number of this structure. Will be 0 */
+ int struct_version;
+ /** A token identifying the failed request. */
+ MQTTAsync_token token;
+ /** The MQTT reason code returned. */
+ enum MQTTReasonCodes reasonCode;
+ /** The MQTT properties on the ack, if any. */
+ MQTTProperties properties;
+ /** Optional further text explaining the error. Can be NULL. */
+ const char *message;
+} MQTTAsync_failureData5;
+
+#define MQTTAsync_failureData5_initializer {{'M', 'Q', 'F', 'D'}, 0, 0, SUCCESS, MQTTProperties_initializer, NULL}
+
/** The data returned on completion of a successful API call in the response callback onSuccess. */
typedef struct
{
@@ -418,6 +450,50 @@ typedef struct
} alt;
} MQTTAsync_successData;
+
+/** The data returned on completion of a successful API call in the response callback onSuccess. */
+typedef struct
+{
+ /** The eyecatcher for this structure. Will be MQSD. */
+ char struct_id[4];
+ /** The version number of this structure. Will be 0 */
+ int struct_version;
+ /** A token identifying the successful request. Can be used to refer to the request later. */
+ MQTTAsync_token token;
+ /** MQTT V5 reason code returned */
+ enum MQTTReasonCodes reasonCode;
+ /** MQTT V5 properties returned, if any */
+ MQTTProperties props;
+ /** A union of the different values that can be returned for subscribe, unsubscribe and publish. */
+ union
+ {
+ /** For subscribe, the granted QoS of the subscription returned by the server. */
+ int qos;
+ /** For subscribeMany, the list of granted QoSs of the subscriptions returned by the server. */
+ int* qosList;
+ /** For publish, the message being sent to the server. */
+ struct
+ {
+ MQTTAsync_message message;
+ char* destinationName;
+ } pub;
+ /* For connect, the server connected to, MQTT version used, and sessionPresent flag */
+ struct
+ {
+ char* serverURI;
+ int MQTTVersion;
+ int sessionPresent;
+ } connect;
+ struct
+ {
+ enum MQTTReasonCodes reasonCode;
+ enum MQTTReasonCodes* reasonCodes;
+ } unsub;
+ } alt;
+} MQTTAsync_successData5;
+
+#define MQTTAsync_successData5_initializer {{'M', 'Q', 'S', 'D'}, 0, 0, SUCCESS, MQTTProperties_initializer}
+
/**
* This is a callback function. The client application
* must provide an implementation of this function to enable asynchronous
@@ -430,6 +506,9 @@ typedef struct
*/
typedef void MQTTAsync_onSuccess(void* context, MQTTAsync_successData* response);
+typedef void MQTTAsync_onSuccess5(void* context, MQTTAsync_successData5* response);
+
+
/**
* This is a callback function. The client application
* must provide an implementation of this function to enable asynchronous
@@ -442,11 +521,14 @@ typedef void MQTTAsync_onSuccess(void* context, MQTTAsync_successData* response)
*/
typedef void MQTTAsync_onFailure(void* context, MQTTAsync_failureData* response);
-typedef struct
+typedef void MQTTAsync_onFailure5(void* context, MQTTAsync_failureData5* response);
+
+typedef struct MQTTAsync_responseOptions
{
/** The eyecatcher for this structure. Must be MQTR */
char struct_id[4];
- /** The version number of this structure. Must be 0 */
+ /** The version number of this structure. Must be 0 or 1
+ * if 0, no MQTTV5 options */
int struct_version;
/**
* A pointer to a callback function to be called if the API call successfully
@@ -460,22 +542,41 @@ typedef struct
* completion will be received.
*/
MQTTAsync_onFailure* onFailure;
- /**
+ /**
* A pointer to any application-specific context. The
* the context pointer is passed to success or failure callback functions to
* provide access to the context information in the callback.
*/
void* context;
- /**
+ /**
* A token is returned from the call. It can be used to track
* the state of this request, both in the callbacks and in future calls
* such as ::MQTTAsync_waitForCompletion.
*/
MQTTAsync_token token;
+ /**
+ * A pointer to a callback function to be called if the API call successfully
+ * completes. Can be set to NULL, in which case no indication of successful
+ * completion will be received.
+ */
+ MQTTAsync_onSuccess5* onSuccess5;
+ /**
+ * A pointer to a callback function to be called if the API call successfully
+ * completes. Can be set to NULL, in which case no indication of successful
+ * completion will be received.
+ */
+ MQTTAsync_onFailure5* onFailure5;
+ /**
+ * MQTT V5 input properties
+ */
+ MQTTProperties properties;
+ MQTTSubscribe_options subscribe_options;
} MQTTAsync_responseOptions;
-#define MQTTAsync_responseOptions_initializer { {'M', 'Q', 'T', 'R'}, 0, NULL, NULL, 0, 0 }
+#define MQTTAsync_responseOptions_initializer { {'M', 'Q', 'T', 'R'}, 1, NULL, NULL, 0, 0, NULL, NULL, MQTTProperties_initializer, MQTTSubscribe_options_initializer }
+typedef struct MQTTAsync_responseOptions MQTTAsync_callOptions;
+#define MQTTAsync_callOptions_initializer MQTTAsync_responseOptions_initializer
/**
* This function sets the global callback functions for a specific client.
@@ -724,12 +825,13 @@ typedef struct
{
/** The eyecatcher for this structure. must be MQTC. */
char struct_id[4];
- /** The version number of this structure. Must be 0, 1, 2, 3 4 or 5.
+ /** The version number of this structure. Must be 0, 1, 2, 3 4 5 or 6.
* 0 signifies no SSL options and no serverURIs
* 1 signifies no serverURIs
* 2 signifies no MQTTVersion
* 3 signifies no automatic reconnect options
* 4 signifies no binary password option (just string)
+ * 5 signifies no MQTTV5 properties
*/
int struct_version;
/** The "keep alive" interval, measured in seconds, defines the maximum time
@@ -851,17 +953,37 @@ typedef struct
*/
int maxRetryInterval;
/**
- * Optional binary password. Only checked and used if the password option is NULL
- */
- struct {
- int len; /**< binary password length */
+ * Optional binary password. Only checked and used if the password option is NULL
+ */
+ struct {
+ int len; /**< binary password length */
const void* data; /**< binary password data */
} binarypwd;
+ /**
+ * MQTT V5 properties for connect
+ */
+ MQTTProperties *connectProperties;
+ /**
+ * MQTT V5 properties for the will message in the connect
+ */
+ MQTTProperties *willProperties;
+ /**
+ * A pointer to a callback function to be called if the connect successfully
+ * completes. Can be set to NULL, in which case no indication of successful
+ * completion will be received.
+ */
+ MQTTAsync_onSuccess5* onSuccess5;
+ /**
+ * A pointer to a callback function to be called if the connect fails.
+ * Can be set to NULL, in which case no indication of unsuccessful
+ * completion will be received.
+ */
+ MQTTAsync_onFailure5* onFailure5;
} MQTTAsync_connectOptions;
-#define MQTTAsync_connectOptions_initializer { {'M', 'Q', 'T', 'C'}, 5, 60, 1, 10, NULL, NULL, NULL, 30, 0,\
-NULL, NULL, NULL, NULL, 0, NULL, 0, 0, 1, 60, {0, NULL}}
+#define MQTTAsync_connectOptions_initializer { {'M', 'Q', 'T', 'C'}, 6, 60, 1, 10, NULL, NULL, NULL, 30, 0,\
+NULL, NULL, NULL, NULL, 0, NULL, 0, 0, 1, 60, {0, NULL}, NULL, NULL, NULL, NULL}
/**
* This function attempts to connect a previously-created client (see
@@ -890,7 +1012,7 @@ typedef struct
{
/** The eyecatcher for this structure. Must be MQTD. */
char struct_id[4];
- /** The version number of this structure. Must be 0 or 1. 0 signifies no SSL options */
+ /** The version number of this structure. Must be 0 or 1. 0 signifies no V5 properties */
int struct_version;
/**
* The client delays disconnection for up to this time (in
@@ -915,9 +1037,29 @@ typedef struct
* provide access to the context information in the callback.
*/
void* context;
+ /**
+ * MQTT V5 input properties
+ */
+ MQTTProperties properties;
+ /**
+ * Reason code for MQTTV5 disconnect
+ */
+ enum MQTTReasonCodes reasonCode;
+ /**
+ * A pointer to a callback function to be called if the disconnect successfully
+ * completes. Can be set to NULL, in which case no indication of successful
+ * completion will be received.
+ */
+ MQTTAsync_onSuccess5* onSuccess5;
+ /**
+ * A pointer to a callback function to be called if the disconnect fails.
+ * Can be set to NULL, in which case no indication of unsuccessful
+ * completion will be received.
+ */
+ MQTTAsync_onFailure5* onFailure5;
} MQTTAsync_disconnectOptions;
-#define MQTTAsync_disconnectOptions_initializer { {'M', 'Q', 'T', 'D'}, 0, 0, NULL, NULL, NULL }
+#define MQTTAsync_disconnectOptions_initializer { {'M', 'Q', 'T', 'D'}, 1, 0, NULL, NULL, NULL, MQTTProperties_initializer, MQTTASYNC_SUCCESS }
/**
@@ -1034,8 +1176,11 @@ DLLExport int MQTTAsync_unsubscribeMany(MQTTAsync handle, int count, char* const
* @return ::MQTTASYNC_SUCCESS if the message is accepted for publication.
* An error code is returned if there was a problem accepting the message.
*/
-DLLExport int MQTTAsync_send(MQTTAsync handle, const char* destinationName, int payloadlen, void* payload, int qos, int retained,
- MQTTAsync_responseOptions* response);
+DLLExport int MQTTAsync_send(MQTTAsync handle, const char* destinationName, int payloadlen, void* payload, int qos,
+ int retained, MQTTAsync_responseOptions* response);
+
+DLLExport int MQTTAsync_send5(MQTTAsync handle, const char* destinationName, int payloadlen, void* payload, int qos,
+ int retained, MQTTProperties* props, MQTTAsync_responseOptions* response);
/**
diff --git a/src/MQTTClient.c b/src/MQTTClient.c
index 9ae7c8e0..05ab640c 100644
--- a/src/MQTTClient.c
+++ b/src/MQTTClient.c
@@ -505,6 +505,7 @@ exit:
void MQTTClient_freeMessage(MQTTClient_message** message)
{
FUNC_ENTRY;
+ MQTTProperties_free(&(*message)->properties);
free((*message)->payload);
free(*message);
*message = NULL;
@@ -827,10 +828,12 @@ void Protocol_processPublication(Publish* publish, Clients* client)
{
qEntry* qe = NULL;
MQTTClient_message* mm = NULL;
+ MQTTClient_message initialized = MQTTClient_message_initializer;
FUNC_ENTRY;
qe = malloc(sizeof(qEntry));
mm = malloc(sizeof(MQTTClient_message));
+ memcpy(mm, &initialized, sizeof(MQTTClient_message));
qe->msg = mm;
@@ -858,6 +861,9 @@ void Protocol_processPublication(Publish* publish, Clients* client)
mm->dup = publish->header.bits.dup;
mm->msgid = publish->msgId;
+ if (publish->MQTTVersion >= 5)
+ mm->properties = MQTTProperties_copy(&publish->properties);
+
ListAppend(client->messageQueue, qe, sizeof(qe) + sizeof(mm) + mm->payloadlen + strlen(qe->topicName)+1);
#if !defined(NO_PERSISTENCE)
if (client->persistence)
@@ -1781,9 +1787,10 @@ int MQTTClient_publish(MQTTClient handle, const char* topicName, int payloadlen,
MQTTResponse MQTTClient_publishMessage5(MQTTClient handle, const char* topicName, MQTTClient_message* message,
- MQTTProperties* props, MQTTClient_deliveryToken* deliveryToken)
+ MQTTClient_deliveryToken* deliveryToken)
{
MQTTResponse rc = {MQTTCLIENT_SUCCESS, NULL};
+ MQTTProperties* props = NULL;
FUNC_ENTRY;
if (message == NULL)
@@ -1792,12 +1799,16 @@ MQTTResponse MQTTClient_publishMessage5(MQTTClient handle, const char* topicName
goto exit;
}
- if (strncmp(message->struct_id, "MQTM", 4) != 0 || message->struct_version != 0)
+ if (strncmp(message->struct_id, "MQTM", 4) != 0 ||
+ (message->struct_version != 0 && message->struct_version != 1))
{
rc.reasonCode = MQTTCLIENT_BAD_STRUCTURE;
goto exit;
}
+ if (message->struct_version == 1)
+ props = &message->properties;
+
rc = MQTTClient_publish5(handle, topicName, message->payloadlen, message->payload,
message->qos, message->retained, props, deliveryToken);
exit:
@@ -1809,7 +1820,13 @@ exit:
int MQTTClient_publishMessage(MQTTClient handle, const char* topicName, MQTTClient_message* message,
MQTTClient_deliveryToken* deliveryToken)
{
- MQTTResponse rc = MQTTClient_publishMessage5(handle, topicName, message, NULL, deliveryToken);
+ MQTTResponse rc = {MQTTCLIENT_SUCCESS, NULL};
+
+ if (strncmp(message->struct_id, "MQTM", 4) != 0 ||
+ (message->struct_version != 0 && message->struct_version != 1))
+ return MQTTCLIENT_BAD_STRUCTURE;
+
+ rc = MQTTClient_publishMessage5(handle, topicName, message, deliveryToken);
return rc.reasonCode;
}
diff --git a/src/MQTTClient.h b/src/MQTTClient.h
index 6ad816c0..3ad9273a 100644
--- a/src/MQTTClient.h
+++ b/src/MQTTClient.h
@@ -124,6 +124,7 @@
#include "MQTTProperties.h"
#include "MQTTReasonCodes.h"
+#include "MQTTSubscribeOpts.h"
#if !defined(NO_PERSISTENCE)
#include "MQTTClientPersistence.h"
#endif
@@ -256,7 +257,8 @@ typedef struct
{
/** The eyecatcher for this structure. must be MQTM. */
char struct_id[4];
- /** The version number of this structure. Must be 0 */
+ /** The version number of this structure. Must be 0 or 1
+ * 0 indicates no message properties */
int struct_version;
/** The length of the MQTT message payload in bytes. */
int payloadlen;
@@ -306,9 +308,13 @@ typedef struct
* MQTT client and server.
*/
int msgid;
+ /**
+ * The MQTT V5 properties associated with the message.
+ */
+ MQTTProperties properties;
} MQTTClient_message;
-#define MQTTClient_message_initializer { {'M', 'Q', 'T', 'M'}, 0, 0, NULL, 0, 0, 0, 0 }
+#define MQTTClient_message_initializer { {'M', 'Q', 'T', 'M'}, 1, 0, NULL, 0, 0, 0, 0, MQTTProperties_initializer }
/**
* This is a callback function. The client application
@@ -831,20 +837,6 @@ DLLExport int MQTTClient_isConnected(MQTTClient handle);
DLLExport int MQTTClient_subscribe(MQTTClient handle, const char* topic, int qos);
-typedef struct MQTTSubscribe_options
-{
- /** The eyecatcher for this structure. Must be MQSO. */
- char struct_id[4];
- /** The version number of this structure. Must be 0.
- */
- int struct_version;
- unsigned char noLocal; /* 0 or 1 */
- unsigned char retainAsPublished; /* 0 or 1 */
- unsigned char retainHandling; /* 0, 1 or 2 */
-} MQTTSubscribe_options;
-
-#define MQTTSubscribe_options_initializer { {'M', 'Q', 'S', 'O'}, 0, 0, 0, 0 }
-
DLLExport MQTTResponse MQTTClient_subscribe5(MQTTClient handle, const char* topic, int qos,
MQTTSubscribe_options* opts, MQTTProperties* props);
@@ -948,7 +940,7 @@ DLLExport int MQTTClient_publishMessage(MQTTClient handle, const char* topicName
DLLExport MQTTResponse MQTTClient_publishMessage5(MQTTClient handle, const char* topicName, MQTTClient_message* msg,
- MQTTProperties* properties, MQTTClient_deliveryToken* dt);
+ MQTTClient_deliveryToken* dt);
/**
* This function is called by the client application to synchronize execution
diff --git a/src/MQTTPacket.c b/src/MQTTPacket.c
index d65c6a60..aa0dcbad 100644
--- a/src/MQTTPacket.c
+++ b/src/MQTTPacket.c
@@ -80,7 +80,7 @@ pf new_packets[] =
NULL, /**< MQTTPacket_subscribe*/
MQTTPacket_suback, /**< SUBACK */
NULL, /**< MQTTPacket_unsubscribe*/
- MQTTPacket_ack, /**< UNSUBACK */
+ MQTTPacket_unsuback, /**< UNSUBACK */
MQTTPacket_header_only, /**< PINGREQ */
MQTTPacket_header_only, /**< PINGRESP */
MQTTPacket_header_only /**< DISCONNECT */
@@ -637,6 +637,24 @@ void MQTTPacket_freeSuback(Suback* pack)
}
+/**
+ * Free allocated storage for a suback packet.
+ * @param pack pointer to the suback packet structure
+ */
+void MQTTPacket_freeUnsuback(Unsuback* pack)
+{
+ FUNC_ENTRY;
+ if (pack->MQTTVersion >= MQTTVERSION_5)
+ {
+ MQTTProperties_free(&pack->properties);
+ if (pack->reasonCodes != NULL)
+ ListFree(pack->reasonCodes);
+ }
+ free(pack);
+ FUNC_EXIT;
+}
+
+
/**
* Send an MQTT PUBREC packet down a socket.
* @param msgid the MQTT message id to use
diff --git a/src/MQTTPacket.h b/src/MQTTPacket.h
index 0990d294..c1b71bb1 100644
--- a/src/MQTTPacket.h
+++ b/src/MQTTPacket.h
@@ -32,6 +32,7 @@ typedef unsigned int bool;
typedef void* (*pf)(int, unsigned char, char*, size_t);
#include "MQTTProperties.h"
+#include "MQTTReasonCodes.h"
enum errors
{
@@ -177,6 +178,19 @@ typedef struct
} Suback;
+/**
+ * Data for an MQTT V5 unsuback packet.
+ */
+typedef struct
+{
+ Header header; /**< MQTT header byte */
+ int msgId; /**< MQTT message id */
+ int MQTTVersion; /**< the version of MQTT */
+ MQTTProperties properties; /**< MQTT 5.0 properties. Not used for MQTT < 5.0 */
+ List* reasonCodes; /**< list of reason codes */
+} Unsuback;
+
+
/**
* Data for a publish packet.
*/
@@ -209,7 +223,6 @@ typedef Ack Puback;
typedef Ack Pubrec;
typedef Ack Pubrel;
typedef Ack Pubcomp;
-typedef Ack Unsuback;
int MQTTPacket_encode(char* buf, size_t length);
int MQTTPacket_decode(networkHandles* net, size_t* value);
@@ -237,6 +250,7 @@ int MQTTPacket_send_puback(int msgid, networkHandles* net, const char* clientID)
void* MQTTPacket_ack(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen);
void MQTTPacket_freeSuback(Suback* pack);
+void MQTTPacket_freeUnsuback(Unsuback* pack);
int MQTTPacket_send_pubrec(int msgid, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, const char* clientID);
diff --git a/src/MQTTPacketOut.c b/src/MQTTPacketOut.c
index e35a9fc2..c874e12f 100644
--- a/src/MQTTPacketOut.c
+++ b/src/MQTTPacketOut.c
@@ -56,14 +56,14 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
packet.header.byte = 0;
packet.header.bits.type = CONNECT;
- len = ((MQTTVersion == 3) ? 12 : 10) + (int)strlen(client->clientID)+2;
+ len = ((MQTTVersion == MQTTVERSION_3_1) ? 12 : 10) + (int)strlen(client->clientID)+2;
if (client->will)
len += (int)strlen(client->will->topic)+2 + client->will->payloadlen+2;
if (client->username)
len += (int)strlen(client->username)+2;
if (client->password)
len += client->passwordlen+2;
- if (MQTTVersion >= 5)
+ if (MQTTVersion >= MQTTVERSION_5)
{
len += MQTTProperties_len(connectProperties);
if (client->will && willProperties)
@@ -71,12 +71,12 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
}
ptr = buf = malloc(len);
- if (MQTTVersion == 3)
+ if (MQTTVersion == MQTTVERSION_3_1)
{
writeUTF(&ptr, "MQIsdp");
- writeChar(&ptr, (char)3);
+ writeChar(&ptr, (char)MQTTVERSION_3_1);
}
- else if (MQTTVersion == 4 || MQTTVersion == 5)
+ else if (MQTTVersion == MQTTVERSION_3_1_1 || MQTTVersion == MQTTVERSION_5)
{
writeUTF(&ptr, "MQTT");
writeChar(&ptr, (char)MQTTVersion);
@@ -99,12 +99,12 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
writeChar(&ptr, packet.flags.all);
writeInt(&ptr, client->keepAliveInterval);
- if (MQTTVersion == 5)
+ if (MQTTVersion >= MQTTVERSION_5)
MQTTProperties_write(&ptr, connectProperties);
writeUTF(&ptr, client->clientID);
if (client->will)
{
- if (MQTTVersion == 5)
+ if (MQTTVersion >= MQTTVERSION_5)
MQTTProperties_write(&ptr, willProperties);
writeUTF(&ptr, client->will->topic);
writeData(&ptr, client->will->payload, client->will->payloadlen);
@@ -230,13 +230,13 @@ int MQTTPacket_send_subscribe(List* topics, List* qoss, MQTTSubscribe_options* o
datalen = 2 + topics->count * 3; /* utf length + char qos == 3 */
while (ListNextElement(topics, &elem))
datalen += (int)strlen((char*)(elem->content));
- if (client->MQTTVersion >= 5)
+ if (client->MQTTVersion >= MQTTVERSION_5)
datalen += MQTTProperties_len(props);
ptr = data = malloc(datalen);
writeInt(&ptr, msgid);
- if (client->MQTTVersion >= 5)
+ if (client->MQTTVersion >= MQTTVERSION_5)
MQTTProperties_write(&ptr, props);
elem = NULL;
@@ -247,13 +247,13 @@ int MQTTPacket_send_subscribe(List* topics, List* qoss, MQTTSubscribe_options* o
ListNextElement(qoss, &qosElem);
writeUTF(&ptr, (char*)(elem->content));
subopts = *(int*)(qosElem->content);
- if (client->MQTTVersion >= 5 && opts != NULL)
+ if (client->MQTTVersion >= MQTTVERSION_5 && opts != NULL)
{
subopts |= (opts[i].noLocal << 2); /* 1 bit */
subopts |= (opts[i].retainAsPublished << 3); /* 1 bit */
subopts |= (opts[i].retainHandling << 4); /* 2 bits */
}
- writeChar(&ptr, *(int*)(qosElem->content));
+ writeChar(&ptr, subopts);
++i;
}
rc = MQTTPacket_send(&client->net, header, data, datalen, 1);
@@ -333,13 +333,13 @@ int MQTTPacket_send_unsubscribe(List* topics, MQTTProperties* props, int msgid,
datalen = 2 + topics->count * 2; /* utf length == 2 */
while (ListNextElement(topics, &elem))
datalen += (int)strlen((char*)(elem->content));
- if (client->MQTTVersion >= 5)
+ if (client->MQTTVersion >= MQTTVERSION_5)
datalen += MQTTProperties_len(props);
ptr = data = malloc(datalen);
writeInt(&ptr, msgid);
- if (client->MQTTVersion >= 5)
+ if (client->MQTTVersion >= MQTTVERSION_5)
MQTTProperties_write(&ptr, props);
elem = NULL;
@@ -352,3 +352,46 @@ int MQTTPacket_send_unsubscribe(List* topics, MQTTProperties* props, int msgid,
FUNC_EXIT_RC(rc);
return rc;
}
+
+
+/**
+ * Function used in the new packets table to create unsuback packets.
+ * @param MQTTVersion the version of MQTT
+ * @param aHeader the MQTT header byte
+ * @param data the rest of the packet
+ * @param datalen the length of the rest of the packet
+ * @return pointer to the packet structure
+ */
+void* MQTTPacket_unsuback(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen)
+{
+ Unsuback* pack = malloc(sizeof(Unsuback));
+ char* curdata = data;
+ char* enddata = &data[datalen];
+
+ FUNC_ENTRY;
+ pack->MQTTVersion = MQTTVersion;
+ pack->header.byte = aHeader;
+ pack->msgId = readInt(&curdata);
+ pack->reasonCodes = NULL;
+ if (MQTTVersion >= MQTTVERSION_5)
+ {
+ MQTTProperties props = MQTTProperties_initializer;
+ pack->properties = props;
+ if (MQTTProperties_read(&pack->properties, &curdata, enddata) != 1)
+ {
+ free(pack->properties.array);
+ free(pack);
+ pack = NULL; /* signal protocol error */
+ }
+ pack->reasonCodes = ListInitialize();
+ while ((size_t)(curdata - data) < datalen)
+ {
+ enum MQTTReasonCodes* newrc;
+ newrc = malloc(sizeof(enum MQTTReasonCodes));
+ *newrc = (enum MQTTReasonCodes)readChar(&curdata);
+ ListAppend(pack->reasonCodes, newrc, sizeof(enum MQTTReasonCodes));
+ }
+ }
+ FUNC_EXIT;
+ return pack;
+}
diff --git a/src/MQTTPacketOut.h b/src/MQTTPacketOut.h
index 693e1ae3..4e0ae108 100644
--- a/src/MQTTPacketOut.h
+++ b/src/MQTTPacketOut.h
@@ -34,5 +34,6 @@ int MQTTPacket_send_subscribe(List* topics, List* qoss, MQTTSubscribe_options* o
void* MQTTPacket_suback(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen);
int MQTTPacket_send_unsubscribe(List* topics, MQTTProperties* props, int msgid, int dup, Clients* client);
+void* MQTTPacket_unsuback(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen);
#endif
diff --git a/src/MQTTProperties.c b/src/MQTTProperties.c
index 4e1faccf..b5a8a8c9 100644
--- a/src/MQTTProperties.c
+++ b/src/MQTTProperties.c
@@ -61,7 +61,7 @@ struct nameToType
};
-static char* datadup(MQTTLenString* str)
+static char* datadup(const MQTTLenString* str)
{
char* temp = malloc(str->len);
memcpy(temp, str->data, str->len);
@@ -92,12 +92,14 @@ int MQTTProperties_len(MQTTProperties* props)
}
-int MQTTProperties_add(MQTTProperties* props, MQTTProperty* prop)
+int MQTTProperties_add(MQTTProperties* props, const MQTTProperty* prop)
{
int rc = 0, type;
if ((type = MQTTProperty_getType(prop->identifier)) < 0)
{
+ printf("id %d\n", prop->identifier);
+ StackTrace_printStack(stdout);
rc = MQTT_INVALID_PROPERTY_ID;
goto exit;
}
@@ -201,7 +203,7 @@ int MQTTProperty_write(char** pptr, MQTTProperty* prop)
* @param properties pointer to the property list, can be NULL
* @return whether the write succeeded or not, number of bytes written or < 0
*/
-int MQTTProperties_write(char** pptr, MQTTProperties* properties)
+int MQTTProperties_write(char** pptr, const MQTTProperties* properties)
{
int rc = -1;
int i = 0, len = 0;
@@ -281,7 +283,7 @@ int MQTTProperties_read(MQTTProperties* properties, char** pptr, char* enddata)
int remlength = 0;
FUNC_ENTRY;
- properties->count = 0;
+ /* we assume an initialized properties structure */
if (enddata - (*pptr) > 0) /* enough length to read the VBI? */
{
*pptr += MQTTPacket_decodeBuf(*pptr, &remlength);
@@ -303,6 +305,12 @@ int MQTTProperties_read(MQTTProperties* properties, char** pptr, char* enddata)
rc = 1; /* data read successfully */
}
+ if (rc != 1 && properties->array != NULL)
+ {
+ free(properties->array);
+ properties->array = NULL;
+ properties->max_count = properties->count = 0;
+ }
FUNC_EXIT_RC(rc);
return rc;
}
@@ -362,6 +370,9 @@ DLLExport void MQTTProperties_free(MQTTProperties* props)
{
int i = 0;
+ FUNC_ENTRY;
+ if (props == NULL)
+ goto exit;
for (i = 0; i < props->count; ++i)
{
int id = props->array[i].identifier;
@@ -381,34 +392,25 @@ DLLExport void MQTTProperties_free(MQTTProperties* props)
if (props->array)
free(props->array);
memset(props, '\0', sizeof(MQTTProperties)); /* zero all fields */
+exit:
+ FUNC_EXIT;
}
-MQTTProperties MQTTProperties_copy(MQTTProperties* props)
+MQTTProperties MQTTProperties_copy(const MQTTProperties* props)
{
int i = 0;
MQTTProperties result = MQTTProperties_initializer;
- for (i = 0; i > props->count; ++i)
+ FUNC_ENTRY;
+ for (i = 0; i < props->count; ++i)
{
- int id = props->array[i].identifier;
- int type = MQTTProperty_getType(id);
+ int rc = 0;
- MQTTProperties_add(&result, &props->array[i]);
- switch (type)
- {
- case BINARY_DATA:
- case UTF_8_ENCODED_STRING:
- case UTF_8_STRING_PAIR:
- result.array[i].value.data.data = malloc(result.array[i].value.data.len);
- memcpy(result.array[i].value.data.data, props->array[i].value.data.data, props->array[i].value.data.len);
- if (type == UTF_8_STRING_PAIR)
- {
- result.array[i].value.value.data = malloc(result.array[i].value.value.len);
- memcpy(result.array[i].value.value.data, props->array[i].value.value.data, props->array[i].value.value.len);
- }
- break;
- }
+ if ((rc = MQTTProperties_add(&result, &props->array[i])) != 0)
+ Log(LOG_ERROR, -1, "Error from MQTTProperties add %d", rc);
}
+
+ FUNC_EXIT;
return result;
}
diff --git a/src/MQTTProperties.h b/src/MQTTProperties.h
index 25e1bb27..8f835008 100644
--- a/src/MQTTProperties.h
+++ b/src/MQTTProperties.h
@@ -109,14 +109,14 @@ int MQTTProperties_len(MQTTProperties* props);
* @param prop
* @return whether the write succeeded or not, number of bytes written or < 0
*/
-DLLExport int MQTTProperties_add(MQTTProperties* props, MQTTProperty* prop);
+DLLExport int MQTTProperties_add(MQTTProperties* props, const MQTTProperty* prop);
-int MQTTProperties_write(char** pptr, MQTTProperties* properties);
+int MQTTProperties_write(char** pptr, const MQTTProperties* properties);
int MQTTProperties_read(MQTTProperties* properties, char** pptr, char* enddata);
DLLExport void MQTTProperties_free(MQTTProperties* properties);
-MQTTProperties MQTTProperties_copy(MQTTProperties* props);
+MQTTProperties MQTTProperties_copy(const MQTTProperties* props);
#endif /* MQTTPROPERTIES_H */
diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c
index 99aa97c3..67202b2e 100644
--- a/src/MQTTProtocolClient.c
+++ b/src/MQTTProtocolClient.c
@@ -301,11 +301,16 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
m->msgid = publish->msgId;
m->qos = publish->header.bits.qos;
m->retain = publish->header.bits.retain;
+ m->MQTTVersion = publish->MQTTVersion;
+ if (m->MQTTVersion >= MQTTVERSION_5)
+ m->properties = MQTTProperties_copy(&publish->properties);
m->nextMessageType = PUBREL;
if ( ( listElem = ListFindItem(client->inboundMsgs, &(m->msgid), messageIDCompare) ) != NULL )
{ /* discard queued publication with same msgID that the current incoming message */
Messages* msg = (Messages*)(listElem->content);
MQTTProtocol_removePublication(msg->publish);
+ if (msg->MQTTVersion >= MQTTVERSION_5)
+ MQTTProperties_free(&msg->properties);
ListInsert(client->inboundMsgs, m, sizeof(Messages) + len, listElem);
ListRemove(client->inboundMsgs, msg);
} else
@@ -349,6 +354,8 @@ int MQTTProtocol_handlePubacks(void* pack, int sock)
rc = MQTTPersistence_remove(client, PERSISTENCE_PUBLISH_SENT, m->qos, puback->msgId);
#endif
MQTTProtocol_removePublication(m->publish);
+ if (m->MQTTVersion >= MQTTVERSION_5)
+ MQTTProperties_free(&m->properties);
ListRemove(client->outboundMsgs, m);
}
}
@@ -460,10 +467,15 @@ int MQTTProtocol_handlePubrels(void* pack, int sock)
publish.topiclen = m->publish->topiclen;
publish.payload = m->publish->payload;
publish.payloadlen = m->publish->payloadlen;
+ publish.MQTTVersion = m->MQTTVersion;
+ if (publish.MQTTVersion >= MQTTVERSION_5)
+ publish.properties = m->properties;
Protocol_processPublication(&publish, client);
#if !defined(NO_PERSISTENCE)
rc += MQTTPersistence_remove(client, PERSISTENCE_PUBLISH_RECEIVED, m->qos, pubrel->msgId);
#endif
+ if (m->MQTTVersion >= MQTTVERSION_5)
+ MQTTProperties_free(&m->properties);
ListRemove(&(state.publications), m->publish);
ListRemove(client->inboundMsgs, m);
++(state.msgs_received);
@@ -515,6 +527,8 @@ int MQTTProtocol_handlePubcomps(void* pack, int sock)
rc = MQTTPersistence_remove(client, PERSISTENCE_PUBLISH_SENT, m->qos, pubcomp->msgId);
#endif
MQTTProtocol_removePublication(m->publish);
+ if (m->MQTTVersion >= MQTTVERSION_5)
+ MQTTProperties_free(&m->properties);
ListRemove(client->outboundMsgs, m);
(++state.msgs_sent);
}
@@ -741,9 +755,9 @@ void MQTTProtocol_emptyMessageList(List* msgList)
while (ListNextElement(msgList, ¤t))
{
Messages* m = (Messages*)(current->content);
+ MQTTProtocol_removePublication(m->publish);
if (m->MQTTVersion >= MQTTVERSION_5)
MQTTProperties_free(&m->properties);
- MQTTProtocol_removePublication(m->publish);
}
ListEmpty(msgList);
FUNC_EXIT;
diff --git a/src/MQTTProtocolOut.c b/src/MQTTProtocolOut.c
index f1b4865c..86d03dcd 100644
--- a/src/MQTTProtocolOut.c
+++ b/src/MQTTProtocolOut.c
@@ -240,9 +240,7 @@ int MQTTProtocol_handleUnsubacks(void* pack, int sock)
FUNC_ENTRY;
client = (Clients*)(ListFindItem(bstate->clients, &sock, clientSocketCompare)->content);
Log(LOG_PROTOCOL, 24, NULL, sock, client->clientID, unsuback->msgId);
- if (unsuback->MQTTVersion >= MQTTVERSION_5)
- MQTTProperties_free(&unsuback->properties);
- free(unsuback);
+ MQTTPacket_freeUnsuback(unsuback);
FUNC_EXIT_RC(rc);
return rc;
}
diff --git a/src/MQTTReasonCodes.h b/src/MQTTReasonCodes.h
index c50d0baf..d0928421 100644
--- a/src/MQTTReasonCodes.h
+++ b/src/MQTTReasonCodes.h
@@ -14,6 +14,9 @@
* Ian Craggs - initial API and implementation and/or initial documentation
*******************************************************************************/
+#if !defined(MQTTREASONCODES_H)
+#define MQTTREASONCODES_H
+
enum MQTTReasonCodes {
SUCCESS = 0,
NORMAL_DISCONNECTION = 0,
@@ -61,3 +64,5 @@ enum MQTTReasonCodes {
SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED = 161,
WILDCARD_SUBSCRIPTION_NOT_SUPPORTED = 162
};
+
+#endif
diff --git a/src/MQTTSubscribeOpts.h b/src/MQTTSubscribeOpts.h
new file mode 100644
index 00000000..67b83dbe
--- /dev/null
+++ b/src/MQTTSubscribeOpts.h
@@ -0,0 +1,35 @@
+/*******************************************************************************
+ * Copyright (c) 2018 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(SUBOPTS_H)
+#define SUBOPTS_H
+
+
+typedef struct MQTTSubscribe_options
+{
+ /** The eyecatcher for this structure. Must be MQSO. */
+ char struct_id[4];
+ /** The version number of this structure. Must be 0.
+ */
+ int struct_version;
+ unsigned char noLocal; /* 0 or 1 */
+ unsigned char retainAsPublished; /* 0 or 1 */
+ unsigned char retainHandling; /* 0, 1 or 2 */
+} MQTTSubscribe_options;
+
+#define MQTTSubscribe_options_initializer { {'M', 'Q', 'S', 'O'}, 0, 0, 0, 0 }
+
+#endif
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 5e8eb77c..03b75e5a 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -274,6 +274,26 @@ SET_TESTS_PROPERTIES(
PROPERTIES TIMEOUT 540
)
+ADD_EXECUTABLE(
+ test45
+ test45.c
+)
+
+TARGET_LINK_LIBRARIES(
+ test45
+ paho-mqtt3a
+)
+
+ADD_TEST(
+ NAME test45-1-basic-connect-subscribe-receive
+ COMMAND test45 "--test_no" "1" "--connection" ${MQTT_TEST_BROKER}
+)
+
+SET_TESTS_PROPERTIES(
+ test45-1-basic-connect-subscribe-receive
+ PROPERTIES TIMEOUT 540
+)
+
IF (PAHO_WITH_SSL)
ADD_EXECUTABLE(
test5
diff --git a/test/test15.c b/test/test15.c
index 89375e4b..c8e97e06 100644
--- a/test/test15.c
+++ b/test/test15.c
@@ -297,10 +297,9 @@ void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
char* topicName = NULL;
int topicLen;
int i = 0;
- int iterations = 50;
+ int iterations = 1; //50;
int rc;
MQTTResponse resp;
- MQTTProperties props = MQTTProperties_initializer;
MQTTProperty property;
MyLog(LOGA_DEBUG, "%d messages at QoS %d", iterations, qos);
@@ -314,15 +313,15 @@ void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
property.value.data.len = strlen(property.value.data.data);
property.value.value.data = "test user property value";
property.value.value.len = strlen(property.value.value.data);
- MQTTProperties_add(&props, &property);
+ MQTTProperties_add(&pubmsg.properties, &property);
for (i = 0; i < iterations; ++i)
{
if (i % 10 == 0)
resp = MQTTClient_publish5(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained,
- &props, &dt);
+ &pubmsg.properties, &dt);
else
- resp = MQTTClient_publishMessage5(c, test_topic, &pubmsg, &props, &dt);
+ resp = MQTTClient_publishMessage5(c, test_topic, &pubmsg, &dt);
assert("Good rc from publish", resp.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", resp.reasonCode);
if (qos > 0)
@@ -330,7 +329,6 @@ void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
rc = MQTTClient_waitForCompletion(c, dt, 5000L);
assert("Good rc from waitforCompletion", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
}
-
rc = MQTTClient_receive(c, &topicName, &topicLen, &m, 5000);
assert("Good rc from receive", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
if (topicName)
@@ -343,6 +341,7 @@ void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
MyLog(LOGA_INFO, "Error: wrong data - received lengths %d %d", pubmsg.payloadlen, m->payloadlen);
break;
}
+ assert("Property count should be > 0", m->properties.count > 0, "property count was %d", m->properties.count);
MQTTClient_free(topicName);
MQTTClient_freeMessage(&m);
}
@@ -360,7 +359,7 @@ void test1_sendAndReceive(MQTTClient* c, int qos, char* test_topic)
MQTTClient_receive(c, &topicName, &topicLen, &m, 2000);
}
- MQTTProperties_free(&props);
+ MQTTProperties_free(&pubmsg.properties);
}
void logProperties(MQTTProperties *props)
@@ -389,11 +388,11 @@ void logProperties(MQTTProperties *props)
break;
case BINARY_DATA:
case UTF_8_ENCODED_STRING:
- MyLog(LOGA_INFO, "Property name %s value %*.s", name,
+ MyLog(LOGA_INFO, "Property name %s value %.*s", name,
props->array[i].value.data.len, props->array[i].value.data.data);
break;
case UTF_8_STRING_PAIR:
- MyLog(LOGA_INFO, "Property name %s key %*.s value %*.s", name,
+ MyLog(LOGA_INFO, "Property name %s key %.*s value %.*s", name,
props->array[i].value.data.len, props->array[i].value.data.data,
props->array[i].value.value.len, props->array[i].value.value.data);
break;
@@ -410,7 +409,8 @@ int test1(struct Options options)
MQTTProperties props = MQTTProperties_initializer;
MQTTProperties willProps = MQTTProperties_initializer;
MQTTProperty property;
- MQTTResponse response;
+ MQTTSubscribe_options subopts = MQTTSubscribe_options_initializer;
+ MQTTResponse response = {SUCCESS, NULL};
int rc = 0;
char* test_topic = "C client test1";
@@ -472,10 +472,11 @@ int test1(struct Options options)
MQTTProperties_free(response.properties);
}
+ subopts.retainAsPublished = 1;
property.identifier = SUBSCRIPTION_IDENTIFIER;
property.value.integer4 = 33;
MQTTProperties_add(&props, &property);
- response = MQTTClient_subscribe5(c, test_topic, subsqos, NULL, &props);
+ response = MQTTClient_subscribe5(c, test_topic, subsqos, &subopts, &props);
assert("Good rc from subscribe", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
MQTTProperties_free(&props);
diff --git a/test/test4.c b/test/test4.c
index 0153fc41..efdba2a8 100644
--- a/test/test4.c
+++ b/test/test4.c
@@ -293,6 +293,7 @@ int test1_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync
pubmsg.qos = 2;
pubmsg.retained = 0;
rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
+ assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
}
else
{
@@ -324,6 +325,7 @@ void test1_onSubscribe(void* context, MQTTAsync_successData* response)
pubmsg.retained = 0;
rc = MQTTAsync_send(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, NULL);
+ assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
}
diff --git a/test/test45.c b/test/test45.c
new file mode 100644
index 00000000..d8ce3348
--- /dev/null
+++ b/test/test45.c
@@ -0,0 +1,1874 @@
+/*******************************************************************************
+ * Copyright (c) 2009, 2018 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 - MQTT 3.1.1 support
+ * Ian Craggs - test8 - failure callbacks
+ * Ian Craggs - MQTT V5
+ *******************************************************************************/
+
+
+/**
+ * @file
+ * Tests for the Paho Asynchronous MQTT C client
+ */
+
+
+#include "MQTTAsync.h"
+#include
+#include
+
+#if !defined(_WINDOWS)
+ #include
+ #include
+ #include
+ #include
+#else
+ #include
+#endif
+
+#define ARRAY_SIZE(a) (sizeof(a) / sizeof(a[0]))
+
+void usage(void)
+{
+ printf("help!!\n");
+ exit(EXIT_FAILURE);
+}
+
+struct Options
+{
+ char* connection; /**< connection to system under test. */
+ int verbose;
+ int test_no;
+ int size; /**< size of big message */
+ int MQTTVersion;
+ int iterations;
+} options =
+{
+ "iot.eclipse.org:1883",
+ 0,
+ -1,
+ 10000,
+ MQTTVERSION_5,
+ 1,
+};
+
+void getopts(int argc, char** argv)
+{
+ int count = 1;
+
+ while (count < argc)
+ {
+ if (strcmp(argv[count], "--test_no") == 0)
+ {
+ if (++count < argc)
+ options.test_no = atoi(argv[count]);
+ else
+ usage();
+ }
+ else if (strcmp(argv[count], "--size") == 0)
+ {
+ if (++count < argc)
+ options.size = atoi(argv[count]);
+ else
+ usage();
+ }
+ else if (strcmp(argv[count], "--connection") == 0)
+ {
+ if (++count < argc)
+ options.connection = argv[count];
+ else
+ usage();
+ }
+ else if (strcmp(argv[count], "--MQTTversion") == 0)
+ {
+ if (++count < argc)
+ {
+ options.MQTTVersion = atoi(argv[count]);
+ printf("setting MQTT version to %d\n", options.MQTTVersion);
+ }
+ else
+ usage();
+ }
+ else if (strcmp(argv[count], "--iterations") == 0)
+ {
+ if (++count < argc)
+ options.iterations = atoi(argv[count]);
+ else
+ usage();
+ }
+ else if (strcmp(argv[count], "--verbose") == 0)
+ options.verbose = 1;
+ count++;
+ }
+}
+
+#if 0
+#include /* For general log messages */
+#define MyLog logaLine
+#else
+#define LOGA_DEBUG 0
+#define LOGA_INFO 1
+#include
+#include
+#include
+void MyLog(int LOGA_level, char* format, ...)
+{
+ static char msg_buf[256];
+ va_list args;
+ struct timeb ts;
+
+ struct tm *timeinfo;
+
+ if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
+ return;
+
+ ftime(&ts);
+ timeinfo = localtime(&ts.time);
+ strftime(msg_buf, 80, "%Y%m%d %H%M%S", timeinfo);
+
+ sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm);
+
+ va_start(args, format);
+ vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args);
+ va_end(args);
+
+ printf("%s\n", msg_buf);
+ fflush(stdout);
+}
+#endif
+
+
+#if defined(WIN32) || defined(_WINDOWS)
+#define mqsleep(A) Sleep(1000*A)
+#define START_TIME_TYPE DWORD
+static DWORD start_time = 0;
+START_TIME_TYPE start_clock(void)
+{
+ return GetTickCount();
+}
+#elif defined(AIX)
+#define mqsleep sleep
+#define START_TIME_TYPE struct timespec
+START_TIME_TYPE start_clock(void)
+{
+ static struct timespec start;
+ clock_gettime(CLOCK_REALTIME, &start);
+ return start;
+}
+#else
+#define mqsleep sleep
+#define START_TIME_TYPE struct timeval
+/* TODO - unused - remove? static struct timeval start_time; */
+START_TIME_TYPE start_clock(void)
+{
+ struct timeval start_time;
+ gettimeofday(&start_time, NULL);
+ return start_time;
+}
+#endif
+
+
+#if defined(WIN32)
+long elapsed(START_TIME_TYPE start_time)
+{
+ return GetTickCount() - start_time;
+}
+#elif defined(AIX)
+#define assert(a)
+long elapsed(struct timespec start)
+{
+ struct timespec now, res;
+
+ clock_gettime(CLOCK_REALTIME, &now);
+ ntimersub(now, start, res);
+ return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L;
+}
+#else
+long elapsed(START_TIME_TYPE start_time)
+{
+ struct timeval now, res;
+
+ gettimeofday(&now, NULL);
+ timersub(&now, &start_time, &res);
+ return (res.tv_sec)*1000 + (res.tv_usec)/1000;
+}
+#endif
+
+#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
+#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
+
+int tests = 0;
+int failures = 0;
+FILE* xml;
+START_TIME_TYPE global_start_time;
+char output[3000];
+char* cur_output = output;
+
+void write_test_result(void)
+{
+ long duration = elapsed(global_start_time);
+
+ fprintf(xml, " time=\"%ld.%.3ld\" >\n", duration / 1000, duration % 1000);
+ if (cur_output != output)
+ {
+ fprintf(xml, "%s", output);
+ cur_output = output;
+ }
+ fprintf(xml, "\n");
+}
+
+void myassert(char* filename, int lineno, char* description, int value, char* format, ...)
+{
+ ++tests;
+ if (!value)
+ {
+ va_list args;
+
+ ++failures;
+ printf("Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description);
+
+ va_start(args, format);
+ vprintf(format, args);
+ va_end(args);
+
+ cur_output += sprintf(cur_output, "file %s, line %d \n",
+ description, filename, lineno);
+ }
+ else
+ MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
+}
+
+
+void logProperties(MQTTProperties *props)
+{
+ int i = 0;
+
+ for (i = 0; i < props->count; ++i)
+ {
+ int id = props->array[i].identifier;
+ const char* name = MQTTPropertyName(id);
+ char* intformat = "Property name %s value %d";
+
+ switch (MQTTProperty_getType(id))
+ {
+ case PROPERTY_TYPE_BYTE:
+ MyLog(LOGA_INFO, intformat, name, props->array[i].value.byte);
+ break;
+ case TWO_BYTE_INTEGER:
+ MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer2);
+ break;
+ case FOUR_BYTE_INTEGER:
+ MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4);
+ break;
+ case VARIABLE_BYTE_INTEGER:
+ MyLog(LOGA_INFO, intformat, name, props->array[i].value.integer4);
+ break;
+ case BINARY_DATA:
+ case UTF_8_ENCODED_STRING:
+ MyLog(LOGA_INFO, "Property name %s value len %.*s", name,
+ props->array[i].value.data.len, props->array[i].value.data.data);
+ break;
+ case UTF_8_STRING_PAIR:
+ MyLog(LOGA_INFO, "Property name %s key %.*s value %.*s", name,
+ props->array[i].value.data.len, props->array[i].value.data.data,
+ props->array[i].value.value.len, props->array[i].value.value.data);
+ break;
+ }
+ }
+}
+
+volatile int test_finished = 0;
+
+char* test_topic = "async test topic";
+
+
+void test1_onDisconnect(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
+ test_finished = 1;
+}
+
+
+void test1_onUnsubscribe(void* context, MQTTAsync_successData5* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
+ MQTTProperty property;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In onUnsubscribe onSuccess callback %p", c);
+ MyLog(LOGA_INFO, "Unsuback properties:");
+ logProperties(&response->props);
+
+ opts.onSuccess = test1_onDisconnect;
+ opts.context = c;
+ opts.reasonCode = UNSPECIFIED_ERROR;
+
+ property.identifier = SESSION_EXPIRY_INTERVAL;
+ property.value.integer4 = 0;
+ MQTTProperties_add(&opts.properties, &property);
+
+ rc = MQTTAsync_disconnect(c, &opts);
+ assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ MQTTProperties_free(&opts.properties);
+}
+
+
+int test1_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ static int message_count = 0;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In messageArrived callback %p", c);
+
+ assert("Message structure version should be 1", message->struct_version == 1,
+ "message->struct_version was %d", message->struct_version);
+ if (message->struct_version == 1)
+ {
+ assert("Properties count should be 2", message->properties.count == 2,
+ "Properties count was %d\n", message->properties.count);
+ logProperties(&message->properties);
+ }
+
+ if (++message_count == 1)
+ {
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
+ MQTTProperty property;
+ MQTTProperties props = MQTTProperties_initializer;
+
+ property.identifier = USER_PROPERTY;
+ property.value.data.data = "test user property";
+ property.value.data.len = strlen(property.value.data.data);
+ property.value.value.data = "test user property value";
+ property.value.value.len = strlen(property.value.value.data);
+ MQTTProperties_add(&props, &property);
+ opts.properties = props;
+
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 11;
+ pubmsg.qos = 2;
+ pubmsg.retained = 0;
+ rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
+ assert("Publish should return 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ MQTTProperties_free(&props);
+ }
+ else
+ {
+ MQTTProperty property;
+ MQTTProperties props = MQTTProperties_initializer;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ property.identifier = USER_PROPERTY;
+ property.value.data.data = "test user property";
+ property.value.data.len = strlen(property.value.data.data);
+ property.value.value.data = "test user property value";
+ property.value.value.len = strlen(property.value.value.data);
+ MQTTProperties_add(&props, &property);
+ opts.properties = props;
+
+ opts.onSuccess5 = test1_onUnsubscribe;
+ opts.context = c;
+ rc = MQTTAsync_unsubscribe(c, test_topic, &opts);
+ assert("Unsubscribe successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ MQTTProperties_free(&props);
+ }
+
+ MQTTAsync_freeMessage(&message);
+ MQTTAsync_free(topicName);
+
+ return 1;
+}
+
+void test1_onSubscribe(void* context, MQTTAsync_successData5* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ int rc;
+ MQTTProperty property;
+ MQTTProperties props = MQTTProperties_initializer;
+ MQTTAsync_callOptions opts = MQTTAsync_callOptions_initializer;
+
+ MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p granted qos %d", c, response->alt.qos);
+
+ MyLog(LOGA_INFO, "Suback properties:");
+ logProperties(&response->props);
+
+ property.identifier = USER_PROPERTY;
+ property.value.data.data = "test user property";
+ property.value.data.len = strlen(property.value.data.data);
+ property.value.value.data = "test user property value";
+ property.value.value.len = strlen(property.value.value.data);
+ MQTTProperties_add(&props, &property);
+ opts.properties = props;
+
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 11;
+ pubmsg.qos = 2;
+ pubmsg.retained = 0;
+
+ rc = MQTTAsync_send(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &opts);
+ MQTTProperties_free(&props);
+}
+
+
+void test1_onConnect(void* context, MQTTAsync_successData5* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ MQTTProperty property;
+ MQTTProperties props = MQTTProperties_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
+
+ assert("Reason code should be 0", response->reasonCode == SUCCESS,
+ "Reason code was %d\n", response->reasonCode);
+
+ MyLog(LOGA_INFO, "Connack properties:");
+ logProperties(&response->props);
+
+ opts.onSuccess5 = test1_onSubscribe;
+ opts.context = c;
+
+ property.identifier = SUBSCRIPTION_IDENTIFIER;
+ property.value.integer4 = 33;
+ MQTTProperties_add(&props, &property);
+ opts.properties = props;
+
+ opts.subscribe_options.retainAsPublished = 1;
+
+ rc = MQTTAsync_subscribe(c, test_topic, 2, &opts);
+ assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ test_finished = 1;
+ MQTTProperties_free(&props);
+}
+
+
+/*********************************************************************
+
+Test1: Basic connect, subscribe send and receive.
+
+*********************************************************************/
+int test1(struct Options options)
+{
+ int subsqos = 2;
+ MQTTAsync c;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ MQTTProperties props = MQTTProperties_initializer;
+ MQTTProperties willProps = MQTTProperties_initializer;
+ MQTTProperty property;
+ int rc = 0;
+ char* test_topic = "V5 C client test1";
+
+ MyLog(LOGA_INFO, "Starting V5 test 1 - asynchronous connect");
+ fprintf(xml, "message = "will message";
+ opts.will->qos = 1;
+ opts.will->retained = 0;
+ opts.will->topicName = "will topic";
+ opts.will = NULL;
+ opts.onSuccess5 = test1_onConnect;
+ opts.onFailure = NULL;
+ opts.context = c;
+
+ property.identifier = SESSION_EXPIRY_INTERVAL;
+ property.value.integer4 = 30;
+ MQTTProperties_add(&props, &property);
+
+ property.identifier = USER_PROPERTY;
+ property.value.data.data = "test user property";
+ property.value.data.len = strlen(property.value.data.data);
+ property.value.value.data = "test user property value";
+ property.value.value.len = strlen(property.value.value.data);
+ MQTTProperties_add(&props, &property);
+
+ opts.connectProperties = &props;
+ opts.willProperties = &willProps;
+
+ MyLog(LOGA_DEBUG, "Connecting");
+ rc = MQTTAsync_connect(c, &opts);
+ rc = 0;
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ goto exit;
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+
+ MQTTProperties_free(&props);
+ MQTTProperties_free(&willProps);
+ MQTTAsync_destroy(&c);
+
+exit:
+ MyLog(LOGA_INFO, "TEST1: test %s. %d tests run, %d failures.",
+ (failures == 0) ? "passed" : "failed", tests, failures);
+ write_test_result();
+ return failures;
+}
+
+int test2_onFailure_called = 0;
+
+void test2_onFailure(void* context, MQTTAsync_failureData* response)
+{
+ MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
+
+ test2_onFailure_called++;
+ test_finished = 1;
+}
+
+
+void test2_onConnect(void* context, MQTTAsync_successData* response)
+{
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p\n", context);
+
+ assert("Connect should not succeed", 0, "connect success callback was called", 0);
+
+ test_finished = 1;
+}
+
+/*********************************************************************
+
+Test2: connect timeout
+
+*********************************************************************/
+int test2(struct Options options)
+{
+ int subsqos = 2;
+ MQTTAsync c;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ int rc = 0;
+ char* test_topic = "C client test2";
+
+ test_finished = 0;
+
+ MyLog(LOGA_INFO, "Starting test 2 - connect timeout");
+ fprintf(xml, "message = "will message";
+ opts.will->qos = 1;
+ opts.will->retained = 0;
+ opts.will->topicName = "will topic";
+ opts.will = NULL;
+ opts.onSuccess = test2_onConnect;
+ opts.onFailure = test2_onFailure;
+ opts.context = c;
+
+ MyLog(LOGA_DEBUG, "Connecting");
+ rc = MQTTAsync_connect(c, &opts);
+ rc = 0;
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ goto exit;
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+
+ MQTTAsync_destroy(&c);
+
+exit:
+ assert("Connect onFailure should be called once", test2_onFailure_called == 1,
+ "connect onFailure was called %d times", test2_onFailure_called);
+
+ MyLog(LOGA_INFO, "TEST2: test %s. %d tests run, %d failures.",
+ (failures == 0) ? "passed" : "failed", tests, failures);
+ write_test_result();
+ return failures;
+}
+
+
+typedef struct
+{
+ MQTTAsync c;
+ int index;
+ char clientid[24];
+ char test_topic[100];
+ int message_count;
+} client_data;
+
+
+void test3_onDisconnect(void* context, MQTTAsync_successData* response)
+{
+ client_data* cd = (client_data*)context;
+ MyLog(LOGA_DEBUG, "In onDisconnect callback for client \"%s\"", cd->clientid);
+ test_finished++;
+}
+
+
+void test3_onPublish(void* context, MQTTAsync_successData* response)
+{
+ client_data* cd = (client_data*)context;
+ MyLog(LOGA_DEBUG, "In QoS 0 onPublish callback for client \"%s\"", cd->clientid);
+}
+
+
+void test3_onUnsubscribe(void* context, MQTTAsync_successData* response)
+{
+ client_data* cd = (client_data*)context;
+ MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In onUnsubscribe onSuccess callback \"%s\"", cd->clientid);
+ opts.onSuccess = test3_onDisconnect;
+ opts.context = cd;
+
+ rc = MQTTAsync_disconnect(cd->c, &opts);
+ assert("Disconnect successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+}
+
+
+int test3_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
+{
+ client_data* cd = (client_data*)context;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In messageArrived callback \"%s\" message count ", cd->clientid);
+
+ if (++cd->message_count == 1)
+ {
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 25;
+ pubmsg.qos = 1;
+ pubmsg.retained = 0;
+ rc = MQTTAsync_sendMessage(cd->c, cd->test_topic, &pubmsg, &opts);
+ assert("Good rc from publish", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ }
+ else if (cd->message_count == 2)
+ {
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ pubmsg.payload = "a QoS 0 message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 29;
+ pubmsg.qos = 0;
+ pubmsg.retained = 0;
+ opts.context = cd;
+ opts.onSuccess = test3_onPublish;
+
+ rc = MQTTAsync_sendMessage(cd->c, cd->test_topic, &pubmsg, &opts);
+ assert("Good rc from publish", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ }
+ else
+ {
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ opts.onSuccess = test3_onUnsubscribe;
+ opts.context = cd;
+ rc = MQTTAsync_unsubscribe(cd->c, cd->test_topic, &opts);
+ assert("Unsubscribe successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ }
+ MQTTAsync_freeMessage(&message);
+ MQTTAsync_free(topicName);
+ return 1;
+}
+
+void test3_onSubscribe(void* context, MQTTAsync_successData* response)
+{
+ client_data* cd = (client_data*)context;
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In subscribe onSuccess callback \"%s\"", cd->clientid);
+
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 11;
+ pubmsg.qos = 2;
+ pubmsg.retained = 0;
+
+ rc = MQTTAsync_send(cd->c, cd->test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, NULL);
+ assert("Good rc from publish", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+}
+
+
+void test3_onConnect(void* context, MQTTAsync_successData* response)
+{
+ client_data* cd = (client_data*)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, \"%s\"", cd->clientid);
+ opts.onSuccess = test3_onSubscribe;
+ opts.context = cd;
+
+ rc = MQTTAsync_subscribe(cd->c, cd->test_topic, 2, &opts);
+ assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ test_finished++;
+}
+
+
+void test3_onFailure(void* context, MQTTAsync_failureData* response)
+{
+ client_data* cd = (client_data*)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ assert("Should have connected", 0, "%s failed to connect\n", cd->clientid);
+ MyLog(LOGA_DEBUG, "In connect onFailure callback, \"%s\" rc %d\n", cd->clientid, response ? response->code : -999);
+ if (response && response->message)
+ MyLog(LOGA_DEBUG, "In connect onFailure callback, \"%s\"\n", response->message);
+
+ test_finished++;
+}
+
+
+/*********************************************************************
+
+Test3: More than one client object - simultaneous working.
+
+*********************************************************************/
+int test3(struct Options options)
+{
+ #define num_clients 10
+ int subsqos = 2;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ int rc = 0;
+ int i;
+ client_data clientdata[num_clients];
+
+ test_finished = 0;
+ MyLog(LOGA_INFO, "Starting test 3 - multiple connections");
+ fprintf(xml, "message = "will message";
+ opts.will->qos = 1;
+ opts.will->retained = 0;
+ opts.will->topicName = "will topic";
+ opts.onSuccess = test3_onConnect;
+ opts.onFailure = test3_onFailure;
+ opts.context = &clientdata[i];
+
+ MyLog(LOGA_DEBUG, "Connecting");
+ rc = MQTTAsync_connect(clientdata[i].c, &opts);
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ }
+
+ while (test_finished < num_clients)
+ {
+ MyLog(LOGA_DEBUG, "num_clients %d test_finished %d\n", num_clients, test_finished);
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+ }
+
+ MyLog(LOGA_DEBUG, "TEST3: destroying clients");
+
+ for (i = 0; i < num_clients; ++i)
+ MQTTAsync_destroy(&clientdata[i].c);
+
+//exit:
+ MyLog(LOGA_INFO, "TEST3: test %s. %d tests run, %d failures.",
+ (failures == 0) ? "passed" : "failed", tests, failures);
+ write_test_result();
+ return failures;
+}
+
+
+void* test4_payload = NULL;
+int test4_payloadlen = 0;
+
+void test4_onPublish(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+
+ MyLog(LOGA_DEBUG, "In publish onSuccess callback, context %p", context);
+}
+
+int test4_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ static int message_count = 0;
+ int rc, i;
+
+ MyLog(LOGA_DEBUG, "In messageArrived callback %p", c);
+
+ assert("Message size correct", message->payloadlen == test4_payloadlen,
+ "message size was %d", message->payloadlen);
+
+ for (i = 0; i < options.size; ++i)
+ {
+ if (((char*)test4_payload)[i] != ((char*)message->payload)[i])
+ {
+ assert("Message contents correct", ((char*)test4_payload)[i] != ((char*)message->payload)[i],
+ "message content was %c", ((char*)message->payload)[i]);
+ break;
+ }
+ }
+
+ if (++message_count == 1)
+ {
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ pubmsg.payload = test4_payload;
+ pubmsg.payloadlen = test4_payloadlen;
+ pubmsg.qos = 1;
+ pubmsg.retained = 0;
+ opts.onSuccess = test4_onPublish;
+ opts.context = c;
+
+ rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
+ }
+ else if (message_count == 2)
+ {
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ pubmsg.payload = test4_payload;
+ pubmsg.payloadlen = test4_payloadlen;
+ pubmsg.qos = 0;
+ pubmsg.retained = 0;
+ opts.onSuccess = test4_onPublish;
+ opts.context = c;
+ rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &opts);
+ }
+ else
+ {
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ opts.onSuccess5 = test1_onUnsubscribe;
+ opts.context = c;
+ rc = MQTTAsync_unsubscribe(c, test_topic, &opts);
+ assert("Unsubscribe successful", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ }
+
+ MQTTAsync_freeMessage(&message);
+ MQTTAsync_free(topicName);
+
+ return 1;
+}
+
+
+void test4_onSubscribe(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ int rc, i;
+
+ MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p", c);
+
+ pubmsg.payload = test4_payload = malloc(options.size);
+ pubmsg.payloadlen = test4_payloadlen = options.size;
+
+ srand(33);
+ for (i = 0; i < options.size; ++i)
+ ((char*)pubmsg.payload)[i] = rand() % 256;
+
+ pubmsg.qos = 2;
+ pubmsg.retained = 0;
+
+ rc = MQTTAsync_send(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, NULL);
+}
+
+
+void test4_onConnect(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
+ opts.onSuccess = test4_onSubscribe;
+ opts.context = c;
+
+ rc = MQTTAsync_subscribe(c, test_topic, 2, &opts);
+ assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ test_finished = 1;
+}
+
+
+/*********************************************************************
+
+Test4: Send and receive big messages
+
+*********************************************************************/
+int test4(struct Options options)
+{
+ int subsqos = 2;
+ MQTTAsync c;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ int rc = 0;
+ char* test_topic = "C client test4";
+
+ test_finished = failures = 0;
+ MyLog(LOGA_INFO, "Starting test 4 - big messages");
+ fprintf(xml, "message = "will message";
+ opts.will->qos = 1;
+ opts.will->retained = 0;
+ opts.will->topicName = "will topic";
+ opts.will = NULL;
+ opts.onSuccess = test4_onConnect;
+ opts.onFailure = NULL;
+ opts.context = c;
+
+ MyLog(LOGA_DEBUG, "Connecting");
+ rc = MQTTAsync_connect(c, &opts);
+ rc = 0;
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ goto exit;
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(1000L);
+ #endif
+
+ MQTTAsync_destroy(&c);
+
+exit:
+ MyLog(LOGA_INFO, "TEST4: test %s. %d tests run, %d failures.",
+ (failures == 0) ? "passed" : "failed", tests, failures);
+ write_test_result();
+ return failures;
+}
+
+
+void test5_onConnectFailure(void* context, MQTTAsync_failureData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
+
+ MyLog(LOGA_INFO, "Connack rc is %d", response ? response->code : -999);
+
+ test_finished = 1;
+}
+
+
+void test5_onConnect(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+
+ MyLog(LOGA_DEBUG, "In connect onFailure callback, context %p", context);
+
+ test_finished = 1;
+}
+
+
+/********************************************************************
+
+Test5: Connack return codes
+
+*********************************************************************/
+int test5(struct Options options)
+{
+ int subsqos = 2;
+ MQTTAsync c;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ int rc = 0;
+ char* test_topic = "C client test1";
+
+ test_finished = failures = 0;
+ MyLog(LOGA_INFO, "Starting test 5 - connack return codes");
+ fprintf(xml, "code);
+
+ assert("Should fail to connect", cinfo.should_fail, "should_fail was %d", cinfo.should_fail);
+
+ test_finished = 1;
+}
+
+
+void test6_onConnect(void* context, MQTTAsync_successData* response)
+{
+ test6_client_info cinfo = *(test6_client_info*)context;
+
+ MyLog(LOGA_DEBUG, "In connect success callback, context %p", context);
+
+ assert("Should connect correctly", !cinfo.should_fail, "should_fail was %d", cinfo.should_fail);
+
+ test_finished = 1;
+}
+
+
+/********************************************************************
+
+Test6: HA connections
+
+*********************************************************************/
+int test6(struct Options options)
+{
+ int subsqos = 2;
+ test6_client_info cinfo;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ int rc = 0;
+ char* test_topic = "C client test1";
+ char* uris[2] = {options.connection, options.connection};
+
+ failures = 0;
+ MyLog(LOGA_INFO, "Starting test 6 - HA connections");
+ fprintf(xml, "msgid);
+
+ test7_messageCount++;
+
+ MQTTAsync_freeMessage(&message);
+ MQTTAsync_free(topicName);
+
+ return 1;
+}
+
+
+static int test7_subscribed = 0;
+
+void test7_onSubscribe(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+
+ MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p granted qos %d", c, response->alt.qos);
+
+ test7_subscribed = 1;
+}
+
+
+void test7_onConnect(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
+ opts.onSuccess = test7_onSubscribe;
+ opts.context = c;
+
+ rc = MQTTAsync_subscribe(c, test7_topic, 2, &opts);
+ assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ test_finished = 1;
+}
+
+
+void test7_onConnectOnly(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
+ dopts.context = context;
+ dopts.timeout = 1000;
+ dopts.onSuccess = test7_onDisconnect;
+ rc = MQTTAsync_disconnect(c, &dopts);
+
+ assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ test_finished = 1;
+}
+
+
+/*********************************************************************
+
+Test7: Pending tokens
+
+*********************************************************************/
+int test7(struct Options options)
+{
+ int subsqos = 2;
+ MQTTAsync c;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ MQTTAsync_willOptions wopts = MQTTAsync_willOptions_initializer;
+ int rc = 0;
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_responseOptions ropts = MQTTAsync_responseOptions_initializer;
+ MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer;
+ MQTTAsync_token* tokens = NULL;
+ int msg_count = 6;
+
+ MyLog(LOGA_INFO, "Starting test 7 - pending tokens");
+ fprintf(xml, "message = "will message";
+ opts.will->qos = 1;
+ opts.will->retained = 0;
+ opts.will->topicName = "will topic";
+ opts.will = NULL;
+
+ opts.onFailure = NULL;
+ opts.context = c;
+
+ opts.cleansession = 1;
+ opts.onSuccess = test7_onConnectOnly;
+ MyLog(LOGA_DEBUG, "Connecting to clean up");
+ rc = MQTTAsync_connect(c, &opts);
+ rc = 0;
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ goto exit;
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+
+ test_finished = 0;
+ MyLog(LOGA_DEBUG, "Connecting");
+ opts.cleansession = 0;
+ opts.onSuccess = test7_onConnect;
+ rc = MQTTAsync_connect(c, &opts);
+ rc = 0;
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ goto exit;
+
+ while (!test7_subscribed)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 11;
+ pubmsg.qos = 2;
+ pubmsg.retained = 0;
+ rc = MQTTAsync_send(c, test_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &ropts);
+ MyLog(LOGA_DEBUG, "Token was %d", ropts.token);
+ rc = MQTTAsync_isComplete(c, ropts.token);
+ /*assert("0 rc from isComplete", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);*/
+ rc = MQTTAsync_waitForCompletion(c, ropts.token, 5000L);
+ assert("Good rc from waitForCompletion", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ rc = MQTTAsync_isComplete(c, ropts.token);
+ assert("1 rc from isComplete", rc == 1, "rc was %d", rc);
+
+ test7_messageCount = 0;
+ int i = 0;
+ pubmsg.qos = 2;
+ for (i = 0; i < msg_count; ++i)
+ {
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 11;
+ //pubmsg.qos = (pubmsg.qos == 2) ? 1 : 2;
+ pubmsg.retained = 0;
+ rc = MQTTAsync_sendMessage(c, test_topic, &pubmsg, &ropts);
+ }
+ /* disconnect immediately without receiving the incoming messages */
+ dopts.timeout = 0;
+ dopts.onSuccess = test7_onDisconnect;
+ dopts.context = c;
+ MQTTAsync_disconnect(c, &dopts); /* now there should be "orphaned" publications */
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+ test_finished = 0;
+
+ rc = MQTTAsync_getPendingTokens(c, &tokens);
+ assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+
+ assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
+ MQTTAsync_free(tokens);
+
+ MQTTAsync_destroy(&c); /* force re-reading persistence on create */
+
+ MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
+ rc = MQTTAsync_create(&c, options.connection, "async_test7", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
+ assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ {
+ MQTTAsync_destroy(&c);
+ goto exit;
+ }
+
+ rc = MQTTAsync_getPendingTokens(c, &tokens);
+ assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+
+ assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
+ if (tokens)
+ {
+ int i = 0;
+ while (tokens[i] != -1)
+ MyLog(LOGA_DEBUG, "Delivery token %d", tokens[i++]);
+ MQTTAsync_free(tokens);
+ //The following assertion should work, does with RSMB, but not Mosquitto
+ //assert1("no of tokens should be count", i == msg_count, "no of tokens %d count %d", i, msg_count);
+ }
+
+ rc = MQTTAsync_setCallbacks(c, c, NULL, test7_messageArrived, NULL);
+ assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+
+ MyLog(LOGA_DEBUG, "Reconnecting");
+ opts.context = c;
+ if (MQTTAsync_connect(c, &opts) != 0)
+ {
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ goto exit;
+ }
+
+ #if defined(WIN32)
+ Sleep(5000);
+ #else
+ usleep(5000000L);
+ #endif
+
+ rc = MQTTAsync_getPendingTokens(c, &tokens);
+ assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ /* assert("should get no tokens back", tokens == NULL, "tokens was %p", tokens);
+
+ assert1("no of messages should be count", test7_messageCount == msg_count, "no of tokens %d count %d",
+ test7_messageCount, msg_count);
+
+ assertions fail against Mosquitto - needs testing */
+
+ dopts.onFailure = test7_onDisconnectFailure;
+ dopts.onSuccess = test7_onDisconnect;
+ dopts.timeout = 1000;
+ MQTTAsync_disconnect(c, &dopts);
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+
+ MQTTAsync_destroy(&c);
+
+exit:
+ MyLog(LOGA_INFO, "TEST7: test %s. %d tests run, %d failures.",
+ (failures == 0) ? "passed" : "failed", tests, failures);
+ write_test_result();
+ return failures;
+}
+
+
+
+/*********************************************************************
+
+Test8: Incomplete commands and requests
+
+*********************************************************************/
+
+char* test8_topic = "C client test8";
+int test8_messageCount = 0;
+int test8_subscribed = 0;
+int test8_publishFailures = 0;
+
+void test8_onPublish(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+
+ MyLog(LOGA_DEBUG, "In publish onSuccess callback %p token %d", c, response->token);
+
+}
+
+void test8_onPublishFailure(void* context, MQTTAsync_failureData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MyLog(LOGA_DEBUG, "In onPublish failure callback %p", c);
+
+ assert("Response code should be interrupted", response->code == MQTTASYNC_OPERATION_INCOMPLETE,
+ "rc was %d", response->code);
+
+ test8_publishFailures++;
+}
+
+
+void test8_onDisconnectFailure(void* context, MQTTAsync_failureData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MyLog(LOGA_DEBUG, "In onDisconnect failure callback %p", c);
+
+ assert("Successful disconnect", 0, "disconnect failed", 0);
+
+ test_finished = 1;
+}
+
+
+void test8_onDisconnect(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MyLog(LOGA_DEBUG, "In onDisconnect callback %p", c);
+ test_finished = 1;
+}
+
+
+void test8_onSubscribe(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+
+ MyLog(LOGA_DEBUG, "In subscribe onSuccess callback %p granted qos %d", c, response->alt.qos);
+
+ test8_subscribed = 1;
+}
+
+
+void test8_onConnect(void* context, MQTTAsync_successData* response)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
+ int rc;
+
+ MyLog(LOGA_DEBUG, "In connect onSuccess callback, context %p", context);
+ opts.onSuccess = test8_onSubscribe;
+ opts.context = c;
+
+ rc = MQTTAsync_subscribe(c, test8_topic, 2, &opts);
+ assert("Good rc from subscribe", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ test_finished = 1;
+}
+
+int test8_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
+{
+ MQTTAsync c = (MQTTAsync)context;
+ static int message_count = 0;
+
+ MyLog(LOGA_DEBUG, "Test8: received message id %d", message->msgid);
+
+ test8_messageCount++;
+
+ MQTTAsync_freeMessage(&message);
+ MQTTAsync_free(topicName);
+
+ return 1;
+}
+
+
+int test8(struct Options options)
+{
+ int subsqos = 2;
+ MQTTAsync c;
+ MQTTAsync_connectOptions opts = MQTTAsync_connectOptions_initializer;
+ int rc = 0;
+ MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
+ MQTTAsync_responseOptions ropts = MQTTAsync_responseOptions_initializer;
+ MQTTAsync_disconnectOptions dopts = MQTTAsync_disconnectOptions_initializer;
+ MQTTAsync_token* tokens = NULL;
+ int msg_count = 6;
+
+ MyLog(LOGA_INFO, "Starting test 8 - incomplete commands");
+ fprintf(xml, " 0", test8_publishFailures > 0,
+ "test8_publishFailures = %d", test8_publishFailures);
+
+ /* Now elicit failure callbacks on destroy */
+
+ test8_subscribed = test8_publishFailures = 0;
+
+ MyLog(LOGA_DEBUG, "Connecting");
+ opts.cleansession = 0;
+ opts.onSuccess = test8_onConnect;
+ rc = MQTTAsync_connect(c, &opts);
+ rc = 0;
+ assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ if (rc != MQTTASYNC_SUCCESS)
+ goto exit;
+
+ while (!test8_subscribed)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+
+ i = 0;
+ pubmsg.qos = 2;
+ ropts.onSuccess = test8_onPublish;
+ ropts.onFailure = test8_onPublishFailure;
+ ropts.context = c;
+ for (i = 0; i < msg_count; ++i)
+ {
+ pubmsg.payload = "a much longer message that we can shorten to the extent that we need to payload up to 11";
+ pubmsg.payloadlen = 11;
+ pubmsg.qos = (pubmsg.qos == 2) ? 1 : 2; /* alternate */
+ pubmsg.retained = 0;
+ rc = MQTTAsync_sendMessage(c, test8_topic, &pubmsg, &ropts);
+ assert("Good rc from sendMessage", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ }
+ /* disconnect immediately without completing the commands */
+ dopts.timeout = 0;
+ dopts.onSuccess = test8_onDisconnect;
+ dopts.context = c;
+ rc = MQTTAsync_disconnect(c, &dopts); /* now there should be incomplete commands */
+ assert("Good rc from disconnect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+
+ while (!test_finished)
+ #if defined(WIN32)
+ Sleep(100);
+ #else
+ usleep(10000L);
+ #endif
+ test_finished = 0;
+
+ rc = MQTTAsync_getPendingTokens(c, &tokens);
+ assert("getPendingTokens rc == 0", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
+ assert("should get some tokens back", tokens != NULL, "tokens was %p", tokens);
+ MQTTAsync_free(tokens);
+
+ assert("test8_publishFailures == 0", test8_publishFailures == 0,
+ "test8_publishFailures = %d", test8_publishFailures);
+
+ MQTTAsync_destroy(&c);
+
+ assert("test8_publishFailures > 0", test8_publishFailures > 0,
+ "test8_publishFailures = %d", test8_publishFailures);
+
+exit:
+ MyLog(LOGA_INFO, "TEST8: test %s. %d tests run, %d failures.",
+ (failures == 0) ? "passed" : "failed", tests, failures);
+ write_test_result();
+ return failures;
+}
+
+
+
+void trace_callback(enum MQTTASYNC_TRACE_LEVELS level, char* message)
+{
+ printf("Trace : %d, %s\n", level, message);
+}
+
+
+
+
+int main(int argc, char** argv)
+{
+ int rc = 0;
+ int (*tests[])() = {NULL, test1, test2, test3, test4, test5, test6, test7, test8}; /* indexed starting from 1 */
+ MQTTAsync_nameValue* info;
+ int i;
+
+ xml = fopen("TEST-test4.xml", "w");
+ fprintf(xml, "\n", (int)(ARRAY_SIZE(tests)) - 1);
+
+ getopts(argc, argv);
+
+ MQTTAsync_setTraceCallback(trace_callback);
+
+ info = MQTTAsync_getVersionInfo();
+ while (info->name)
+ {
+ MyLog(LOGA_INFO, "%s: %s", info->name, info->value);
+ info++;
+ }
+
+ for (i = 0; i < options.iterations; ++i)
+ {
+ if (options.test_no == -1)
+ { /* run all the tests */
+ for (options.test_no = 1; options.test_no < ARRAY_SIZE(tests); ++options.test_no)
+ {
+ failures = 0;
+ MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
+ rc += tests[options.test_no](options); /* return number of failures. 0 = test succeeded */
+ }
+ }
+ else
+ {
+ MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
+ //MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_PROTOCOL);
+ rc = tests[options.test_no](options); /* run just the selected test */
+ }
+ }
+
+ if (rc == 0)
+ MyLog(LOGA_INFO, "verdict pass");
+ else
+ MyLog(LOGA_INFO, "verdict fail");
+
+ fprintf(xml, "\n");
+ fclose(xml);
+
+ return rc;
+}