Skip to content

P
paho.mqtt.c
Commit e5a1a084 authored by Ian Craggs's avatar Ian Craggs
Browse files
Options

Merge branch 'master' into develop

parents 60a470cc f804a18e
Expand all Show whitespace changes
Inline Side-by-side
Showing with 323 additions and 240 deletions
Makefile 100644 → 100755
View file @ e5a1a084
......@@ -110,9 +110,9 @@ CCFLAGS_SO = -g -fPIC -Os -Wall -fvisibility=hidden
FLAGS_EXE = -I ${srcdir} -lpthread -L ${blddir}
LDFLAGS_C = -shared -Wl,-soname,lib$(MQTTLIB_C).so.${MAJOR_VERSION} -Wl,-init,MQTTClient_init -lpthread
LDFLAGS_CS = -shared -Wl,-soname,lib$(MQTTLIB_CS).so.${MAJOR_VERSION} -lpthread -ldl -lcrypto -lssl -Wl,-no-whole-archive -Wl,-init,MQTTClient_init
LDFLAGS_CS = -shared -Wl,-soname,lib$(MQTTLIB_CS).so.${MAJOR_VERSION} -lpthread -ldl -lssl -lcrypto -Wl,-no-whole-archive -Wl,-init,MQTTClient_init
LDFLAGS_A = -shared -Wl,-soname,lib${MQTTLIB_A}.so.${MAJOR_VERSION} -Wl,-init,MQTTAsync_init -lpthread
LDFLAGS_AS = -shared -Wl,-soname,lib${MQTTLIB_AS}.so.${MAJOR_VERSION} -lpthread -ldl -lcrypto -lssl -Wl,-no-whole-archive -Wl,-init,MQTTAsync_init
LDFLAGS_AS = -shared -Wl,-soname,lib${MQTTLIB_AS}.so.${MAJOR_VERSION} -lpthread -ldl -lssl -lcrypto -Wl,-no-whole-archive -Wl,-init,MQTTAsync_init
all: build
......@@ -236,9 +236,9 @@ CCFLAGS_SO = -g -fPIC -Os -Wall -fvisibility=hidden -Wno-deprecated-declarations
FLAGS_EXE = -I ${srcdir} -lpthread -L ${blddir}
LDFLAGS_C = -shared -Wl,-install_name,lib$(MQTTLIB_C).so.${MAJOR_VERSION} -Wl,-init,_MQTTClient_init -lpthread
LDFLAGS_CS = -shared -Wl,-install_name,lib$(MQTTLIB_CS).so.${MAJOR_VERSION} -lpthread -ldl -lcrypto -lssl -Wl,-init,_MQTTClient_init
LDFLAGS_CS = -shared -Wl,-install_name,lib$(MQTTLIB_CS).so.${MAJOR_VERSION} -lpthread -ldl -lssl -lcrypto -Wl,-init,_MQTTClient_init
LDFLAGS_A = -shared -Wl,-install_name,lib${MQTTLIB_A}.so.${MAJOR_VERSION} -Wl,-init,_MQTTAsync_init -lpthread
LDFLAGS_AS = -shared -Wl,-install_name,lib${MQTTLIB_AS}.so.${MAJOR_VERSION} -lpthread -ldl -lcrypto -lssl -Wl,-init,_MQTTAsync_init
LDFLAGS_AS = -shared -Wl,-install_name,lib${MQTTLIB_AS}.so.${MAJOR_VERSION} -lpthread -ldl -lssl -lcrypto -Wl,-init,_MQTTAsync_init
all: build
......
src/Clients.h 100644 → 100755
View file @ e5a1a084
......@@ -162,8 +162,8 @@ typedef struct
typedef struct
{
char* clientID; /**< the string id of the client */
char* username; /**< MQTT v3.1 user name */
char* password; /**< MQTT v3.1 password */
const char* username; /**< MQTT v3.1 user name */
const char* password; /**< MQTT v3.1 password */
unsigned int cleansession : 1; /**< MQTT clean session flag */
unsigned int connected : 1; /**< whether it is currently connected */
unsigned int good : 1; /**< if we have an error on the socket we turn this off */
......@@ -197,7 +197,7 @@ int clientSocketCompare(void* a, void* b);
*/
typedef struct
{
char* version;
const char* version;
List* clients;
} ClientStates;
......
src/Log.c
View file @ e5a1a084
......@@ -439,6 +439,7 @@ void Log_stackTrace(int log_level, int msgno, int thread_id, int current_depth,
if (log_level < trace_settings.trace_level)
return;
Thread_lock_mutex(log_mutex);
cur_entry = Log_pretrace();
memcpy(&(cur_entry->ts), &ts, sizeof(ts));
......@@ -458,6 +459,7 @@ void Log_stackTrace(int log_level, int msgno, int thread_id, int current_depth,
}
Log_posttrace(log_level, cur_entry);
Thread_unlock_mutex(log_mutex);
}
......
src/MQTTAsync.c
View file @ e5a1a084
This diff is collapsed.
src/MQTTAsync.h 100644 → 100755
View file @ e5a1a084
......@@ -76,6 +76,10 @@
/// @cond EXCLUDE
#if defined(__cplusplus)
extern "C" {
#endif
#if !defined(MQTTASYNC_H)
#define MQTTASYNC_H
......@@ -476,7 +480,7 @@ DLLExport int MQTTAsync_setCallbacks(MQTTAsync handle, void* context, MQTTAsync_
* @return ::MQTTASYNC_SUCCESS if the client is successfully created, otherwise
* an error code is returned.
*/
DLLExport int MQTTAsync_create(MQTTAsync* handle, char* serverURI, char* clientId,
DLLExport int MQTTAsync_create(MQTTAsync* handle, const char* serverURI, const char* clientId,
int persistence_type, void* persistence_context);
/**
......@@ -494,13 +498,13 @@ DLLExport int MQTTAsync_create(MQTTAsync* handle, char* serverURI, char* clientI
typedef struct
{
/** The eyecatcher for this structure. must be MQTW. */
char struct_id[4];
const char struct_id[4];
/** The version number of this structure. Must be 0 */
int struct_version;
/** The LWT topic to which the LWT message will be published. */
char* topicName;
const char* topicName;
/** The LWT payload. */
char* message;
const char* message;
/**
* The retained flag for the LWT message (see MQTTAsync_message.retained).
*/
......@@ -529,24 +533,24 @@ typedef struct
typedef struct
{
/** The eyecatcher for this structure. Must be MQTS */
char struct_id[4];
const char struct_id[4];
/** The version number of this structure. Must be 0 */
int struct_version;
/** The file in PEM format containing the public digital certificates trusted by the client. */
char* trustStore;
const char* trustStore;
/** The file in PEM format containing the public certificate chain of the client. It may also include
* the client's private key.
*/
char* keyStore;
const char* keyStore;
/** If not included in the sslKeyStore, this setting points to the file in PEM format containing
* the client's private key.
*/
char* privateKey;
const char* privateKey;
/** The password to load the client's privateKey if encrypted. */
char* privateKeyPassword;
const char* privateKeyPassword;
/**
* The list of cipher suites that the client will present to the server during the SSL handshake. For a
......@@ -556,7 +560,7 @@ typedef struct
* those offering no encryption- will be considered.
* This setting can be used to set an SSL anonymous connection ("aNULL" string value, for instance).
*/
char* enabledCipherSuites;
const char* enabledCipherSuites;
/** True/False option to enable verification of the server certificate **/
int enableServerCertAuth;
......@@ -573,8 +577,8 @@ typedef struct
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 or 3.
const char struct_id[4];
/** The version number of this structure. Must be 0, 1 or 2.
* 0 signifies no SSL options and no serverURIs
* 1 signifies no serverURIs
* 2 signifies no MQTTVersion
......@@ -628,13 +632,13 @@ typedef struct
* and authorisation by user name and password. This is the user name
* parameter.
*/
char* username;
const char* username;
/**
* MQTT servers that support the MQTT v3.1 protocol provide authentication
* and authorisation by user name and password. This is the password
* parameter.
*/
char* password;
const char* password;
/**
* The time interval in seconds to allow a connect to complete.
*/
......@@ -678,7 +682,7 @@ typedef struct
* a server running on the local machines with the default MQTT port, specify
* <i>tcp://localhost:1883</i>.
*/
char** serverURIs;
char* const* serverURIs;
/**
* Sets the version of MQTT to be used on the connect.
* MQTTVERSION_DEFAULT (0) = default: start with 3.1.1, and if that fails, fall back to 3.1
......@@ -711,13 +715,13 @@ typedef struct
* <b>5</b>: Connection refused: Not authorized<br>
* <b>6-255</b>: Reserved for future use<br>
*/
DLLExport int MQTTAsync_connect(MQTTAsync handle, MQTTAsync_connectOptions* options);
DLLExport int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options);
typedef struct
{
/** The eyecatcher for this structure. Must be MQTD. */
char struct_id[4];
const char struct_id[4];
/** The version number of this structure. Must be 0 or 1. 0 signifies no SSL options */
int struct_version;
/**
......@@ -766,7 +770,7 @@ typedef struct
* the server. An error code is returned if the client was unable to disconnect
* from the server
*/
DLLExport int MQTTAsync_disconnect(MQTTAsync handle, MQTTAsync_disconnectOptions* options);
DLLExport int MQTTAsync_disconnect(MQTTAsync handle, const MQTTAsync_disconnectOptions* options);
/**
......@@ -793,7 +797,7 @@ DLLExport int MQTTAsync_isConnected(MQTTAsync handle);
* An error code is returned if there was a problem registering the
* subscription.
*/
DLLExport int MQTTAsync_subscribe(MQTTAsync handle, char* topic, int qos, MQTTAsync_responseOptions* response);
DLLExport int MQTTAsync_subscribe(MQTTAsync handle, const char* topic, int qos, MQTTAsync_responseOptions* response);
/**
......@@ -813,7 +817,7 @@ DLLExport int MQTTAsync_subscribe(MQTTAsync handle, char* topic, int qos, MQTTAs
* An error code is returned if there was a problem registering the
* subscriptions.
*/
DLLExport int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char** topic, int* qos, MQTTAsync_responseOptions* response);
DLLExport int MQTTAsync_subscribeMany(MQTTAsync handle, size_t count, char* const* topic, int* qos, MQTTAsync_responseOptions* response);
/**
* This function attempts to remove an existing subscription made by the
......@@ -827,7 +831,7 @@ DLLExport int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char** topic,
* An error code is returned if there was a problem removing the
* subscription.
*/
DLLExport int MQTTAsync_unsubscribe(MQTTAsync handle, char* topic, MQTTAsync_responseOptions* response);
DLLExport int MQTTAsync_unsubscribe(MQTTAsync handle, const char* topic, MQTTAsync_responseOptions* response);
/**
* This function attempts to remove existing subscriptions to a list of topics
......@@ -841,7 +845,7 @@ DLLExport int MQTTAsync_unsubscribe(MQTTAsync handle, char* topic, MQTTAsync_res
* @return ::MQTTASYNC_SUCCESS if the subscriptions are removed.
* An error code is returned if there was a problem removing the subscriptions.
*/
DLLExport int MQTTAsync_unsubscribeMany(MQTTAsync handle, int count, char** topic, MQTTAsync_responseOptions* response);
DLLExport int MQTTAsync_unsubscribeMany(MQTTAsync handle, size_t count, char* const* topic, MQTTAsync_responseOptions* response);
/**
......@@ -862,7 +866,7 @@ DLLExport int MQTTAsync_unsubscribeMany(MQTTAsync handle, int count, char** topi
* @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, char* destinationName, int payloadlen, void* payload, int qos, int retained,
DLLExport int MQTTAsync_send(MQTTAsync handle, const char* destinationName, size_t payloadlen, void* payload, int qos, int retained,
MQTTAsync_responseOptions* response);
......@@ -881,7 +885,7 @@ DLLExport int MQTTAsync_send(MQTTAsync handle, char* destinationName, int payloa
* @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_sendMessage(MQTTAsync handle, char* destinationName, MQTTAsync_message* msg, MQTTAsync_responseOptions* response);
DLLExport int MQTTAsync_sendMessage(MQTTAsync handle, const char* destinationName, const MQTTAsync_message* msg, MQTTAsync_responseOptions* response);
/**
......@@ -1506,3 +1510,7 @@ exit:
#endif
#ifdef __cplusplus
}
#endif
src/MQTTClient.c
View file @ e5a1a084
......@@ -23,6 +23,7 @@
* Ian Craggs - fix for bug 432903 - queue persistence
* Ian Craggs - MQTT 3.1.1 support
* Ian Craggs - fix for bug 438176 - MQTT version selection
* Rong Xiang, Ian Craggs - C++ compatibility
*******************************************************************************/
/**
......@@ -228,7 +229,7 @@ long MQTTClient_elapsed(struct timeval start)
#endif
int MQTTClient_create(MQTTClient* handle, char* serverURI, char* clientId,
int MQTTClient_create(MQTTClient* handle, const char* serverURI, const char* clientId,
int persistence_type, void* persistence_context)
{
int rc = 0;
......@@ -276,8 +277,7 @@ int MQTTClient_create(MQTTClient* handle, char* serverURI, char* clientId,
m->ssl = 1;
}
#endif
m->serverURI = malloc(strlen(serverURI)+1);
strcpy(m->serverURI, serverURI);
m->serverURI = MQTTStrdup(serverURI);
ListAppend(handles, m, sizeof(MQTTClients));
m->c = malloc(sizeof(Clients));
......@@ -286,8 +286,7 @@ int MQTTClient_create(MQTTClient* handle, char* serverURI, char* clientId,
m->c->outboundMsgs = ListInitialize();
m->c->inboundMsgs = ListInitialize();
m->c->messageQueue = ListInitialize();
m->c->clientID = malloc(strlen(clientId)+1);
strcpy(m->c->clientID, clientId);
m->c->clientID = MQTTStrdup(clientId);
m->connect_sem = Thread_create_sem();
m->connack_sem = Thread_create_sem();
m->suback_sem = Thread_create_sem();
......@@ -367,8 +366,7 @@ void MQTTClient_destroy(MQTTClient* handle)
if (m->c)
{
int saved_socket = m->c->net.socket;
char* saved_clientid = malloc(strlen(m->c->clientID)+1);
strcpy(saved_clientid, m->c->clientID);
char* saved_clientid = MQTTStrdup(m->c->clientID);
#if !defined(NO_PERSISTENCE)
MQTTPersistence_close(m->c);
#endif
......@@ -758,7 +756,7 @@ void Protocol_processPublication(Publish* publish, Clients* client)
}
int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* options, char* serverURI, int MQTTVersion,
int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI, int MQTTVersion,
START_TIME_TYPE start, long millisecsTimeout)
{
MQTTClients* m = handle;
......@@ -935,7 +933,7 @@ exit:
}
int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options, char* serverURI)
int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options, const char* serverURI)
{
MQTTClients* m = handle;
START_TIME_TYPE start;
......@@ -962,12 +960,10 @@ int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options,
if (options->will && options->will->struct_version == 0)
{
m->c->will = malloc(sizeof(willMessages));
m->c->will->msg = malloc(strlen(options->will->message) + 1);
strcpy(m->c->will->msg, options->will->message);
m->c->will->msg = MQTTStrdup(options->will->message);
m->c->will->qos = options->will->qos;
m->c->will->retained = options->will->retained;
m->c->will->topic = malloc(strlen(options->will->topicName) + 1);
strcpy(m->c->will->topic, options->will->topicName);
m->c->will->topic = MQTTStrdup(options->will->topicName);
}
#if defined(OPENSSL)
......@@ -992,30 +988,15 @@ int MQTTClient_connectURI(MQTTClient handle, MQTTClient_connectOptions* options,
m->c->sslopts = malloc(sizeof(MQTTClient_SSLOptions));
memset(m->c->sslopts, '\0', sizeof(MQTTClient_SSLOptions));
if (options->ssl->trustStore)
{
m->c->sslopts->trustStore = malloc(strlen(options->ssl->trustStore) + 1);
strcpy(m->c->sslopts->trustStore, options->ssl->trustStore);
}
m->c->sslopts->trustStore = MQTTStrdup(options->ssl->trustStore);
if (options->ssl->keyStore)
{
m->c->sslopts->keyStore = malloc(strlen(options->ssl->keyStore) + 1);
strcpy(m->c->sslopts->keyStore, options->ssl->keyStore);
}
m->c->sslopts->keyStore = MQTTStrdup(options->ssl->keyStore);
if (options->ssl->privateKey)
{
m->c->sslopts->privateKey = malloc(strlen(options->ssl->privateKey) + 1);
strcpy(m->c->sslopts->privateKey, options->ssl->privateKey);
}
m->c->sslopts->privateKey = MQTTStrdup(options->ssl->privateKey);
if (options->ssl->privateKeyPassword)
{
m->c->sslopts->privateKeyPassword = malloc(strlen(options->ssl->privateKeyPassword) + 1);
strcpy(m->c->sslopts->privateKeyPassword, options->ssl->privateKeyPassword);
}
m->c->sslopts->privateKeyPassword = MQTTStrdup(options->ssl->privateKeyPassword);
if (options->ssl->enabledCipherSuites)
{
m->c->sslopts->enabledCipherSuites = malloc(strlen(options->ssl->enabledCipherSuites) + 1);
strcpy(m->c->sslopts->enabledCipherSuites, options->ssl->enabledCipherSuites);
}
m->c->sslopts->enabledCipherSuites = MQTTStrdup(options->ssl->enabledCipherSuites);
m->c->sslopts->enableServerCertAuth = options->ssl->enableServerCertAuth;
}
#endif
......@@ -1219,13 +1200,14 @@ int MQTTClient_isConnected(MQTTClient handle)
}
int MQTTClient_subscribeMany(MQTTClient handle, int count, char** topic, int* qos)
int MQTTClient_subscribeMany(MQTTClient handle, int count, char* const* topic, int* qos)
{
MQTTClients* m = handle;
List* topics = ListInitialize();
List* qoss = ListInitialize();
int i = 0;
int rc = MQTTCLIENT_FAILURE;
int msgid = 0;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
......@@ -1254,13 +1236,19 @@ int MQTTClient_subscribeMany(MQTTClient handle, int count, char** topic, int* qo
goto exit;
}
}
if ((msgid = MQTTProtocol_assignMsgId(m->c)) == 0)
{
rc = MQTTCLIENT_MAX_MESSAGES_INFLIGHT;
goto exit;
}
for (i = 0; i < count; i++)
{
ListAppend(topics, topic[i], strlen(topic[i]));
ListAppend(qoss, &qos[i], sizeof(int));
}
rc = MQTTProtocol_subscribe(m->c, topics, qoss);
rc = MQTTProtocol_subscribe(m->c, topics, qoss, msgid);
ListFreeNoContent(topics);
ListFreeNoContent(qoss);
......@@ -1304,12 +1292,13 @@ exit:
}
int MQTTClient_subscribe(MQTTClient handle, char* topic, int qos)
int MQTTClient_subscribe(MQTTClient handle, const char* topic, int qos)
{
int rc = 0;
char *const topics[] = {(char*)topic};
FUNC_ENTRY;
rc = MQTTClient_subscribeMany(handle, 1, &topic, &qos);
rc = MQTTClient_subscribeMany(handle, 1, topics, &qos);
if (qos == MQTT_BAD_SUBSCRIBE) /* addition for MQTT 3.1.1 - error code from subscribe */
rc = MQTT_BAD_SUBSCRIBE;
FUNC_EXIT_RC(rc);
......@@ -1317,12 +1306,13 @@ int MQTTClient_subscribe(MQTTClient handle, char* topic, int qos)
}
int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char** topic)
int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char* const* topic)
{
MQTTClients* m = handle;
List* topics = ListInitialize();
int i = 0;
int rc = SOCKET_ERROR;
int msgid = 0;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
......@@ -1337,7 +1327,6 @@ int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char** topic)
rc = MQTTCLIENT_DISCONNECTED;
goto exit;
}
for (i = 0; i < count; i++)
{
if (!UTF8_validateString(topic[i]))
......@@ -1346,10 +1335,15 @@ int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char** topic)
goto exit;
}
}
if ((msgid = MQTTProtocol_assignMsgId(m->c)) == 0)
{
rc = MQTTCLIENT_MAX_MESSAGES_INFLIGHT;
goto exit;
}
for (i = 0; i < count; i++)
ListAppend(topics, topic[i], strlen(topic[i]));
rc = MQTTProtocol_unsubscribe(m->c, topics);
rc = MQTTProtocol_unsubscribe(m->c, topics, msgid);
ListFreeNoContent(topics);
if (rc == TCPSOCKET_COMPLETE)
......@@ -1382,18 +1376,18 @@ exit:
}
int MQTTClient_unsubscribe(MQTTClient handle, char* topic)
int MQTTClient_unsubscribe(MQTTClient handle, const char* topic)
{
int rc = 0;
char *const topics[] = {(char*)topic};
FUNC_ENTRY;
rc = MQTTClient_unsubscribeMany(handle, 1, &topic);
rc = MQTTClient_unsubscribeMany(handle, 1, topics);
FUNC_EXIT_RC(rc);
return rc;
}
int MQTTClient_publish(MQTTClient handle, char* topicName, int payloadlen, void* payload,
int MQTTClient_publish(MQTTClient handle, const char* topicName, int payloadlen, void* payload,
int qos, int retained, MQTTClient_deliveryToken* deliveryToken)
{
int rc = MQTTCLIENT_SUCCESS;
......@@ -1401,6 +1395,7 @@ int MQTTClient_publish(MQTTClient handle, char* topicName, int payloadlen, void*
Messages* msg = NULL;
Publish* p = NULL;
int blocked = 0;
int msgid = 0;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
......@@ -1434,13 +1429,18 @@ int MQTTClient_publish(MQTTClient handle, char* topicName, int payloadlen, void*
}
if (blocked == 1)
Log(TRACE_MIN, -1, "Resuming publish now queue not full for client %s", m->c->clientID);
if (qos > 0 && (msgid = MQTTProtocol_assignMsgId(m->c)) == 0)
{ /* this should never happen as we've waited for spaces in the queue */
rc = MQTTCLIENT_MAX_MESSAGES_INFLIGHT;
goto exit;
}
p = malloc(sizeof(Publish));
p->payload = payload;
p->payloadlen = payloadlen;
p->topic = topicName;
p->msgId = -1;
p->topic = (char*)topicName;
p->msgId = msgid;
rc = MQTTProtocol_startPublish(m->c, p, qos, retained, &msg);
......@@ -1482,7 +1482,7 @@ exit:
int MQTTClient_publishMessage(MQTTClient handle, char* topicName, MQTTClient_message* message,
int MQTTClient_publishMessage(MQTTClient handle, const char* topicName, MQTTClient_message* message,
MQTTClient_deliveryToken* deliveryToken)
{
int rc = MQTTCLIENT_SUCCESS;
......
src/MQTTClient.h 100644 → 100755
View file @ e5a1a084
......@@ -97,6 +97,9 @@
*/
/// @cond EXCLUDE
#if defined(__cplusplus)
extern "C" {
#endif
#if !defined(MQTTCLIENT_H)
#define MQTTCLIENT_H
......@@ -401,7 +404,7 @@ DLLExport int MQTTClient_setCallbacks(MQTTClient handle, void* context, MQTTClie
* @return ::MQTTCLIENT_SUCCESS if the client is successfully created, otherwise
* an error code is returned.
*/
DLLExport int MQTTClient_create(MQTTClient* handle, char* serverURI, char* clientId,
DLLExport int MQTTClient_create(MQTTClient* handle, const char* serverURI, const char* clientId,
int persistence_type, void* persistence_context);
/**
......@@ -419,13 +422,13 @@ DLLExport int MQTTClient_create(MQTTClient* handle, char* serverURI, char* clien
typedef struct
{
/** The eyecatcher for this structure. must be MQTW. */
char struct_id[4];
const char struct_id[4];
/** The version number of this structure. Must be 0 */
int struct_version;
/** The LWT topic to which the LWT message will be published. */
char* topicName;
const char* topicName;
/** The LWT payload. */
char* message;
const char* message;
/**
* The retained flag for the LWT message (see MQTTClient_message.retained).
*/
......@@ -454,7 +457,7 @@ typedef struct
typedef struct
{
/** The eyecatcher for this structure. Must be MQTS */
char struct_id[4];
const char struct_id[4];
/** The version number of this structure. Must be 0 */
int struct_version;
......@@ -507,7 +510,7 @@ typedef struct
typedef struct
{
/** The eyecatcher for this structure. must be MQTC. */
char struct_id[4];
const char struct_id[4];
/** The version number of this structure. Must be 0, 1, 2, 3 or 4.
* 0 signifies no SSL options and no serverURIs
* 1 signifies no serverURIs
......@@ -568,13 +571,13 @@ typedef struct
* and authorisation by user name and password. This is the user name
* parameter.
*/
char* username;
const char* username;
/**
* MQTT servers that support the MQTT v3.1 protocol provide authentication
* and authorisation by user name and password. This is the password
* parameter.
*/
char* password;
const char* password;
/**
* The time interval in seconds to allow a connect to complete.
*/
......@@ -602,7 +605,7 @@ typedef struct
* If this list is empty (the default), the server URI specified on MQTTClient_create()
* is used.
*/
char** serverURIs;
char* const* serverURIs;
/**
* Sets the version of MQTT to be used on the connect.
* MQTTVERSION_DEFAULT (0) = default: start with 3.1.1, and if that fails, fall back to 3.1
......@@ -615,7 +618,7 @@ typedef struct
*/
struct
{
char* serverURI; /**< the serverURI connected to */
const char* serverURI; /**< the serverURI connected to */
int MQTTVersion; /**< the MQTT version used to connect with */
int sessionPresent; /**< if the MQTT version is 3.1.1, the value of sessionPresent returned in the connack */
} returned;
......@@ -711,7 +714,7 @@ DLLExport int MQTTClient_isConnected(MQTTClient handle);
* An error code is returned if there was a problem registering the
* subscription.
*/
DLLExport int MQTTClient_subscribe(MQTTClient handle, char* topic, int qos);
DLLExport int MQTTClient_subscribe(MQTTClient handle, const char* topic, int qos);
/**
* This function attempts to subscribe a client to a list of topics, which may
......@@ -729,7 +732,7 @@ DLLExport int MQTTClient_subscribe(MQTTClient handle, char* topic, int qos);
* An error code is returned if there was a problem registering the
* subscriptions.
*/
DLLExport int MQTTClient_subscribeMany(MQTTClient handle, int count, char** topic, int* qos);
DLLExport int MQTTClient_subscribeMany(MQTTClient handle, int count, char* const* topic, int* qos);
/**
* This function attempts to remove an existing subscription made by the
......@@ -742,7 +745,7 @@ DLLExport int MQTTClient_subscribeMany(MQTTClient handle, int count, char** topi
* An error code is returned if there was a problem removing the
* subscription.
*/
DLLExport int MQTTClient_unsubscribe(MQTTClient handle, char* topic);
DLLExport int MQTTClient_unsubscribe(MQTTClient handle, const char* topic);
/**
* This function attempts to remove existing subscriptions to a list of topics
......@@ -755,7 +758,7 @@ DLLExport int MQTTClient_unsubscribe(MQTTClient handle, char* topic);
* @return ::MQTTCLIENT_SUCCESS if the subscriptions are removed.
* An error code is returned if there was a problem removing the subscriptions.
*/
DLLExport int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char** topic);
DLLExport int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char* const* topic);
/**
* This function attempts to publish a message to a given topic (see also
......@@ -778,7 +781,7 @@ DLLExport int MQTTClient_unsubscribeMany(MQTTClient handle, int count, char** to
* @return ::MQTTCLIENT_SUCCESS if the message is accepted for publication.
* An error code is returned if there was a problem accepting the message.
*/
DLLExport int MQTTClient_publish(MQTTClient handle, char* topicName, int payloadlen, void* payload, int qos, int retained,
DLLExport int MQTTClient_publish(MQTTClient handle, const char* topicName, int payloadlen, void* payload, int qos, int retained,
MQTTClient_deliveryToken* dt);
/**
* This function attempts to publish a message to a given topic (see also
......@@ -799,7 +802,7 @@ DLLExport int MQTTClient_publish(MQTTClient handle, char* topicName, int payload
* @return ::MQTTCLIENT_SUCCESS if the message is accepted for publication.
* An error code is returned if there was a problem accepting the message.
*/
DLLExport int MQTTClient_publishMessage(MQTTClient handle, char* topicName, MQTTClient_message* msg, MQTTClient_deliveryToken* dt);
DLLExport int MQTTClient_publishMessage(MQTTClient handle, const char* topicName, MQTTClient_message* msg, MQTTClient_deliveryToken* dt);
/**
......@@ -914,6 +917,9 @@ DLLExport void MQTTClient_free(void* ptr);
DLLExport void MQTTClient_destroy(MQTTClient* handle);
#endif
#ifdef __cplusplus
}
#endif
/**
* @cond MQTTClient_main
......
src/MQTTClientPersistence.h 100644 → 100755
View file @ e5a1a084
......@@ -106,7 +106,7 @@
* @return Return 0 if the function completes successfully, otherwise return
* ::MQTTCLIENT_PERSISTENCE_ERROR.
*/
typedef int (*Persistence_open)(void** handle, char* clientID, char* serverURI, void* context);
typedef int (*Persistence_open)(void** handle, const char* clientID, const char* serverURI, void* context);
/**
* @brief Close the persistent store referred to by the handle.
......
src/MQTTPacket.c 100644 → 100755
View file @ e5a1a084
......@@ -97,6 +97,7 @@ void* MQTTPacket_Factory(networkHandles* net, int* error)
char* data = NULL;
static Header header;
int remaining_length, ptype;
size_t remaining_length_new;
void* pack = NULL;
int actual_len = 0;
......@@ -147,8 +148,9 @@ void* MQTTPacket_Factory(networkHandles* net, int* error)
char *buf = malloc(10);
buf[0] = header.byte;
buf0len = 1 + MQTTPacket_encode(&buf[1], remaining_length);
remaining_length_new = remaining_length;
*error = MQTTPersistence_put(net->socket, buf, buf0len, 1,
&data, &remaining_length, header.bits.type, ((Publish *)pack)->msgId, 1);
&data, &remaining_length_new, header.bits.type, ((Publish *)pack)->msgId, 1);
free(buf);
}
#endif
......@@ -168,7 +170,7 @@ exit:
* @param buflen the length of the data in buffer to be written
* @return the completion code (TCPSOCKET_COMPLETE etc)
*/
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, int buflen, int free)
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buflen, int free)
{
int rc, buf0len;
char *buf;
......@@ -214,7 +216,7 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, int buflen
* @param buflens the lengths of the data in the array of buffers to be written
* @return the completion code (TCPSOCKET_COMPLETE etc)
*/
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, int* buflens, int* frees)
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, size_t* buflens, int* frees)
{
int i, rc, buf0len, total = 0;
char *buf;
......@@ -423,7 +425,7 @@ void writeInt(char** pptr, int anInt)
* @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
* @param string the C string to write
*/
void writeUTF(char** pptr, char* string)
void writeUTF(char** pptr, const char* string)
{
int len = strlen(string);
writeInt(pptr, len);
......@@ -439,7 +441,7 @@ void writeUTF(char** pptr, char* string)
* @param datalen the length of the rest of the packet
* @return pointer to the packet structure
*/
void* MQTTPacket_header_only(unsigned char aHeader, char* data, int datalen)
void* MQTTPacket_header_only(unsigned char aHeader, char* data, size_t datalen)
{
static unsigned char header = 0;
header = aHeader;
......@@ -452,7 +454,7 @@ void* MQTTPacket_header_only(unsigned char aHeader, char* data, int datalen)
* @param socket the open socket to send the data to
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_disconnect(networkHandles *net, char* clientID)
int MQTTPacket_send_disconnect(networkHandles *net, const char* clientID)
{
Header header;
int rc = 0;
......@@ -474,7 +476,7 @@ int MQTTPacket_send_disconnect(networkHandles *net, char* clientID)
* @param datalen the length of the rest of the packet
* @return pointer to the packet structure
*/
void* MQTTPacket_publish(unsigned char aHeader, char* data, int datalen)
void* MQTTPacket_publish(unsigned char aHeader, char* data, size_t datalen)
{
Publish* pack = malloc(sizeof(Publish));
char* curdata = data;
......@@ -550,7 +552,7 @@ int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_puback(int msgid, networkHandles* net, char* clientID)
int MQTTPacket_send_puback(int msgid, networkHandles* net, const char* clientID)
{
int rc = 0;
......@@ -583,7 +585,7 @@ void MQTTPacket_freeSuback(Suback* pack)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pubrec(int msgid, networkHandles* net, char* clientID)
int MQTTPacket_send_pubrec(int msgid, networkHandles* net, const char* clientID)
{
int rc = 0;
......@@ -603,7 +605,7 @@ int MQTTPacket_send_pubrec(int msgid, networkHandles* net, char* clientID)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, char* clientID)
int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, const char* clientID)
{
int rc = 0;
......@@ -622,7 +624,7 @@ int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, char* client
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, char* clientID)
int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, const char* clientID)
{
int rc = 0;
......@@ -641,7 +643,7 @@ int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, char* clientID)
* @param datalen the length of the rest of the packet
* @return pointer to the packet structure
*/
void* MQTTPacket_ack(unsigned char aHeader, char* data, int datalen)
void* MQTTPacket_ack(unsigned char aHeader, char* data, size_t datalen)
{
Ack* pack = malloc(sizeof(Ack));
char* curdata = data;
......@@ -664,7 +666,7 @@ void* MQTTPacket_ack(unsigned char aHeader, char* data, int datalen)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, networkHandles* net, char* clientID)
int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, networkHandles* net, const char* clientID)
{
Header header;
char *topiclen;
......@@ -682,7 +684,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo
char *buf = malloc(2);
char *ptr = buf;
char* bufs[4] = {topiclen, pack->topic, buf, pack->payload};
int lens[4] = {2, strlen(pack->topic), 2, pack->payloadlen};
size_t lens[4] = {2, strlen(pack->topic), 2, pack->payloadlen};
int frees[4] = {1, 0, 1, 0};
writeInt(&ptr, pack->msgId);
......@@ -696,7 +698,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo
{
char* ptr = topiclen;
char* bufs[3] = {topiclen, pack->topic, pack->payload};
int lens[3] = {2, strlen(pack->topic), pack->payloadlen};
size_t lens[3] = {2, strlen(pack->topic), pack->payloadlen};
int frees[3] = {1, 0, 0};
writeInt(&ptr, lens[1]);
......
src/MQTTPacket.h 100644 → 100755
View file @ e5a1a084
......@@ -33,7 +33,7 @@ include "Clients"
BE*/
typedef unsigned int bool;
typedef void* (*pf)(unsigned char, char*, int);
typedef void* (*pf)(unsigned char, char*, size_t);
#define BAD_MQTT_PACKET -4
......@@ -223,27 +223,27 @@ char* readUTF(char** pptr, char* enddata);
unsigned char readChar(char** pptr);
void writeChar(char** pptr, char c);
void writeInt(char** pptr, int anInt);
void writeUTF(char** pptr, char* string);
void writeUTF(char** pptr, const char* string);
char* MQTTPacket_name(int ptype);
void* MQTTPacket_Factory(networkHandles* net, int* error);
int MQTTPacket_send(networkHandles* net, Header header, char* buffer, int buflen, int free);
int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffers, int* buflens, int* frees);
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);
void* MQTTPacket_header_only(unsigned char aHeader, char* data, int datalen);
int MQTTPacket_send_disconnect(networkHandles* net, char* clientID);
void* MQTTPacket_header_only(unsigned char aHeader, char* data, size_t datalen);
int MQTTPacket_send_disconnect(networkHandles* net, const char* clientID);
void* MQTTPacket_publish(unsigned char aHeader, char* data, int datalen);
void* MQTTPacket_publish(unsigned char aHeader, char* data, size_t datalen);
void MQTTPacket_freePublish(Publish* pack);
int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, networkHandles* net, char* clientID);
int MQTTPacket_send_puback(int msgid, networkHandles* net, char* clientID);
void* MQTTPacket_ack(unsigned char aHeader, char* data, int datalen);
int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, networkHandles* net, const char* clientID);
int MQTTPacket_send_puback(int msgid, networkHandles* net, const char* clientID);
void* MQTTPacket_ack(unsigned char aHeader, char* data, size_t datalen);
void MQTTPacket_freeSuback(Suback* pack);
int MQTTPacket_send_pubrec(int msgid, networkHandles* net, char* clientID);
int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, char* clientID);
int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, char* clientID);
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);
void MQTTPacket_free_packet(MQTTPacket* pack);
......
src/MQTTPacketOut.c 100644 → 100755
View file @ e5a1a084
......@@ -14,6 +14,7 @@
* Ian Craggs - initial API and implementation and/or initial documentation
* Ian Craggs, Allan Stockdill-Mander - SSL updates
* Ian Craggs - MQTT 3.1.1 support
* Rong Xiang, Ian Craggs - C++ compatibility
*******************************************************************************/
/**
......@@ -116,7 +117,7 @@ exit:
* @param datalen the length of the rest of the packet
* @return pointer to the packet structure
*/
void* MQTTPacket_connack(unsigned char aHeader, char* data, int datalen)
void* MQTTPacket_connack(unsigned char aHeader, char* data, size_t datalen)
{
Connack* pack = malloc(sizeof(Connack));
char* curdata = data;
......@@ -136,15 +137,16 @@ void* MQTTPacket_connack(unsigned char aHeader, char* data, int datalen)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pingreq(networkHandles* net, char* clientID)
int MQTTPacket_send_pingreq(networkHandles* net, const char* clientID)
{
Header header;
int rc = 0;
size_t buflen = 0;
FUNC_ENTRY;
header.byte = 0;
header.bits.type = PINGREQ;
rc = MQTTPacket_send(net, header, NULL, 0, 0);
rc = MQTTPacket_send(net, header, NULL, buflen,0);
Log(LOG_PROTOCOL, 20, NULL, net->socket, clientID, rc);
FUNC_EXIT_RC(rc);
return rc;
......@@ -161,7 +163,7 @@ int MQTTPacket_send_pingreq(networkHandles* net, char* clientID)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_subscribe(List* topics, List* qoss, int msgid, int dup, networkHandles* net, char* clientID)
int MQTTPacket_send_subscribe(List* topics, List* qoss, int msgid, int dup, networkHandles* net, const char* clientID)
{
Header header;
char *data, *ptr;
......@@ -204,7 +206,7 @@ int MQTTPacket_send_subscribe(List* topics, List* qoss, int msgid, int dup, netw
* @param datalen the length of the rest of the packet
* @return pointer to the packet structure
*/
void* MQTTPacket_suback(unsigned char aHeader, char* data, int datalen)
void* MQTTPacket_suback(unsigned char aHeader, char* data, size_t datalen)
{
Suback* pack = malloc(sizeof(Suback));
char* curdata = data;
......@@ -213,7 +215,7 @@ void* MQTTPacket_suback(unsigned char aHeader, char* data, int datalen)
pack->header.byte = aHeader;
pack->msgId = readInt(&curdata);
pack->qoss = ListInitialize();
while (curdata - data < datalen)
while ((size_t)(curdata - data) < datalen)
{
int* newint;
newint = malloc(sizeof(int));
......@@ -234,7 +236,7 @@ void* MQTTPacket_suback(unsigned char aHeader, char* data, int datalen)
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_unsubscribe(List* topics, int msgid, int dup, networkHandles* net, char* clientID)
int MQTTPacket_send_unsubscribe(List* topics, int msgid, int dup, networkHandles* net, const char* clientID)
{
Header header;
char *data, *ptr;
......
src/MQTTPacketOut.h 100644 → 100755
View file @ e5a1a084
......@@ -22,13 +22,13 @@
#include "MQTTPacket.h"
int MQTTPacket_send_connect(Clients* client, int MQTTVersion);
void* MQTTPacket_connack(unsigned char aHeader, char* data, int datalen);
void* MQTTPacket_connack(unsigned char aHeader, char* data, size_t datalen);
int MQTTPacket_send_pingreq(networkHandles* net, char* clientID);
int MQTTPacket_send_pingreq(networkHandles* net, const char* clientID);
int MQTTPacket_send_subscribe(List* topics, List* qoss, int msgid, int dup, networkHandles* net, char* clientID);
void* MQTTPacket_suback(unsigned char aHeader, char* data, int datalen);
int MQTTPacket_send_subscribe(List* topics, List* qoss, int msgid, int dup, networkHandles* net, const char* clientID);
void* MQTTPacket_suback(unsigned char aHeader, char* data, size_t datalen);
int MQTTPacket_send_unsubscribe(List* topics, int msgid, int dup, networkHandles* net, char* clientID);
int MQTTPacket_send_unsubscribe(List* topics, int msgid, int dup, networkHandles* net, const char* clientID);
#endif
src/MQTTPersistence.c 100644 → 100755
View file @ e5a1a084
......@@ -102,7 +102,7 @@ int MQTTPersistence_create(MQTTClient_persistence** persistence, int type, void*
* @param serverURI the URI of the remote end.
* @return 0 if success, #MQTTCLIENT_PERSISTENCE_ERROR otherwise.
*/
int MQTTPersistence_initialize(Clients *c, char *serverURI)
int MQTTPersistence_initialize(Clients *c, const char *serverURI)
{
int rc = 0;
......@@ -264,7 +264,7 @@ int MQTTPersistence_restore(Clients *c)
* @param buffer the persisted data.
* @param buflen the number of bytes of the data buffer.
*/
void* MQTTPersistence_restorePacket(char* buffer, int buflen)
void* MQTTPersistence_restorePacket(char* buffer, size_t buflen)
{
void* pack = NULL;
Header header;
......@@ -302,7 +302,7 @@ void* MQTTPersistence_restorePacket(char* buffer, int buflen)
* @param content the message to add.
* @param size size of the message.
*/
void MQTTPersistence_insertInOrder(List* list, void* content, int size)
void MQTTPersistence_insertInOrder(List* list, void* content, size_t size)
{
ListElement* index = NULL;
ListElement* current = NULL;
......@@ -333,8 +333,8 @@ void MQTTPersistence_insertInOrder(List* list, void* content, int size)
* receiving direction.
* @return 0 if success, #MQTTCLIENT_PERSISTENCE_ERROR otherwise.
*/
int MQTTPersistence_put(int socket, char* buf0, int buf0len, int count,
char** buffers, int* buflens, int htype, int msgId, int scr )
int MQTTPersistence_put(int socket, char* buf0, size_t buf0len, int count,
char** buffers, size_t* buflens, int htype, int msgId, int scr )
{
int rc = 0;
extern ClientStates* bstate;
......@@ -531,7 +531,7 @@ int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry*
}
MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, int buflen)
MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, size_t buflen)
{
MQTTPersistence_qEntry* qe = NULL;
char* ptr = buffer;
......@@ -577,7 +577,7 @@ MQTTPersistence_qEntry* MQTTPersistence_restoreQueueEntry(char* buffer, int bufl
}
void MQTTPersistence_insertInSeqOrder(List* list, MQTTPersistence_qEntry* qEntry, int size)
void MQTTPersistence_insertInSeqOrder(List* list, MQTTPersistence_qEntry* qEntry, size_t size)
{
ListElement* index = NULL;
ListElement* current = NULL;
......
src/MQTTPersistence.h 100644 → 100755
View file @ e5a1a084
......@@ -16,6 +16,10 @@
* Ian Craggs - fix for bug 432903 - queue persistence
*******************************************************************************/
#if defined(__cplusplus)
extern "C" {
#endif
#include "Clients.h"
/** Stem of the key for a sent PUBLISH QoS1 or QoS2 */
......@@ -31,14 +35,14 @@
#define PERSISTENCE_MAX_KEY_LENGTH 8
int MQTTPersistence_create(MQTTClient_persistence** per, int type, void* pcontext);
int MQTTPersistence_initialize(Clients* c, char* serverURI);
int MQTTPersistence_initialize(Clients* c, const char* serverURI);
int MQTTPersistence_close(Clients* c);
int MQTTPersistence_clear(Clients* c);
int MQTTPersistence_restore(Clients* c);
void* MQTTPersistence_restorePacket(char* buffer, int buflen);
void MQTTPersistence_insertInOrder(List* list, void* content, int size);
int MQTTPersistence_put(int socket, char* buf0, int buf0len, int count,
char** buffers, int* buflens, int htype, int msgId, int scr);
void* MQTTPersistence_restorePacket(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);
int MQTTPersistence_remove(Clients* c, char* type, int qos, int msgId);
void MQTTPersistence_wrapMsgID(Clients *c);
......@@ -65,3 +69,6 @@ typedef struct
int MQTTPersistence_unpersistQueueEntry(Clients* client, MQTTPersistence_qEntry* qe);
int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry* qe);
int MQTTPersistence_restoreMessageQueue(Clients* c);
#ifdef __cplusplus
}
#endif
src/MQTTPersistenceDefault.c 100644 → 100755
View file @ e5a1a084
......@@ -58,7 +58,7 @@
* See ::Persistence_open
*/
int pstopen(void **handle, char* clientID, char* serverURI, void* context)
int pstopen(void **handle, const char* clientID, const char* serverURI, void* context)
{
int rc = 0;
char *dataDir = context;
......
src/MQTTPersistenceDefault.h 100644 → 100755
View file @ e5a1a084
......@@ -20,7 +20,7 @@
#define MESSAGE_FILENAME_EXTENSION ".msg"
/* prototypes of the functions for the default file system persistence */
int pstopen(void** handle, char* clientID, char* serverURI, void* context);
int pstopen(void** handle, const char* clientID, const char* serverURI, void* context);
int pstclose(void* handle);
int pstput(void* handle, char* key, int bufcount, char* buffers[], int buflens[]);
int pstget(void* handle, char* key, char** buffer, int* buflen);
......
src/MQTTProtocolClient.c 100644 → 100755
View file @ e5a1a084
......@@ -15,6 +15,7 @@
* Ian Craggs, Allan Stockdill-Mander - SSL updates
* Ian Craggs - fix for bug 413429 - connectionLost not called
* Ian Craggs - fix for bug 421103 - trying to write to same socket, in retry
* Rong Xiang, Ian Craggs - C++ compatibility
*******************************************************************************/
/**
......@@ -145,7 +146,6 @@ int MQTTProtocol_startPublish(Clients* pubclient, Publish* publish, int qos, int
FUNC_ENTRY;
if (qos > 0)
{
p.msgId = publish->msgId = MQTTProtocol_assignMsgId(pubclient);
*mm = MQTTProtocol_createMessage(publish, mm, qos, retained);
ListAppend(pubclient->outboundMsgs, *mm, (*mm)->len);
/* we change these pointers to the saved message location just in case the packet could not be written
......@@ -714,3 +714,44 @@ void MQTTProtocol_freeMessageList(List* msgList)
FUNC_EXIT;
}
/**
* Copy no more than dest_size -1 characters from the string pointed to by src to the array pointed to by dest.
* The destination string will always be null-terminated.
* @param dest the array which characters copy to
* @param src the source string which characters copy from
* @param dest_size the size of the memory pointed to by dest: copy no more than this -1 (allow for null). Must be >= 1
* @return the destination string pointer
*/
char* MQTTStrncpy(char *dest, const char *src, size_t dest_size)
{
size_t count = dest_size;
char *temp = dest;
FUNC_ENTRY;
if (dest_size < strlen(src))
Log(TRACE_MIN, -1, "the src string is truncated");
/* We must copy only the first (dest_size - 1) bytes */
while (count > 1 && (*temp++ = *src++))
count--;
*temp = '\0';
FUNC_EXIT;
return dest;
}
/**
* Duplicate a string, safely, allocating space on the heap
* @param src the source string which characters copy from
* @return the duplicated, allocated string
*/
char* MQTTStrdup(const char* src)
{
size_t mlen = strlen(src) + 1;
char* temp = malloc(mlen);
MQTTStrncpy(temp, src, mlen);
return temp;
}
src/MQTTProtocolClient.h 100644 → 100755
View file @ e5a1a084
......@@ -14,6 +14,7 @@
* Ian Craggs - initial API and implementation and/or initial documentation
* Ian Craggs, Allan Stockdill-Mander - SSL updates
* Ian Craggs - MQTT 3.1.1 updates
* Rong Xiang, Ian Craggs - C++ compatibility
*******************************************************************************/
#if !defined(MQTTPROTOCOLCLIENT_H)
......@@ -28,7 +29,6 @@
#define MAX_MSG_ID 65535
#define MAX_CLIENTID_LEN 65535
int MQTTProtocol_assignMsgId(Clients* client);
int MQTTProtocol_startPublish(Clients* pubclient, Publish* publish, int qos, int retained, Messages** m);
Messages* MQTTProtocol_createMessage(Publish* publish, Messages** mm, int qos, int retained);
Publications* MQTTProtocol_storePublication(Publish* publish, int* len);
......@@ -48,4 +48,6 @@ void MQTTProtocol_freeClient(Clients* client);
void MQTTProtocol_emptyMessageList(List* msgList);
void MQTTProtocol_freeMessageList(List* msgList);
char* MQTTStrncpy(char *dest, const char* src, size_t num);
char* MQTTStrdup(const char* src);
#endif
src/MQTTProtocolOut.c 100644 → 100755
View file @ e5a1a084
......@@ -15,6 +15,7 @@
* Ian Craggs, Allan Stockdill-Mander - SSL updates
* Ian Craggs - fix for buffer overflow in addressPort bug #433290
* Ian Craggs - MQTT 3.1.1 support
* Rong Xiang, Ian Craggs - C++ compatibility
*******************************************************************************/
/**
......@@ -40,10 +41,10 @@ extern ClientStates* bstate;
* @param port the returned port integer
* @return the address string
*/
char* MQTTProtocol_addressPort(char* uri, int* port)
char* MQTTProtocol_addressPort(const char* uri, int* port)
{
char* colon_pos = strrchr(uri, ':'); /* reverse find to allow for ':' in IPv6 addresses */
char* buf = uri;
char* buf = (char*)uri;
int len;
FUNC_ENTRY;
......@@ -58,8 +59,7 @@ char* MQTTProtocol_addressPort(char* uri, int* port)
int addr_len = colon_pos - uri;
buf = malloc(addr_len + 1);
*port = atoi(colon_pos + 1);
strncpy(buf, uri, addr_len);
buf[addr_len] = '\0';
MQTTStrncpy(buf, uri, addr_len+1);
}
else
*port = DEFAULT_PORT;
......@@ -82,9 +82,9 @@ char* MQTTProtocol_addressPort(char* uri, int* port)
* @return return code
*/
#if defined(OPENSSL)
int MQTTProtocol_connect(char* ip_address, Clients* aClient, int ssl, int MQTTVersion)
int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int ssl, int MQTTVersion)
#else
int MQTTProtocol_connect(char* ip_address, Clients* aClient, int MQTTVersion)
int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int MQTTVersion)
#endif
{
int rc, port;
......@@ -157,13 +157,13 @@ int MQTTProtocol_handlePingresps(void* pack, int sock)
* @param qoss corresponding list of QoSs
* @return completion code
*/
int MQTTProtocol_subscribe(Clients* client, List* topics, List* qoss)
int MQTTProtocol_subscribe(Clients* client, List* topics, List* qoss, int msgID)
{
int rc = 0;
FUNC_ENTRY;
/* we should stack this up for retry processing too */
rc = MQTTPacket_send_subscribe(topics, qoss, MQTTProtocol_assignMsgId(client), 0, &client->net, client->clientID);
rc = MQTTPacket_send_subscribe(topics, qoss, msgID, 0, &client->net, client->clientID);
FUNC_EXIT_RC(rc);
return rc;
}
......@@ -196,13 +196,13 @@ int MQTTProtocol_handleSubacks(void* pack, int sock)
* @param topics list of topics
* @return completion code
*/
int MQTTProtocol_unsubscribe(Clients* client, List* topics)
int MQTTProtocol_unsubscribe(Clients* client, List* topics, int msgID)
{
int rc = 0;
FUNC_ENTRY;
/* we should stack this up for retry processing too? */
rc = MQTTPacket_send_unsubscribe(topics, MQTTProtocol_assignMsgId(client), 0, &client->net, client->clientID);
rc = MQTTPacket_send_unsubscribe(topics, msgID, 0, &client->net, client->clientID);
FUNC_EXIT_RC(rc);
return rc;
}
......
src/MQTTProtocolOut.h 100644 → 100755
View file @ e5a1a084
......@@ -29,16 +29,16 @@
#define DEFAULT_PORT 1883
void MQTTProtocol_reconnect(char* ip_address, Clients* client);
void MQTTProtocol_reconnect(const char* ip_address, Clients* client);
#if defined(OPENSSL)
int MQTTProtocol_connect(char* ip_address, Clients* acClients, int ssl, int MQTTVersion);
int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int ssl, int MQTTVersion);
#else
int MQTTProtocol_connect(char* ip_address, Clients* acClients, int MQTTVersion);
int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int MQTTVersion);
#endif
int MQTTProtocol_handlePingresps(void* pack, int sock);
int MQTTProtocol_subscribe(Clients* client, List* topics, List* qoss);
int MQTTProtocol_subscribe(Clients* client, List* topics, List* qoss, int msgID);
int MQTTProtocol_handleSubacks(void* pack, int sock);
int MQTTProtocol_unsubscribe(Clients* client, List* topics);
int MQTTProtocol_unsubscribe(Clients* client, List* topics, int msgID);
int MQTTProtocol_handleUnsubacks(void* pack, int sock);
#endif
src/SSLSocket.c 100644 → 100755
View file @ e5a1a084
......@@ -703,7 +703,7 @@ int SSLSocket_close(networkHandles* net)
/* No SSL_writev() provided by OpenSSL. Boo. */
int SSLSocket_putdatas(SSL* ssl, int socket, char* buf0, int buf0len, int count, char** buffers, int* buflens, int* frees)
int SSLSocket_putdatas(SSL* ssl, int socket, char* buf0, size_t buf0len, int count, char** buffers, size_t* buflens, int* frees)
{
int rc = 0;
int i;
......
src/SSLSocket.h 100644 → 100755
View file @ e5a1a084
......@@ -37,7 +37,7 @@ int SSLSocket_getch(SSL* ssl, int socket, char* c);
char *SSLSocket_getdata(SSL* ssl, int socket, int bytes, int* actual_len);
int SSLSocket_close(networkHandles* net);
int SSLSocket_putdatas(SSL* ssl, int socket, char* buf0, int buf0len, int count, char** buffers, int* buflens, int* frees);
int SSLSocket_putdatas(SSL* ssl, int socket, char* buf0, size_t buf0len, int count, char** buffers, size_t* buflens, int* frees);
int SSLSocket_connect(SSL* ssl, int socket);
int SSLSocket_getPendingRead();
......
src/Socket.c 100644 → 100755
View file @ e5a1a084
......@@ -433,7 +433,7 @@ int Socket_writev(int socket, iobuf* iovecs, int count, unsigned long* bytes)
* @param buflens an array of corresponding buffer lengths
* @return completion code, especially TCPSOCKET_INTERRUPTED
*/
int Socket_putdatas(int socket, char* buf0, int buf0len, int count, char** buffers, int* buflens, int* frees)
int Socket_putdatas(int socket, char* buf0, size_t buf0len, int count, char** buffers, size_t* buflens, int* frees)
{
unsigned long bytes = 0L;
iobuf iovecs[5];
......
src/Socket.h 100644 → 100755
View file @ e5a1a084
......@@ -114,7 +114,7 @@ void Socket_outTerminate(void);
int Socket_getReadySocket(int more_work, struct timeval *tp);
int Socket_getch(int socket, char* c);
char *Socket_getdata(int socket, int bytes, int* actual_len);
int Socket_putdatas(int socket, char* buf0, int buf0len, int count, char** buffers, int* buflens, int* frees);
int Socket_putdatas(int socket, char* buf0, size_t buf0len, int count, char** buffers, size_t* buflens, int* frees);
void Socket_close(int socket);
int Socket_new(char* addr, int port, int* socket);
......
src/utf-8.c 100644 → 100755
View file @ e5a1a084
......@@ -69,12 +69,12 @@ valid_ranges[] =
* @param data the bytes to check for a valid UTF-8 char
* @return pointer to the start of the next UTF-8 character in "data"
*/
char* UTF8_char_validate(int len, char* data)
const char* UTF8_char_validate(int len, const char* data)
{
int good = 0;
int charlen = 2;
int i, j;
char *rc = NULL;
const char *rc = NULL;
FUNC_ENTRY;
/* first work out how many bytes this char is encoded in */
......@@ -121,9 +121,9 @@ char* UTF8_char_validate(int len, char* data)
* @param data the bytes to check for valid UTF-8 characters
* @return 1 (true) if the string has only UTF-8 characters, 0 (false) otherwise
*/
int UTF8_validate(int len, char* data)
int UTF8_validate(int len, const char* data)
{
char* curdata = NULL;
const char* curdata = NULL;
int rc = 0;
FUNC_ENTRY;
......@@ -148,7 +148,7 @@ exit:
* @param string the string to check for valid UTF-8 characters
* @return 1 (true) if the string has only UTF-8 characters, 0 (false) otherwise
*/
int UTF8_validateString(char* string)
int UTF8_validateString(const char* string)
{
int rc = 0;
......
src/utf-8.h 100644 → 100755
View file @ e5a1a084
......@@ -18,6 +18,6 @@
#define UTF8_H
int UTF8_validate(int len, char* data);
int UTF8_validateString(char* string);
int UTF8_validateString(const char* string);
#endif
test/ssl/gen.sh
View file @ e5a1a084
......@@ -64,6 +64,8 @@ openssl ca -config openssl.cnf -name CA_signing -revoke client-revoked.crt
openssl ca -config openssl.cnf -name CA_signing -gencrl -out crl.pem
cat test-signing-ca.crt test-root-ca.crt > all-ca.crt
cat client.crt client.key all-ca.crt > client.pem
#mkdir certs
#cp test-signing-ca.crt certs/test-signing-ca.pem
#cp test-root-ca.crt certs/test-root.ca.pem
......
test/test1.c
View file @ e5a1a084
......@@ -1095,12 +1095,16 @@ int test6(struct Options options)
MyLog(LOGA_INFO, "Waiting to receive the will message");
count = 0;
while (test6_will_message_arrived != 1 && test6_connection_lost_called != 1 && ++count < 40)
while (++count < 40)
{
#if defined(WIN32)
Sleep(1000L);
#else
sleep(1);
#endif
if (test6_will_message_arrived == 1 && test6_connection_lost_called == 1)
break;
}
assert("will message arrived", test6_will_message_arrived == 1,
"will_message_arrived was %d\n", test6_will_message_arrived);
assert("connection lost called", test6_connection_lost_called == 1,
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment