natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MQTT V5 persistence updates #534

This commit is contained in:
Ian Craggs 2018-09-05 22:13:40 +01:00
parent e24c1b4ce6
commit 296fe57cd2
22 changed files with 2967 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -107,7 +107,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 test45 test6 test9 test_mqtt4async test11
TEST_FILES_A = test4 test45 test6 test9 test95 test_mqtt4async test11
ASYNC_TESTS = ${addprefix ${blddir}/test/,${TEST_FILES_A}}
TEST_FILES_AS = test5

107
src/MQTTAsync.c
View File

@ -379,7 +379,7 @@ static void MQTTAsync_terminate(void);
#if !defined(NO_PERSISTENCE)
static int MQTTAsync_unpersistCommand(MQTTAsync_queuedCommand* qcmd);
static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd);
static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen);
static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen, int MQTTVersion);
/*static void MQTTAsync_insertInOrder(List* list, void* content, int size);*/
static int MQTTAsync_restoreCommands(MQTTAsyncs* client);
#endif
@ -571,6 +571,7 @@ int MQTTAsync_createWithOptions(MQTTAsync* handle, const char* serverURI, const
{
m->createOptions = malloc(sizeof(MQTTAsync_createOptions));
memcpy(m->createOptions, options, sizeof(MQTTAsync_createOptions));
m->c->MQTTVersion = options->MQTTVersion;
}
#if !defined(NO_PERSISTENCE)
@ -650,12 +651,15 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd)
void** bufs = NULL;
int bufindex = 0, i, nbufs = 0;
char key[PERSISTENCE_MAX_KEY_LENGTH + 1];
int props_allocated = 0;
int process = 1;
FUNC_ENTRY;
switch (command->type)
{
case SUBSCRIBE:
nbufs = 3 + (command->details.sub.count * 2);
nbufs = ((aclient->c->MQTTVersion >= MQTTVERSION_5) ? 4 : 3) +
(command->details.sub.count * 2);
lens = (int*)malloc(nbufs * sizeof(int));
bufs = malloc(nbufs * sizeof(char *));
@ -673,14 +677,31 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd)
{
bufs[bufindex] = command->details.sub.topics[i];
lens[bufindex++] = (int)strlen(command->details.sub.topics[i]) + 1;
bufs[bufindex] = &command->details.sub.qoss[i];
lens[bufindex++] = sizeof(command->details.sub.qoss[i]);
if (aclient->c->MQTTVersion < MQTTVERSION_5)
{
bufs[bufindex] = &command->details.sub.qoss[i];
lens[bufindex++] = sizeof(command->details.sub.qoss[i]);
}
else
{
if (command->details.sub.count == 1)
{
bufs[bufindex] = &command->details.sub.opts;
lens[bufindex++] = sizeof(command->details.sub.opts);
}
else
{
bufs[bufindex] = &command->details.sub.optlist[i];
lens[bufindex++] = sizeof(command->details.sub.optlist[i]);
}
}
}
sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno);
break;
case UNSUBSCRIBE:
nbufs = 3 + command->details.unsub.count;
nbufs = ((aclient->c->MQTTVersion >= MQTTVERSION_5) ? 4 : 3) +
command->details.unsub.count;
lens = (int*)malloc(nbufs * sizeof(int));
bufs = malloc(nbufs * sizeof(char *));
@ -699,11 +720,10 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd)
bufs[bufindex] = command->details.unsub.topics[i];
lens[bufindex++] = (int)strlen(command->details.unsub.topics[i]) + 1;
}
sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno);
break;
case PUBLISH:
nbufs = 7;
nbufs = (aclient->c->MQTTVersion >= MQTTVERSION_5) ? 8 : 7;
lens = (int*)malloc(nbufs * sizeof(int));
bufs = malloc(nbufs * sizeof(char *));
@ -728,10 +748,27 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd)
bufs[bufindex] = &command->details.pub.retained;
lens[bufindex++] = sizeof(command->details.pub.retained);
break;
sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno);
default:
process = 0;
break;
}
if (aclient->c->MQTTVersion >= MQTTVERSION_5 && process) /* persist properties */
{
int temp_len = 0;
char* ptr = NULL;
temp_len = MQTTProperties_len(&command->properties);
ptr = bufs[bufindex] = malloc(temp_len);
props_allocated = bufindex;
rc = MQTTProperties_write(&ptr, &command->properties);
lens[bufindex++] = temp_len;
sprintf(key, "%s%u", PERSISTENCE_V5_COMMAND_KEY, ++aclient->command_seqno);
}
else
sprintf(key, "%s%u", PERSISTENCE_COMMAND_KEY, ++aclient->command_seqno);
if (nbufs > 0)
{
if ((rc = aclient->c->persistence->pput(aclient->c->phandle, key, nbufs, (char**)bufs, lens)) != 0)
@ -742,12 +779,14 @@ static int MQTTAsync_persistCommand(MQTTAsync_queuedCommand* qcmd)
free(lens);
if (bufs)
free(bufs);
if (props_allocated > 0)
free(bufs[props_allocated]);
FUNC_EXIT_RC(rc);
return rc;
}
static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen)
static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int buflen, int MQTTVersion)
{
MQTTAsync_command* command = NULL;
MQTTAsync_queuedCommand* qcommand = NULL;
@ -774,8 +813,11 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle
if (command->details.sub.count > 0)
{
command->details.sub.topics = (char **)malloc(sizeof(char *) * command->details.sub.count);
command->details.sub.topics = (char **)malloc(sizeof(char *) * command->details.sub.count);
if (MQTTVersion < MQTTVERSION_5)
command->details.sub.qoss = (int *)malloc(sizeof(int) * command->details.sub.count);
else if (command->details.sub.count > 1)
command->details.sub.optlist = (MQTTSubscribe_options*)malloc(sizeof(MQTTSubscribe_options) * command->details.sub.count);
}
for (i = 0; i < command->details.sub.count; ++i)
@ -786,8 +828,24 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle
strcpy(command->details.sub.topics[i], ptr);
ptr += data_size;
command->details.sub.qoss[i] = *(int*)ptr;
ptr += sizeof(int);
if (MQTTVersion < MQTTVERSION_5)
{
command->details.sub.qoss[i] = *(int*)ptr;
ptr += sizeof(int);
}
else
{
if (command->details.sub.count == 1)
{
command->details.sub.opts = *(MQTTSubscribe_options*)ptr;
ptr += sizeof(MQTTSubscribe_options);
}
else
{
command->details.sub.optlist[i] = *(MQTTSubscribe_options*)ptr;
ptr += sizeof(MQTTSubscribe_options);
}
}
}
break;
@ -797,7 +855,7 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle
if (command->details.unsub.count > 0)
{
command->details.unsub.topics = (char **)malloc(sizeof(char *) * command->details.unsub.count);
command->details.unsub.topics = (char **)malloc(sizeof(char *) * command->details.unsub.count);
}
for (i = 0; i < command->details.unsub.count; ++i)
@ -836,6 +894,13 @@ static MQTTAsync_queuedCommand* MQTTAsync_restoreCommand(char* buffer, int bufle
qcommand = NULL;
}
if (qcommand != NULL && MQTTVersion >= MQTTVERSION_5 &&
MQTTProperties_read(&command->properties, &ptr, buffer + buflen) != 1)
{
Log(LOG_ERROR, -1, "Error restoring properties from persistence");
free(qcommand);
qcommand = NULL;
}
FUNC_EXIT;
return qcommand;
@ -876,18 +941,22 @@ static int MQTTAsync_restoreCommands(MQTTAsyncs* client)
char *buffer = NULL;
int buflen;
if (strncmp(msgkeys[i], PERSISTENCE_COMMAND_KEY, strlen(PERSISTENCE_COMMAND_KEY)) != 0)
if (strncmp(msgkeys[i], PERSISTENCE_COMMAND_KEY, strlen(PERSISTENCE_COMMAND_KEY)) != 0 &&
strncmp(msgkeys[i], PERSISTENCE_V5_COMMAND_KEY, strlen(PERSISTENCE_V5_COMMAND_KEY)) != 0)
{
;
}
else if ((rc = c->persistence->pget(c->phandle, msgkeys[i], &buffer, &buflen)) == 0)
{
MQTTAsync_queuedCommand* cmd = MQTTAsync_restoreCommand(buffer, buflen);
int MQTTVersion =
(strncmp(msgkeys[i], PERSISTENCE_V5_COMMAND_KEY, strlen(PERSISTENCE_V5_COMMAND_KEY)) == 0)
? MQTTVERSION_5 : MQTTVERSION_3_1_1;
MQTTAsync_queuedCommand* cmd = MQTTAsync_restoreCommand(buffer, buflen, MQTTVersion);
if (cmd)
{
cmd->client = client;
cmd->seqno = atoi(msgkeys[i]+2);
cmd->seqno = atoi(strchr(msgkeys[i], '-')+1); /* key format is tag'-'seqno */
MQTTPersistence_insertInOrder(commands, cmd, sizeof(MQTTAsync_queuedCommand));
free(buffer);
client->command_seqno = max(client->command_seqno, cmd->seqno);
@ -1395,7 +1464,7 @@ static int MQTTAsync_processCommand(void)
rc = MQTTProtocol_subscribe(command->client->c, topics, qoss, command->command.token, subopts, props);
ListFreeNoContent(topics);
ListFreeNoContent(qoss);
if (command->command.details.sub.count > 1)
if (command->client->c->MQTTVersion >= MQTTVERSION_5 && command->command.details.sub.count > 1)
free(command->command.details.sub.optlist);
}
else if (command->command.type == UNSUBSCRIBE)
@ -3393,7 +3462,7 @@ int MQTTAsync_sendMessage(MQTTAsync handle, const char* destinationName, const M
goto exit;
}
if (m->c->MQTTVersion >= MQTTVERSION_5)
if (m->c->MQTTVersion >= MQTTVERSION_5 && response)
response->properties = message->properties;
rc = MQTTAsync_send(handle, destinationName, message->payloadlen, message->payload,

8
src/MQTTAsync.h
View File

@ -816,15 +816,19 @@ typedef struct
{
/** The eyecatcher for this structure. must be MQCO. */
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 means no MQTTVersion
*/
int struct_version;
/** Whether to allow messages to be sent when the client library is not connected. */
int sendWhileDisconnected;
/** the maximum number of messages allowed to be buffered while not connected. */
int maxBufferedMessages;
/** Whether the MQTT version is 3 and 4, or 5. To use 5, this must be set. */
int MQTTVersion;
} MQTTAsync_createOptions;
#define MQTTAsync_createOptions_initializer { {'M', 'Q', 'C', 'O'}, 0, 0, 100 }
#define MQTTAsync_createOptions_initializer { {'M', 'Q', 'C', 'O'}, 0, 0, 100, MQTTVERSION_DEFAULT }
DLLExport int MQTTAsync_createWithOptions(MQTTAsync* handle, const char* serverURI, const char* clientId,

26
src/MQTTClient.c
View File

@ -1493,10 +1493,17 @@ static MQTTResponse MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectO
return rc;
}
MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions* options,
MQTTProperties* connectProperties, MQTTProperties* willProperties);
int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
{
MQTTResponse response = MQTTClient_connect5(handle, options, NULL, NULL);
MQTTClients* m = handle;
if (m->c->MQTTVersion >= MQTTVERSION_5)
return MQTTCLIENT_WRONG_MQTT_VERSION;
MQTTResponse response = MQTTClient_connectAll(handle, options, NULL, NULL);
return response.reasonCode;
}
@ -1504,6 +1511,22 @@ int MQTTClient_connect(MQTTClient handle, MQTTClient_connectOptions* options)
MQTTResponse MQTTClient_connect5(MQTTClient handle, MQTTClient_connectOptions* options,
MQTTProperties* connectProperties, MQTTProperties* willProperties)
{
MQTTClients* m = handle;
MQTTResponse response = MQTTResponse_initializer;
if (m->c->MQTTVersion < MQTTVERSION_5)
{
response.reasonCode = MQTTCLIENT_WRONG_MQTT_VERSION;
return response;
}
return MQTTClient_connectAll(handle, options, connectProperties, willProperties);
}
MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions* options,
MQTTProperties* connectProperties, MQTTProperties* willProperties)
{
MQTTClients* m = handle;
MQTTResponse rc = MQTTResponse_initializer;
@ -2130,7 +2153,6 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p
{
while (m->c->connected == 1 && SocketBuffer_getWrite(m->c->net.socket))
{
printf("getwrite %p\n", SocketBuffer_getWrite(m->c->net.socket));
Thread_unlock_mutex(mqttclient_mutex);
MQTTClient_yield();
Thread_lock_mutex(mqttclient_mutex);

4
src/MQTTClient.h
View File

@ -191,6 +191,10 @@
* Return code: option not applicable to the requested version of MQTT
*/
#define MQTTCLIENT_BAD_MQTT_OPTION -15
/**
* Return code: call not applicable to the requested version of MQTT
*/
#define MQTTCLIENT_WRONG_MQTT_VERSION -16
/**

26
src/MQTTPacket.c
View File

@ -152,7 +152,7 @@ void* MQTTPacket_Factory(int MQTTVersion, networkHandles* net, int* error)
buf[0] = header.byte;
buf0len = 1 + MQTTPacket_encode(&buf[1], remaining_length);
*error = MQTTPersistence_put(net->socket, buf, buf0len, 1,
&data, &remaining_length, header.bits.type, ((Publish *)pack)->msgId, 1);
&data, &remaining_length, header.bits.type, ((Publish *)pack)->msgId, 1, MQTTVersion);
free(buf);
}
#endif
@ -172,9 +172,11 @@ exit:
* @param header the one-byte MQTT header
* @param buffer the rest of the buffer to write (not including remaining length)
* @param buflen the length of the data in buffer to be written
* @param MQTTVersion the version of MQTT being used
* @return the completion code (TCPSOCKET_COMPLETE etc)
*/
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int freeData)
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int freeData,
int MQTTVersion)
{
int rc;
size_t buf0len;
@ -200,7 +202,7 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buf
int msgId = readInt(&ptraux);
rc = MQTTPersistence_put(net->socket, &buf[ws_header], buf0len, count, &buffer, &buflen,
header.bits.type, msgId, 0);
header.bits.type, msgId, 0, MQTTVersion);
}
#endif
rc = WebSocket_putdatas(net, &buf[ws_header], buf0len, count, &buffer, &buflen, &freeData);
@ -223,9 +225,11 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buf
* @param count the number of buffers
* @param buffers the rest of the buffers to write (not including remaining length)
* @param buflens the lengths of the data in the array of buffers to be written
* @param the MQTT version being used
* @return the completion code (TCPSOCKET_COMPLETE etc)
*/
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees)
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens,
int* frees, int MQTTVersion)
{
int i, rc;
size_t buf0len, total = 0;
@ -249,7 +253,7 @@ int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffe
char *ptraux = buffers[2];
int msgId = readInt(&ptraux);
rc = MQTTPersistence_put(net->socket, &buf[ws_header], buf0len, count, buffers, buflens,
header.bits.type, msgId, 0);
header.bits.type, msgId, 0, MQTTVersion);
}
#endif
rc = WebSocket_putdatas(net, &buf[ws_header], buf0len, count, buffers, buflens, frees);
@ -496,12 +500,13 @@ int MQTTPacket_send_disconnect(Clients* client, enum MQTTReasonCodes reason, MQT
writeChar(&ptr, reason);
if (props)
MQTTProperties_write(&ptr, props);
if ((rc = MQTTPacket_send(&client->net, header, buf, buflen, 1)) != TCPSOCKET_INTERRUPTED)
if ((rc = MQTTPacket_send(&client->net, header, buf, buflen, 1,
client->MQTTVersion)) != TCPSOCKET_INTERRUPTED)
free(buf);
}
}
else
rc = MQTTPacket_send(&client->net, header, NULL, 0, 0);
rc = MQTTPacket_send(&client->net, header, NULL, 0, 0, client->MQTTVersion);
Log(LOG_PROTOCOL, 28, NULL, client->net.socket, client->clientID, rc);
FUNC_EXIT_RC(rc);
return rc;
@ -523,6 +528,7 @@ void* MQTTPacket_publish(int MQTTVersion, unsigned char aHeader, char* data, siz
char* enddata = &data[datalen];
FUNC_ENTRY;
memset(pack, '\0', sizeof(Publish));
pack->MQTTVersion = MQTTVersion;
pack->header.byte = aHeader;
if ((pack->topic = readUTFlen(&curdata, enddata, &pack->topiclen)) == NULL) /* Topic name on which to publish */
@ -606,7 +612,7 @@ static int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net
if (type == PUBREL)
header.bits.qos = 1;
writeInt(&ptr, msgid);
if ((rc = MQTTPacket_send(net, header, buf, 2, 1)) != TCPSOCKET_INTERRUPTED)
if ((rc = MQTTPacket_send(net, header, buf, 2, 1, MQTTVERSION_3_1_1)) != TCPSOCKET_INTERRUPTED)
free(buf);
FUNC_EXIT_RC(rc);
return rc;
@ -805,7 +811,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo
ptr = topiclen;
writeInt(&ptr, (int)lens[1]);
rc = MQTTPacket_sends(net, header, 4, bufs, lens, frees);
rc = MQTTPacket_sends(net, header, 4, bufs, lens, frees, pack->MQTTVersion);
if (rc != TCPSOCKET_INTERRUPTED)
free(bufs[2]);
}
@ -817,7 +823,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo
int frees[3] = {1, 0, 0};
writeInt(&ptr, (int)lens[1]);
rc = MQTTPacket_sends(net, header, 3, bufs, lens, frees);
rc = MQTTPacket_sends(net, header, 3, bufs, lens, frees, pack->MQTTVersion);
}
if (rc != TCPSOCKET_INTERRUPTED)
free(topiclen);

4
src/MQTTPacket.h
View File

@ -237,8 +237,8 @@ void writeData(char** pptr, const void* data, int datalen);
const char* MQTTPacket_name(int ptype);
void* MQTTPacket_Factory(int MQTTVersion, networkHandles* net, int* error);
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int free);
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees);
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int free, int MQTTVersion);
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees, int MQTTVersion);
void* MQTTPacket_header_only(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen);
int MQTTPacket_send_disconnect(Clients* client, enum MQTTReasonCodes reason, MQTTProperties* props);

8
src/MQTTPacketOut.c
View File

@ -117,7 +117,7 @@ int MQTTPacket_send_connect(Clients* client, int MQTTVersion,
if (client->password)
writeData(&ptr, client->password, client->passwordlen);
rc = MQTTPacket_send(&client->net, packet.header, buf, len, 1);
rc = MQTTPacket_send(&client->net, packet.header, buf, len, 1, MQTTVersion);
Log(LOG_PROTOCOL, 0, NULL, client->net.socket, client->clientID, client->cleansession, rc);
exit:
if (rc != TCPSOCKET_INTERRUPTED)
@ -197,7 +197,7 @@ int MQTTPacket_send_pingreq(networkHandles* net, const char* clientID)
FUNC_ENTRY;
header.byte = 0;
header.bits.type = PINGREQ;
rc = MQTTPacket_send(net, header, NULL, 0, 0);
rc = MQTTPacket_send(net, header, NULL, 0, 0, MQTTVERSION_3_1_1);
Log(LOG_PROTOCOL, 20, NULL, net->socket, clientID, rc);
FUNC_EXIT_RC(rc);
return rc;
@ -258,7 +258,7 @@ int MQTTPacket_send_subscribe(List* topics, List* qoss, MQTTSubscribe_options* o
writeChar(&ptr, subopts);
++i;
}
rc = MQTTPacket_send(&client->net, header, data, datalen, 1);
rc = MQTTPacket_send(&client->net, header, data, datalen, 1, client->MQTTVersion);
Log(LOG_PROTOCOL, 22, NULL, client->net.socket, client->clientID, msgid, rc);
if (rc != TCPSOCKET_INTERRUPTED)
free(data);
@ -347,7 +347,7 @@ int MQTTPacket_send_unsubscribe(List* topics, MQTTProperties* props, int msgid,
elem = NULL;
while (ListNextElement(topics, &elem))
writeUTF(&ptr, (char*)(elem->content));
rc = MQTTPacket_send(&client->net, header, data, datalen, 1);
rc = MQTTPacket_send(&client->net, header, data, datalen, 1, client->MQTTVersion);
Log(LOG_PROTOCOL, 25, NULL, client->net.socket, client->clientID, msgid, rc);
if (rc != TCPSOCKET_INTERRUPTED)
free(data);

176
src/MQTTPersistence.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2013 IBM Corp.
* 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
@ -14,6 +14,7 @@
* Ian Craggs - initial API and implementation and/or initial documentation
* Ian Craggs - async client updates
* Ian Craggs - fix for bug 432903 - queue persistence
* Ian Craggs - MQTT V5 updates
*******************************************************************************/
/**
@ -31,7 +32,7 @@
#include "Heap.h"
static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen);
static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen, int MQTTVersion);
static void MQTTPersistence_insertInSeqOrder(List* list, MQTTPersistence_qEntry* qEntry, size_t size);
/**
@ -196,13 +197,43 @@ int MQTTPersistence_restore(Clients *c)
}
else if ((rc = c->persistence->pget(c->phandle, msgkeys[i], &buffer, &buflen)) == 0)
{
MQTTPacket* pack = MQTTPersistence_restorePacket(c->MQTTVersion, buffer, buflen);
int data_MQTTVersion = MQTTVERSION_3_1_1;
char* cur_key = msgkeys[i];
if ( strncmp(cur_key, PERSISTENCE_V5_PUBLISH_RECEIVED,
strlen(PERSISTENCE_V5_PUBLISH_RECEIVED)) == 0)
{
data_MQTTVersion = MQTTVERSION_5;
cur_key = PERSISTENCE_PUBLISH_RECEIVED;
}
else if (strncmp(cur_key, PERSISTENCE_V5_PUBLISH_SENT,
strlen(PERSISTENCE_V5_PUBLISH_SENT)) == 0)
{
data_MQTTVersion = MQTTVERSION_5;
cur_key = PERSISTENCE_PUBLISH_SENT;
}
else if (strncmp(cur_key, PERSISTENCE_V5_PUBREL,
strlen(PERSISTENCE_V5_PUBREL)) == 0)
{
data_MQTTVersion = MQTTVERSION_5;
cur_key = PERSISTENCE_PUBREL;
}
if (data_MQTTVersion == MQTTVERSION_5 && c->MQTTVersion < MQTTVERSION_5)
{
rc = MQTTCLIENT_PERSISTENCE_ERROR; /* can't restore version 5 data with a version 3 client */
goto exit;
}
MQTTPacket* pack = MQTTPersistence_restorePacket(data_MQTTVersion, buffer, buflen);
if ( pack != NULL )
{
if ( strstr(msgkeys[i],PERSISTENCE_PUBLISH_RECEIVED) != NULL )
if (strncmp(cur_key, PERSISTENCE_PUBLISH_RECEIVED,
strlen(PERSISTENCE_PUBLISH_RECEIVED)) == 0)
{
Publish* publish = (Publish*)pack;
Messages* msg = NULL;
publish->MQTTVersion = c->MQTTVersion;
msg = MQTTProtocol_createMessage(publish, &msg, publish->header.bits.qos, publish->header.bits.retain);
msg->nextMessageType = PUBREL;
/* order does not matter for persisted received messages */
@ -211,14 +242,20 @@ int MQTTPersistence_restore(Clients *c)
MQTTPacket_freePublish(publish);
msgs_rcvd++;
}
else if ( strstr(msgkeys[i],PERSISTENCE_PUBLISH_SENT) != NULL )
else if (strncmp(cur_key, PERSISTENCE_PUBLISH_SENT,
strlen(PERSISTENCE_PUBLISH_SENT)) == 0)
{
Publish* publish = (Publish*)pack;
Messages* msg = NULL;
char *key = malloc(MESSAGE_FILENAME_LENGTH + 1);
sprintf(key, "%s%d", PERSISTENCE_PUBREL, publish->msgId);
publish->MQTTVersion = c->MQTTVersion;
if (publish->MQTTVersion >= MQTTVERSION_5)
sprintf(key, "%s%d", PERSISTENCE_V5_PUBREL, publish->msgId);
else
sprintf(key, "%s%d", PERSISTENCE_PUBREL, publish->msgId);
msg = MQTTProtocol_createMessage(publish, &msg, publish->header.bits.qos, publish->header.bits.retain);
if ( c->persistence->pcontainskey(c->phandle, key) == 0 )
if (c->persistence->pcontainskey(c->phandle, key) == 0)
/* PUBLISH Qo2 and PUBREL sent */
msg->nextMessageType = PUBCOMP;
/* else: PUBLISH QoS1, or PUBLISH QoS2 and PUBREL not sent */
@ -230,13 +267,18 @@ int MQTTPersistence_restore(Clients *c)
free(key);
msgs_sent++;
}
else if ( strstr(msgkeys[i],PERSISTENCE_PUBREL) != NULL )
else if (strncmp(cur_key, PERSISTENCE_PUBREL, strlen(PERSISTENCE_PUBREL)) == 0)
{
/* orphaned PUBRELs ? */
Pubrel* pubrel = (Pubrel*)pack;
char *key = malloc(MESSAGE_FILENAME_LENGTH + 1);
sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, pubrel->msgId);
if ( c->persistence->pcontainskey(c->phandle, key) != 0 )
pubrel->MQTTVersion = c->MQTTVersion;
if (pubrel->MQTTVersion >= MQTTVERSION_5)
sprintf(key, "%s%d", PERSISTENCE_V5_PUBLISH_SENT, pubrel->msgId);
else
sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, pubrel->msgId);
if (c->persistence->pcontainskey(c->phandle, key) != 0)
rc = c->persistence->premove(c->phandle, msgkeys[i]);
free(pubrel);
free(key);
@ -260,7 +302,7 @@ int MQTTPersistence_restore(Clients *c)
Log(TRACE_MINIMUM, -1, "%d sent messages and %d received messages restored for client %s\n",
msgs_sent, msgs_rcvd, c->clientID);
MQTTPersistence_wrapMsgID(c);
exit:
FUNC_EXIT_RC(rc);
return rc;
}
@ -335,13 +377,15 @@ void MQTTPersistence_insertInOrder(List* list, void* content, size_t size)
* @param count number of buffers representing the variable header and/or the payload.
* @param buffers the buffers representing the variable header and/or the payload.
* @param buflens length of the buffers representing the variable header and/or the payload.
* @param htype MQTT packet type - PUBLISH or PUBREL
* @param msgId the message ID.
* @param scr 0 indicates message in the sending direction; 1 indicates message in the
* receiving direction.
* @param the MQTT version being used (>= MQTTVERSION_5 means properties included)
* @return 0 if success, #MQTTCLIENT_PERSISTENCE_ERROR otherwise.
*/
int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count,
char** buffers, size_t* buflens, int htype, int msgId, int scr )
char** buffers, size_t* buflens, int htype, int msgId, int scr, int MQTTVersion)
{
int rc = 0;
extern ClientStates* bstate;
@ -368,15 +412,32 @@ int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count,
}
/* key */
if ( scr == 0 )
if (scr == 0)
{ /* sending */
if (htype == PUBLISH) /* PUBLISH QoS1 and QoS2*/
sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, msgId);
if (htype == PUBREL) /* PUBREL */
sprintf(key, "%s%d", PERSISTENCE_PUBREL, msgId);
char* key_id = PERSISTENCE_PUBLISH_SENT;
if (htype == PUBLISH) /* PUBLISH QoS1 and QoS2*/
{
if (MQTTVersion >= MQTTVERSION_5)
key_id = PERSISTENCE_V5_PUBLISH_SENT;
}
else if (htype == PUBREL) /* PUBREL */
{
if (MQTTVersion >= MQTTVERSION_5)
key_id = PERSISTENCE_V5_PUBREL;
else
key_id = PERSISTENCE_PUBREL;
}
sprintf(key, "%s%d", key_id, msgId);
}
else if (scr == 1) /* receiving PUBLISH QoS2 */
{
char* key_id = PERSISTENCE_PUBLISH_RECEIVED;
if (MQTTVersion >= MQTTVERSION_5)
key_id = PERSISTENCE_V5_PUBLISH_RECEIVED;
sprintf(key, "%s%d", key_id, msgId);
}
if ( scr == 1 ) /* receiving PUBLISH QoS2 */
sprintf(key, "%s%d", PERSISTENCE_PUBLISH_RECEIVED, msgId);
rc = client->persistence->pput(client->phandle, key, nbufs, bufs, lens);
@ -407,17 +468,24 @@ int MQTTPersistence_remove(Clients* c, char *type, int qos, int msgId)
if (c->persistence != NULL)
{
char *key = malloc(MESSAGE_FILENAME_LENGTH + 1);
if ( (strcmp(type,PERSISTENCE_PUBLISH_SENT) == 0) && qos == 2 )
if (strcmp(type, PERSISTENCE_PUBLISH_SENT) == 0) //&& qos == 2 )
{
sprintf(key, "%s%d", PERSISTENCE_V5_PUBLISH_SENT, msgId) ;
rc = c->persistence->premove(c->phandle, key);
sprintf(key, "%s%d", PERSISTENCE_V5_PUBREL, msgId) ;
rc += c->persistence->premove(c->phandle, key);
sprintf(key, "%s%d", PERSISTENCE_PUBLISH_SENT, msgId) ;
rc = c->persistence->premove(c->phandle, key);
rc += c->persistence->premove(c->phandle, key);
sprintf(key, "%s%d", PERSISTENCE_PUBREL, msgId) ;
rc = c->persistence->premove(c->phandle, key);
rc += c->persistence->premove(c->phandle, key);
}
else /* PERSISTENCE_PUBLISH_SENT && qos == 1 */
{ /* or PERSISTENCE_PUBLISH_RECEIVED */
sprintf(key, "%s%d", type, msgId) ;
sprintf(key, "%s%d", PERSISTENCE_V5_PUBLISH_RECEIVED, msgId);
rc = c->persistence->premove(c->phandle, key);
sprintf(key, "%s%d", PERSISTENCE_PUBLISH_RECEIVED, msgId);
rc += c->persistence->premove(c->phandle, key);
}
free(key);
}
@ -489,17 +557,15 @@ int MQTTPersistence_unpersistQueueEntry(Clients* client, MQTTPersistence_qEntry*
int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry* qe)
{
const int MAX_NO_OF_BUFFERS = 9;
int rc = 0;
int nbufs = 8;
int bufindex = 0;
char key[PERSISTENCE_MAX_KEY_LENGTH + 1];
int* lens = NULL;
void** bufs = NULL;
int lens[MAX_NO_OF_BUFFERS];
void* bufs[MAX_NO_OF_BUFFERS];
int props_allocated = 0;
FUNC_ENTRY;
lens = (int*)malloc(nbufs * sizeof(int));
bufs = malloc(nbufs * sizeof(char *));
bufs[bufindex] = &qe->msg->payloadlen;
lens[bufindex++] = sizeof(qe->msg->payloadlen);
@ -522,23 +588,43 @@ int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry*
lens[bufindex++] = (int)strlen(qe->topicName) + 1;
bufs[bufindex] = &qe->topicLen;
lens[bufindex++] = sizeof(qe->topicLen);
sprintf(key, "%s%u", PERSISTENCE_QUEUE_KEY, ++aclient->qentry_seqno);
lens[bufindex++] = sizeof(qe->topicLen);
if (aclient->MQTTVersion >= MQTTVERSION_5) /* persist properties */
{
MQTTProperties no_props = MQTTProperties_initializer;
MQTTProperties* props = &no_props;
int temp_len = 0;
char* ptr = NULL;
if (qe->msg->struct_version >= 1)
props = &qe->msg->properties;
temp_len = MQTTProperties_len(props);
ptr = bufs[bufindex] = malloc(temp_len);
props_allocated = bufindex;
rc = MQTTProperties_write(&ptr, props);
lens[bufindex++] = temp_len;
sprintf(key, "%s%u", PERSISTENCE_V5_QUEUE_KEY, ++aclient->qentry_seqno);
}
else
sprintf(key, "%s%u", PERSISTENCE_QUEUE_KEY, ++aclient->qentry_seqno);
qe->seqno = aclient->qentry_seqno;
if ((rc = aclient->persistence->pput(aclient->phandle, key, nbufs, (char**)bufs, lens)) != 0)
if ((rc = aclient->persistence->pput(aclient->phandle, key, bufindex, (char**)bufs, lens)) != 0)
Log(LOG_ERROR, 0, "Error persisting queue entry, rc %d", rc);
free(lens);
free(bufs);
if (props_allocated != 0)
free(bufs[props_allocated]);
FUNC_EXIT_RC(rc);
return rc;
}
static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen)
static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen, int MQTTVersion)
{
MQTTPersistence_qEntry* qe = NULL;
char* ptr = buffer;
@ -551,6 +637,8 @@ static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, s
qe->msg = malloc(sizeof(MQTTPersistence_message));
memset(qe->msg, '\0', sizeof(MQTTPersistence_message));
qe->msg->struct_version = 1;
qe->msg->payloadlen = *(int*)ptr;
ptr += sizeof(int);
@ -579,6 +667,10 @@ static MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, s
qe->topicLen = *(int*)ptr;
ptr += sizeof(int);
if (MQTTVersion >= MQTTVERSION_5 &&
MQTTProperties_read(&qe->msg->properties, &ptr, buffer + buflen) != 1)
Log(LOG_ERROR, -1, "Error restoring properties from persistence");
FUNC_EXIT;
return qe;
}
@ -621,17 +713,21 @@ int MQTTPersistence_restoreMessageQueue(Clients* c)
char *buffer = NULL;
int buflen;
if (strncmp(msgkeys[i], PERSISTENCE_QUEUE_KEY, strlen(PERSISTENCE_QUEUE_KEY)) != 0)
if (strncmp(msgkeys[i], PERSISTENCE_QUEUE_KEY, strlen(PERSISTENCE_QUEUE_KEY)) != 0 &&
strncmp(msgkeys[i], PERSISTENCE_V5_QUEUE_KEY, strlen(PERSISTENCE_V5_QUEUE_KEY)) != 0)
{
;
; /* ignore if not a queue entry key */
}
else if ((rc = c->persistence->pget(c->phandle, msgkeys[i], &buffer, &buflen)) == 0)
{
MQTTPersistence_qEntry* qe = MQTTPersistence_restoreQueueEntry(buffer, buflen);
int MQTTVersion =
(strncmp(msgkeys[i], PERSISTENCE_V5_QUEUE_KEY, strlen(PERSISTENCE_V5_QUEUE_KEY)) == 0)
? MQTTVERSION_5 : MQTTVERSION_3_1_1;
MQTTPersistence_qEntry* qe = MQTTPersistence_restoreQueueEntry(buffer, buflen, MQTTVersion);
if (qe)
{
qe->seqno = atoi(msgkeys[i]+2);
qe->seqno = atoi(strchr(msgkeys[i], '-')+1); /* key format is tag'-'seqno */
MQTTPersistence_insertInSeqOrder(c->messageQueue, qe, sizeof(MQTTPersistence_qEntry));
free(buffer);
c->qentry_seqno = max(c->qentry_seqno, qe->seqno);

17
src/MQTTPersistence.h
View File

@ -14,6 +14,7 @@
* Ian Craggs - initial API and implementation and/or initial documentation
* Ian Craggs - async client updates
* Ian Craggs - fix for bug 432903 - queue persistence
* Ian Craggs - MQTT V5 updates
*******************************************************************************/
#if !defined(MQTTPERSISTENCE_H)
@ -24,6 +25,7 @@
#endif
#include "Clients.h"
#include "MQTTProperties.h"
/** Stem of the key for a sent PUBLISH QoS1 or QoS2 */
#define PERSISTENCE_PUBLISH_SENT "s-"
@ -31,10 +33,22 @@
#define PERSISTENCE_PUBREL "sc-"
/** Stem of the key for a received PUBLISH QoS2 */
#define PERSISTENCE_PUBLISH_RECEIVED "r-"
/** Stem of the key for a sent MQTT V5 PUBLISH QoS1 or QoS2 */
#define PERSISTENCE_V5_PUBLISH_SENT "s5-"
/** Stem of the key for a sent MQTT V5 PUBREL */
#define PERSISTENCE_V5_PUBREL "sc5-"
/** Stem of the key for a received MQTT V5 PUBLISH QoS2 */
#define PERSISTENCE_V5_PUBLISH_RECEIVED "r5-"
/** Stem of the key for an async client command */
#define PERSISTENCE_COMMAND_KEY "c-"
/** Stem of the key for an MQTT V5 async client command */
#define PERSISTENCE_V5_COMMAND_KEY "c-"
/** Stem of the key for an async client message queue */
#define PERSISTENCE_QUEUE_KEY "q-"
/** Stem of the key for an MQTT V5 message queue */
#define PERSISTENCE_V5_QUEUE_KEY "q5-"
#define PERSISTENCE_MAX_KEY_LENGTH 8
int MQTTPersistence_create(MQTTClient_persistence** per, int type, void* pcontext);
@ -45,7 +59,7 @@ int MQTTPersistence_restore(Clients* c);
void* MQTTPersistence_restorePacket(int MQTTVersion, char* buffer, size_t buflen);
void MQTTPersistence_insertInOrder(List* list, void* content, size_t size);
int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count,
char** buffers, size_t* buflens, int htype, int msgId, int scr);
char** buffers, size_t* buflens, int htype, int msgId, int scr, int MQTTVersion);
int MQTTPersistence_remove(Clients* c, char* type, int qos, int msgId);
void MQTTPersistence_wrapMsgID(Clients *c);
@ -59,6 +73,7 @@ typedef struct
int retained;
int dup;
int msgid;
MQTTProperties properties;
} MQTTPersistence_message;
typedef struct

2
src/MQTTProtocolClient.c
View File

@ -309,7 +309,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
if (m->MQTTVersion >= MQTTVERSION_5)
m->properties = MQTTProperties_copy(&publish->properties);
m->nextMessageType = PUBREL;
if ( ( listElem = ListFindItem(client->inboundMsgs, &(m->msgid), messageIDCompare) ) != NULL )
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);

50
test/CMakeLists.txt
View File

@ -597,6 +597,56 @@ SET_TESTS_PROPERTIES(
PROPERTIES TIMEOUT 540
)
ADD_EXECUTABLE(
test95
test95.c
)
TARGET_LINK_LIBRARIES(
test95
paho-mqtt3a
)
ADD_TEST(
NAME test95-1-offline-buffering-send-disconnected
COMMAND test95 "--test_no" "1" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY}
)
ADD_TEST(
NAME test95-2-offline-buffering-send-disconnected-serverURIs
COMMAND test95 "--test_no" "2" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY}
)
ADD_TEST(
NAME test95-3-offline-buffering-auto-reconnect
COMMAND test95 "--test_no" "3" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY}
)
ADD_TEST(
NAME test95-4-offline-buffering-auto-reconnect-serverURIs
COMMAND test95 "--test_no" "4" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY}
)
ADD_TEST(
NAME test95-5-offline-buffering-max-buffered
COMMAND test95 "--test_no" "5" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY}
)
ADD_TEST(
NAME test95-6-offline-buffering-max-buffered-binary-will
COMMAND test95 "--test_no" "6" "--connection" ${MQTT_TEST_BROKER} "--proxy_connection" ${MQTT_TEST_PROXY}
)
SET_TESTS_PROPERTIES(
test95-1-offline-buffering-send-disconnected
test95-2-offline-buffering-send-disconnected-serverURIs
test95-3-offline-buffering-auto-reconnect
test95-4-offline-buffering-auto-reconnect-serverURIs
test95-5-offline-buffering-max-buffered
test95-6-offline-buffering-max-buffered-binary-will
PROPERTIES TIMEOUT 540
)
ADD_EXECUTABLE(
test10
test10.c

9
test/test1.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2017 IBM Corp.
* 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
@ -22,13 +22,6 @@
* Tests for the MQ Telemetry MQTT C client
*/
/*
#if !defined(_RTSHEADER)
#include <rts.h>
#endif
*/
#include "MQTTClient.h"
#include <string.h>
#include <stdlib.h>

58
test/test10.c
View File

@ -391,14 +391,16 @@ int test_client_topic_aliases(struct Options options)
int count = 0;
char* test_topic = "test_client_topic_aliases";
int topicAliasMaximum = 0;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_client_topic_aliases\" name=\"client topic aliases\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 1 - client topic aliases");
rc = MQTTClient_create(&c, options.connection, "client_topic_alias_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "client_topic_alias_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -647,14 +649,16 @@ int test_server_topic_aliases(struct Options options)
int topicAliasMaximum = 0;
int qos = 0;
const int msg_count = 3;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_server_topic_aliases\" name=\"server topic aliases\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 2 - server topic aliases");
rc = MQTTClient_create(&c, options.connection, "server_topic_alias_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "server_topic_alias_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -784,14 +788,16 @@ int test_subscription_ids(struct Options options)
char* test_topic = "test_subscription_ids";
const int msg_count = 1;
int subsids = 1;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_subscription_ids\" name=\"subscription ids\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 3 - subscription ids");
rc = MQTTClient_create(&c, options.connection, "subscription_ids",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "subscription_ids",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -917,6 +923,7 @@ int test_flow_control(struct Options options)
int rc = 0, i = 0, count = 0;
char* test_topic = "test_flow_control";
int receive_maximum = 65535;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_flow_control\" name=\"flow control\"");
global_start_time = start_clock();
@ -925,8 +932,9 @@ int test_flow_control(struct Options options)
//MQTTClient_setTraceCallback(test_flow_control_trace_callback);
rc = MQTTClient_create(&c, options.connection, "flow_control",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "flow_control",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
goto exit;
@ -1012,6 +1020,7 @@ int test_error_reporting(struct Options options)
int rc = 0, i = 0, count = 0;
char* test_topic = "test_error_reporting";
int receive_maximum = 65535;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_error_reporting\" name=\"error reporting\"");
global_start_time = start_clock();
@ -1020,8 +1029,9 @@ int test_error_reporting(struct Options options)
//MQTTClient_setTraceCallback(test_flow_control_trace_callback);
rc = MQTTClient_create(&c, options.connection, "error_reporting",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "error_reporting",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
goto exit;
@ -1140,6 +1150,7 @@ int test_qos_1_2_errors(struct Options options)
int rc = 0, i = 0, count = 0;
char* test_topic = "test_qos_1_2_errors";
int receive_maximum = 65535;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_qos_1_2_errors\" name=\"qos 1 2 errors\"");
global_start_time = start_clock();
@ -1148,8 +1159,9 @@ int test_qos_1_2_errors(struct Options options)
//MQTTClient_setTraceCallback(test_trace_callback);
rc = MQTTClient_create(&c, options.connection, "error_reporting",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "error_reporting",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
goto exit;
@ -1354,14 +1366,16 @@ int test_request_response(struct Options options)
char* test_topic = "test_request_response";
const int msg_count = 1;
int subsids = 1;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_request_response\" name=\"request/response\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 7 - request response");
rc = MQTTClient_create(&c, options.connection, "request_response",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "request_response",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -1529,14 +1543,16 @@ int test_subscribe_options(struct Options options)
int count = 0;
const int msg_count = 1;
MQTTSubscribe_options subopts = MQTTSubscribe_options_initializer;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_subscribe_options\" name=\"subscribe options\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 8 - subscribe options");
rc = MQTTClient_create(&c, options.connection, "subscribe_options",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "subscribe_options",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -1673,14 +1689,16 @@ int test_shared_subscriptions(struct Options options)
const int msg_count = 1;
MQTTSubscribe_options subopts = MQTTSubscribe_options_initializer;
int i;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test_shared_subscriptions\" name=\"shared subscriptions\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 8 - shared subscriptions");
rc = MQTTClient_create(&c, options.connection, "shared_subscriptions",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "shared_subscriptions",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -1688,8 +1706,8 @@ int test_shared_subscriptions(struct Options options)
goto exit;
}
rc = MQTTClient_create(&d, options.connection, "shared_subscriptions_1",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
rc = MQTTClient_createWithOptions(&d, options.connection, "shared_subscriptions_1",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{

170
test/test15.c
View File

@ -407,14 +407,16 @@ int test1(struct Options options)
MQTTResponse response = MQTTResponse_initializer;
int rc = 0;
char* test_topic = "C client test1";
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test1\" name=\"single threaded client using receive\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 1 - single threaded client using receive");
rc = MQTTClient_create(&c, options.connection, "single_threaded_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&c, options.connection, "single_threaded_test",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
{
@ -647,13 +649,16 @@ int test2(struct Options options)
MQTTSubscribe_options subopts = MQTTSubscribe_options_initializer;
int rc = 0;
char* test_topic = "C client test2";
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test1\" name=\"multi-threaded client using callbacks\"");
MyLog(LOGA_INFO, "Starting test 2 - multi-threaded client using callbacks");
global_start_time = start_clock();
failures = 0;
MQTTClient_create(&c, options.connection, "multi_threaded_sample", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
MQTTClient_createWithOptions(&c, options.connection, "multi_threaded_sample",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
opts.keepAliveInterval = 20;
opts.cleanstart = 1;
@ -724,12 +729,15 @@ int test3(struct Options options)
MQTTClient_connectOptions opts = MQTTClient_connectOptions_initializer5;
MQTTClient_willOptions wopts = MQTTClient_willOptions_initializer;
MQTTResponse response;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test1\" name=\"connack return codes\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 3 - connack return codes");
createOpts.MQTTVersion = MQTTVERSION_5;
#if 0
/* clientid too long (RC = 2) */
rc = MQTTClient_create(&c, options.connection, "client_ID_too_long_for_MQTT_protocol_version_3",
@ -740,7 +748,8 @@ int test3(struct Options options)
MQTTClient_destroy(&c);
#endif
/* broker unavailable (RC = 3) - TDD when allow_anonymous not set*/
rc = MQTTClient_create(&c, options.connection, "The C Client", MQTTCLIENT_PERSISTENCE_NONE, NULL);
rc = MQTTClient_createWithOptions(&c, options.connection, "The C Client", MQTTCLIENT_PERSISTENCE_NONE,
NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
#if 0
rc = MQTTClient_connect(c, &opts);
@ -790,7 +799,7 @@ Test 4: client persistence 1
*********************************************************************/
int test4_run(int qos)
int test4_run(int qos, int start_mqtt_version, int restore_mqtt_version)
{
char* testname = "test 4";
char* topic = "Persistence test 1";
@ -806,49 +815,112 @@ int test4_run(int qos)
int count = 3;
MQTTProperty property;
MQTTProperties props = MQTTProperties_initializer;
MQTTProperties pub_props = MQTTProperties_initializer;
MQTTResponse response = MQTTResponse_initializer;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
int i, rc;
failures = 0;
MyLog(LOGA_INFO, "Starting test 4 - persistence, qos %d", qos);
MyLog(LOGA_INFO, "Starting test 4 - persistence, qos %d, MQTT versions: %s then %s", qos,
(start_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1",
(restore_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1");
MQTTClient_create(&c, options.connection, "xrctest15_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = start_mqtt_version;
rc = MQTTClient_createWithOptions(&c, options.connection, "xrctest15_test_4",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
/* we might get some tokens back because they may not be cleaned up until
* we connect cleanstart
*/
rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
if (tokens)
MQTTClient_free(tokens);
opts.keepAliveInterval = 20;
opts.reliable = 0;
opts.MQTTVersion = options.MQTTVersion;
opts.MQTTVersion = start_mqtt_version;
if (options.haconnections != NULL)
{
opts.serverURIs = options.haconnections;
opts.serverURIcount = options.hacount;
}
MyLog(LOGA_DEBUG, "Cleanup by connecting clean start, add session expiry > 0\n");
opts.cleanstart = 1;
property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
property.value.integer4 = 30; /* in seconds */
MQTTProperties_add(&props, &property);
response = MQTTClient_connect5(c, &opts, &props, NULL);
assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS,
if (start_mqtt_version == MQTTVERSION_5)
{
MyLog(LOGA_DEBUG, "Cleanup by connecting clean start, add session expiry > 0\n");
opts.cleanstart = 1;
property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
property.value.integer4 = 30; /* in seconds */
MQTTProperties_add(&props, &property);
response = MQTTClient_connect5(c, &opts, &props, NULL);
assert("Good rc from connect", response.reasonCode == MQTTCLIENT_SUCCESS,
"rc was %d", response.reasonCode);
MQTTResponse_free(response);
if (response.reasonCode != MQTTCLIENT_SUCCESS)
return -1;
MQTTResponse_free(response);
if (response.reasonCode != MQTTCLIENT_SUCCESS)
return -1;
}
else
{
MyLog(LOGA_DEBUG, "Cleanup by connecting clean session, then reconnecting non-cleansession\n");
opts.cleanstart = 0; /* only applies to MQTT V5 */
opts.cleansession = 1;
rc = MQTTClient_connect(c, &opts);
assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
rc = MQTTClient_disconnect(c, 1000);
assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
opts.cleansession = 0;
rc = MQTTClient_connect(c, &opts);
assert("Good rc from connect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
if (rc != MQTTCLIENT_SUCCESS)
return -1;
}
/* subscribe so we can get messages back */
response = MQTTClient_subscribe5(c, topic, subsqos, NULL, NULL);
assert("Good rc from subscribe", response.reasonCode == subsqos, "rc was %d", response.reasonCode);
if (start_mqtt_version == MQTTVERSION_5)
{
response = MQTTClient_subscribe5(c, topic, subsqos, NULL, NULL);
assert("Good rc from subscribe", response.reasonCode == subsqos, "rc was %d", response.reasonCode);
}
else
{
rc = MQTTClient_subscribe(c, topic, subsqos);
assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
}
/* send messages so that we can receive the same ones */
if (start_mqtt_version == MQTTVERSION_5)
{
property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
property.value.data.data = "test user property";
property.value.data.len = (int)strlen(property.value.data.data);
property.value.value.data = "test user property value";
property.value.value.len = (int)strlen(property.value.value.data);
MQTTProperties_add(&pub_props, &property);
}
for (i = 0; i < count; ++i)
{
sprintf(buffer, "Message sequence no %d", i);
response = MQTTClient_publish5(c, topic, 10, buffer, qos, 0, NULL, NULL);
assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
if (start_mqtt_version == MQTTVERSION_5)
{
response = MQTTClient_publish5(c, topic, 10, buffer, qos, 0, &pub_props, NULL);
assert("Good rc from publish", response.reasonCode == MQTTCLIENT_SUCCESS, "rc was %d", response.reasonCode);
}
else
{
rc = MQTTClient_publish(c, topic, 10, buffer, qos, 0, NULL);
assert("Good rc from publish", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
}
}
/* disconnect immediately without receiving the incoming messages */
MQTTClient_disconnect5(c, 0, MQTTREASONCODE_SUCCESS, NULL); /* now there should be "orphaned" publications */
if (start_mqtt_version == MQTTVERSION_5)
rc = MQTTClient_disconnect5(c, 0, MQTTREASONCODE_SUCCESS, NULL); /* now there should be "orphaned" publications */
else
rc = MQTTClient_disconnect(c, 0); /* now there should be "orphaned" publications */
assert("Good rc from disconnect", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
@ -865,9 +937,19 @@ int test4_run(int qos)
mytoken = tokens[0];
}
MQTTProperties_free(&props);
MQTTProperties_free(&pub_props);
MQTTClient_destroy(&c); /* force re-reading persistence on create */
MQTTClient_create(&c, options.connection, "xrctest15_test_4", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = restore_mqtt_version;
rc = MQTTClient_createWithOptions(&c, options.connection, "xrctest15_test_4",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
if (start_mqtt_version == MQTTVERSION_5 && restore_mqtt_version == MQTTVERSION_3_1_1)
{
assert("Persistence error from create", rc == MQTTCLIENT_PERSISTENCE_ERROR, "rc was %d", rc);
goto exit;
}
else
assert("Good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
rc = MQTTClient_getPendingDeliveryTokens(c, &tokens);
assert("getPendingDeliveryTokens rc == 0", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
@ -901,6 +983,9 @@ int test4_run(int qos)
{
assert("No duplicates should be received for qos 2", qos == 1, "qos is %d", qos);
MyLog(LOGA_DEBUG, "Duplicate message id %d", m->msgid);
assert("properties are received", m->properties.count > 0, "property count is %d",
m->properties.count);
logProperties(&m->properties);
MQTTClient_freeMessage(&m);
MQTTClient_free(topicName);
dup = 1;
@ -911,6 +996,13 @@ int test4_run(int qos)
{
MyLog(LOGA_DEBUG, "Received message id %d", m->msgid);
assert("topicName is correct", strcmp(topicName, topic) == 0, "topicName is %s", topicName);
if (start_mqtt_version == MQTTVERSION_5)
assert("properties are received", m->properties.count > 0, "property count is %d",
m->properties.count);
else
assert("properties are not received", m->properties.count == 0, "property count is %d",
m->properties.count);
logProperties(&m->properties);
MQTTClient_freeMessage(&m);
MQTTClient_free(topicName);
}
@ -926,7 +1018,7 @@ int test4_run(int qos)
MQTTClient_destroy(&c);
/* TODO - unused -remove? exit: */
exit:
MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
(failures == 0) ? "passed" : "failed", testname, tests, failures);
@ -937,9 +1029,12 @@ int test4_run(int qos)
int test4(struct Options options)
{
int rc = 0;
fprintf(xml, "<testcase classname=\"test1\" name=\"persistence\"");
fprintf(xml, "<testcase classname=\"test4\" name=\"persistence\"");
global_start_time = start_clock();
rc = test4_run(1) + test4_run(2);
rc = test4_run(1, MQTTVERSION_5, MQTTVERSION_5) +
test4_run(2, MQTTVERSION_5, MQTTVERSION_5) +
test4_run(2, MQTTVERSION_3_1_1, MQTTVERSION_5) +
test4_run(2, MQTTVERSION_5, MQTTVERSION_3_1_1);
fprintf(xml, " time=\"%ld\" >\n", elapsed(global_start_time) / 1000);
if (cur_output != output)
{
@ -970,13 +1065,16 @@ int test5(struct Options options)
MQTTProperties props = MQTTProperties_initializer;
MQTTResponse response = MQTTResponse_initializer;
int i, rc;
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
fprintf(xml, "<testcase classname=\"test1\" name=\"disconnect with quiesce timeout should allow exchanges to complete\"");
global_start_time = start_clock();
failures = 0;
MyLog(LOGA_INFO, "Starting test 5 - disconnect with quiesce timeout should allow exchanges to complete");
MQTTClient_create(&c, options.connection, "xrctest15_test_5", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
MQTTClient_createWithOptions(&c, options.connection, "xrctest15_test_5", MQTTCLIENT_PERSISTENCE_DEFAULT,
NULL, &createOpts);
opts.keepAliveInterval = 20;
opts.cleanstart = 1;
@ -1076,6 +1174,7 @@ int test6(struct Options options)
MQTTResponse response = MQTTResponse_initializer;
int rc, count;
char* mqttsas_topic = "MQTTSAS topic";
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
failures = 0;
MyLog(LOGA_INFO, "Starting test 6 - connectionLost and will messages");
@ -1097,7 +1196,9 @@ int test6(struct Options options)
}
/* Client-1 with Will options */
rc = MQTTClient_create(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT,
NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
goto exit;
@ -1115,7 +1216,8 @@ int test6(struct Options options)
goto exit;
/* Client - 2 (multi-threaded) */
rc = MQTTClient_create(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
rc = MQTTClient_createWithOptions(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT,
NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
/* Set the callback functions for the client */
@ -1186,6 +1288,7 @@ int test6a(struct Options options)
int rc, count;
MQTTResponse response = MQTTResponse_initializer;
char* mqttsas_topic = "MQTTSAS topic";
MQTTClient_createOptions createOpts = MQTTClient_createOptions_initializer;
failures = 0;
MyLog(LOGA_INFO, "Starting test 6 - connectionLost and binary will messages");
@ -1208,7 +1311,9 @@ int test6a(struct Options options)
}
/* Client-1 with Will options */
rc = MQTTClient_create(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = MQTTVERSION_5;
rc = MQTTClient_createWithOptions(&test6_c1, options.proxy_connection, "Client_1", MQTTCLIENT_PERSISTENCE_DEFAULT,
NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTCLIENT_SUCCESS)
goto exit;
@ -1227,7 +1332,8 @@ int test6a(struct Options options)
goto exit;
/* Client - 2 (multi-threaded) */
rc = MQTTClient_create(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
rc = MQTTClient_createWithOptions(&test6_c2, options.connection, "Client_2", MQTTCLIENT_PERSISTENCE_DEFAULT,
NULL, &createOpts);
assert("good rc from create", rc == MQTTCLIENT_SUCCESS, "rc was %d\n", rc);
/* Set the callback functions for the client */

4
test/test2.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2016 IBM Corp.
* 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
@ -17,7 +17,7 @@
/**
* @file
* Multi-threaded tests for the MQ Telemetry MQTT C client
* Multi-threaded tests for the Eclipse Paho MQTT C client
*/
#include "MQTTClient.h"

2
test/test3.c
View File

@ -17,7 +17,7 @@
/**
* @file
* SSL tests for the MQ Telemetry MQTT C client
* SSL tests for the Eclipse Paho MQTT C client
*/
#include "MQTTClient.h"

83
test/test45.c
View File

@ -1326,7 +1326,10 @@ int test7_messageArrived(void* context, char* topicName, int topicLen, MQTTAsync
MyLog(LOGA_DEBUG, "Test7: received message id %d", message->msgid);
test7_messageCount++;
if (message->qos == 1 && message->dup == 1)
; /* ignore dups */
else
test7_messageCount++;
MQTTAsync_freeMessage(&message);
MQTTAsync_free(topicName);
@ -1369,7 +1372,7 @@ void test7_onConnect(void* context, MQTTAsync_successData5* response)
Test7: Pending tokens
*********************************************************************/
int test7(struct Options options)
int test7_run(int qos, int start_mqtt_version, int restore_mqtt_version)
{
int subsqos = 2;
MQTTAsync c;
@ -1383,14 +1386,20 @@ int test7(struct Options options)
int msg_count = 6;
MQTTProperty property;
MQTTProperties props = MQTTProperties_initializer;
MQTTAsync_createOptions createOpts = MQTTAsync_createOptions_initializer;
MyLog(LOGA_INFO, "Starting test 7 - persistence, qos %d, MQTT versions: %s then %s", qos,
(start_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1",
(restore_mqtt_version == MQTTVERSION_5) ? "5" : "3.1.1");
MyLog(LOGA_INFO, "Starting test 7 - pending tokens");
fprintf(xml, "<testcase classname=\"test4\" name=\"pending tokens\"");
global_start_time = start_clock();
test_finished = 0;
rc = MQTTAsync_create(&c, options.connection, "async_test7",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL);
createOpts.MQTTVersion = start_mqtt_version;
MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTASYNC_SUCCESS)
{
@ -1404,7 +1413,7 @@ int test7(struct Options options)
opts.keepAliveInterval = 20;
opts.username = "testuser";
opts.password = "testpassword";
opts.MQTTVersion = options.MQTTVersion;
opts.MQTTVersion = start_mqtt_version;
opts.will = &wopts;
opts.will->message = "will message";
@ -1418,17 +1427,26 @@ int test7(struct Options options)
/* connect clean and then leave messages lying around */
test_finished = 0;
test7_subscribed = 0;
MyLog(LOGA_DEBUG, "Connecting");
opts.cleanstart = 1;
opts.connectProperties = &props;
property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
property.value.integer4 = 999999;
MQTTProperties_add(opts.connectProperties, &property);
opts.onSuccess5 = test7_onConnect;
rc = MQTTAsync_connect(c, &opts);
assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
if (rc != MQTTASYNC_SUCCESS)
goto exit;
if (start_mqtt_version == MQTTVERSION_5)
{
opts.cleanstart = 1;
opts.connectProperties = &props;
property.identifier = MQTTPROPERTY_CODE_SESSION_EXPIRY_INTERVAL;
property.value.integer4 = 999999;
MQTTProperties_add(opts.connectProperties, &property);
opts.onSuccess5 = test7_onConnect;
rc = MQTTAsync_connect(c, &opts);
assert("Good rc from connect", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
if (rc != MQTTASYNC_SUCCESS)
goto exit;
}
else
{
/* MQTT 3 version */
}
while (!test7_subscribed)
#if defined(WIN32)
@ -1439,12 +1457,13 @@ int test7(struct Options options)
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.qos = qos;
pubmsg.retained = 0;
rc = MQTTAsync_send(c, test7_topic, pubmsg.payloadlen, pubmsg.payload, pubmsg.qos, pubmsg.retained, &ropts);
assert("Good rc from send", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
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);*/
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);
@ -1452,12 +1471,11 @@ int test7(struct Options options)
test7_messageCount = 0;
int i = 0;
pubmsg.qos = 2;
pubmsg.qos = qos;
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, test7_topic, &pubmsg, &ropts);
}
@ -1485,7 +1503,9 @@ int test7(struct Options options)
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);
createOpts.MQTTVersion = restore_mqtt_version;
rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
if (rc != MQTTASYNC_SUCCESS)
{
@ -1510,6 +1530,7 @@ int test7(struct Options options)
assert("Good rc from setCallbacks", rc == MQTTASYNC_SUCCESS, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Reconnecting");
opts.MQTTVersion = restore_mqtt_version;
opts.context = c;
opts.cleanstart = 0;
if (MQTTAsync_connect(c, &opts) != 0)
@ -1556,6 +1577,26 @@ exit:
}
int test7(struct Options options)
{
int rc = 0;
fprintf(xml, "<testcase classname=\"test7\" name=\"persistence\"");
global_start_time = start_clock();
rc = test7_run(1, MQTTVERSION_5, MQTTVERSION_5) +
test7_run(2, MQTTVERSION_5, MQTTVERSION_5) /*+
test7_run(2, MQTTVERSION_3_1_1, MQTTVERSION_5) +
test7_run(2, MQTTVERSION_5, MQTTVERSION_3_1_1)*/;
fprintf(xml, " time=\"%ld\" >\n", elapsed(global_start_time) / 1000);
if (cur_output != output)
{
fprintf(xml, "%s", output);
cur_output = output;
}
fprintf(xml, "</testcase>\n");
return rc;
}
/*********************************************************************

2
test/test5.c
View File

@ -17,7 +17,7 @@
/**
* @file
* SSL tests for the MQ Telemetry Asynchronous MQTT C client
* SSL tests for the Eclipse Paho Asynchronous MQTT C client
*/
#include "MQTTAsync.h"

35
test/test8.c
View File

@ -1,28 +1,25 @@
/*--------------------------------------------------------------------*/
/* [Platforms]UNIX NT[/Platforms] */
/* [Title]MQ Telemetry MQTT Async C client tests - HA and connect */
/* failures */
/* [/Title] */
/* [Testclasses]stcom1 stmqcom1[/Category] */
/* [Category]MQ Telemetry[/Category] */
/* */
/* Copyright IBM 2013 */
/* All rights reserved. */
/*--------------------------------------------------------------------*/
/*******************************************************************************
* Copyright (c) 2012, 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
*******************************************************************************/
/**
* @file
* Tests for the MQ Telemetry MQTT Async C client
* Tests for the Paho MQTT Async C client
*/
/*
#if !defined(_RTSHEADER)
#include <rts.h>
#endif
*/
#include "MQTTAsync.h"
#include <string.h>
#include <stdlib.h>

4
test/test9.c
View File

@ -19,7 +19,7 @@
/**
* @file
* Offline buffering and automatic reconnect tests for the MQ Telemetry Asynchronous MQTT C client
* Offline buffering and automatic reconnect tests for the Paho Asynchronous MQTT C client
*
*/
@ -2121,7 +2121,7 @@ int main(int argc, char** argv)
int (*tests[])() = { NULL, test1, test2, test3, test4, test5, test6, test7};
time_t randtime;
srand((unsigned) time(&randtime));
srand((unsigned) time(&randtime));
sprintf(unique, "%u", rand());
MyLog(LOGA_INFO, "Random prefix/suffix is %s", unique);

2358
test/test95.c Normal file
View File

File diff suppressed because it is too large Load Diff