From cd2d10b6507cd562dc6d5042968a970868531442 Mon Sep 17 00:00:00 2001 From: fpagliughi Date: Sat, 21 Dec 2024 20:03:02 -0500 Subject: [PATCH] Added support for UNIX-domain sockets for v1.3.x --- CMakeLists.txt | 8 +++ README.md | 24 ++++++++- src/MQTTAsync.c | 10 ++++ src/MQTTAsyncUtils.c | 15 ++++-- src/MQTTAsyncUtils.h | 2 + src/MQTTClient.c | 27 ++++++++-- src/MQTTProtocolOut.c | 50 +++++++++++------- src/MQTTProtocolOut.h | 8 +-- src/Socket.c | 71 +++++++++++++++++++++++--- src/Socket.h | 1 + src/samples/MQTTAsync_publish.c | 5 +- src/samples/MQTTAsync_subscribe.c | 5 +- src/samples/MQTTClient_publish.c | 5 +- src/samples/MQTTClient_publish_async.c | 5 +- src/samples/MQTTClient_subscribe.c | 5 +- 15 files changed, 197 insertions(+), 44 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index b96eccb7..8576107a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -65,6 +65,14 @@ option(PAHO_ENABLE_CPACK "Enable CPack" TRUE) option(PAHO_HIGH_PERFORMANCE "Disable tracing and heap tracking" FALSE) option(PAHO_USE_SELECT "Revert to select system call instead of poll" FALSE) +if(NOT WIN32) + option(PAHO_WITH_UNIX_SOCKETS "Flag that defines whether to enable Unix-domain sockets" FALSE) + + if(PAHO_WITH_UNIX_SOCKETS) + add_definitions(-DUNIXSOCK=1) + endif() +endif() + if(PAHO_HIGH_PERFORMANCE) add_definitions(-DHIGH_PERFORMANCE=1) endif() diff --git a/README.md b/README.md index 53f36e37..ebc7fde6 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,27 @@ Some potentially useful blog posts: [Various MQTT and MQTT-SN talks I've given.](https://modelbasedtesting.co.uk/talks-ive-given/) +### Supported Network Protocols + +The library supports connecting to an MQTT server using TCP, SSL/TLS, Unix-domain sockets, and websockets (secure and insecure). This is chosen by the client using the URI supplied in the connect options. It can be specified as: + + "mqtt://:" - TCP, unsecure + "tcp://:" (same) + + "mqtts://:" - SSL/TLS + "ssl://:" (same) + + "unix:///path/to/socket - UNIX-domain socket (*nix systems only) + + "ws://:[/path]" - Websockets, unsecure + "wss://:[/path]" - Websockets, secure + +The "mqtt://" and "tcp://" schemas are identical. They indicate an insecure connection over TCP. The "mqtt://" variation is new for the library, but becoming more common across different MQTT libraries. + +Similarly, the "mqtts://" and "ssl://" schemas are identical. They specify a secure connection over SSL/TLS sockets. The use any of the secure connect options requires that you compile the library with the `PAHO_WITH_SSL=TRUE` CMake option to include OpenSSL. In addition, you _must_ specify `ssl_options` when you connect to the broker - i.e. you must add an instance of `ssl_options` to the `connect_options` when calling `connect()`. + +The use of Unix-domain sockets requires the build option of `PAHO_WITH_UNIX_SOCKETS=TRUE` is required. This is only available on *nix-style systems like Linux and macOS. It is not vailable on Windows. + ## Runtime tracing A number of environment variables control runtime tracing of the C library. @@ -161,10 +182,11 @@ Variable | Default Value | Description PAHO_BUILD_SHARED | TRUE | Build a shared version of the libraries PAHO_BUILD_STATIC | FALSE | Build a static version of the libraries PAHO_HIGH_PERFORMANCE | FALSE | When set to true, the debugging aids internal tracing and heap tracking are not included. -PAHO_WITH_SSL | FALSE | Flag that defines whether to build ssl-enabled binaries too. +PAHO_WITH_SSL | FALSE | Flag that defines whether to build ssl-enabled binaries too. OPENSSL_ROOT_DIR | "" (system default) | Directory containing your OpenSSL installation (i.e. `/usr/local` when headers are in `/usr/local/include` and libraries are in `/usr/local/lib`) PAHO_WITH_LIBRESSL | FALSE | Flag that defines whether to build ssl-enabled binaries with LibreSSL instead of OpenSSL. LIBRESSL_ROOT_DIR | "" (system default) | Directory containing your LibreSSL installation (i.e. `/usr/local` when headers are in `/usr/local/include` and libraries are in `/usr/local/lib`) +PAHO_WITH_UNIX_SOCKETS | FALSE | (*nix systems only) Flag to enable support for UNIX-domain sockets PAHO_BUILD_DOCUMENTATION | FALSE | Create and install the HTML based API documentation (requires Doxygen) PAHO_BUILD_SAMPLES | FALSE | Build sample programs PAHO_ENABLE_TESTING | TRUE | Build test and run diff --git a/src/MQTTAsync.c b/src/MQTTAsync.c index 26f87b97..c548ae31 100644 --- a/src/MQTTAsync.c +++ b/src/MQTTAsync.c @@ -329,6 +329,9 @@ int MQTTAsync_createWithOptions(MQTTAsync* handle, const char* serverURI, const { if (strncmp(URI_TCP, serverURI, strlen(URI_TCP)) != 0 && strncmp(URI_MQTT, serverURI, strlen(URI_MQTT)) != 0 +#if defined(UNIXSOCK) + && strncmp(URI_UNIX, serverURI, strlen(URI_UNIX)) != 0 +#endif && strncmp(URI_WS, serverURI, strlen(URI_WS)) != 0 #if defined(OPENSSL) && strncmp(URI_SSL, serverURI, strlen(URI_SSL)) != 0 @@ -384,6 +387,13 @@ int MQTTAsync_createWithOptions(MQTTAsync* handle, const char* serverURI, const serverURI += strlen(URI_TCP); else if (strncmp(URI_MQTT, serverURI, strlen(URI_MQTT)) == 0) serverURI += strlen(URI_MQTT); +#if defined(UNIXSOCK) + else if (strncmp(URI_UNIX, serverURI, strlen(URI_UNIX)) == 0) + { + serverURI += strlen(URI_UNIX); + m->unixsock = 1; + } +#endif else if (strncmp(URI_WS, serverURI, strlen(URI_WS)) == 0) { serverURI += strlen(URI_WS); diff --git a/src/MQTTAsyncUtils.c b/src/MQTTAsyncUtils.c index c6725d73..85158b62 100644 --- a/src/MQTTAsyncUtils.c +++ b/src/MQTTAsyncUtils.c @@ -1330,6 +1330,13 @@ static int MQTTAsync_processCommand(void) serverURI += strlen(URI_TCP); else if (strncmp(URI_MQTT, serverURI, strlen(URI_MQTT)) == 0) serverURI += strlen(URI_MQTT); +#if defined(UNIXSOCK) + else if (strncmp(URI_UNIX, serverURI, strlen(URI_UNIX)) == 0) + { + serverURI += strlen(URI_UNIX); + command->client->unixsock = 1; + } +#endif else if (strncmp(URI_WS, serverURI, strlen(URI_WS)) == 0) { serverURI += strlen(URI_WS); @@ -1369,18 +1376,18 @@ static int MQTTAsync_processCommand(void) Log(TRACE_PROTOCOL, -1, "Connecting to serverURI %s with MQTT version %d", serverURI, command->command.details.conn.MQTTVersion); #if defined(OPENSSL) #if defined(__GNUC__) && defined(__linux__) - rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->ssl, command->client->websocket, + rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->unixsock, command->client->ssl, command->client->websocket, command->command.details.conn.MQTTVersion, command->client->connectProps, command->client->willProps, 100); #else - rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->ssl, command->client->websocket, + rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->unixsock, command->client->ssl, command->client->websocket, command->command.details.conn.MQTTVersion, command->client->connectProps, command->client->willProps); #endif #else #if defined(__GNUC__) && defined(__linux__) - rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->websocket, + rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->unixsock, command->client->websocket, command->command.details.conn.MQTTVersion, command->client->connectProps, command->client->willProps, 100); #else - rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->websocket, + rc = MQTTProtocol_connect(serverURI, command->client->c, command->client->unixsock, command->client->websocket, command->command.details.conn.MQTTVersion, command->client->connectProps, command->client->willProps); #endif #endif diff --git a/src/MQTTAsyncUtils.h b/src/MQTTAsyncUtils.h index f242d0a8..541c6611 100644 --- a/src/MQTTAsyncUtils.h +++ b/src/MQTTAsyncUtils.h @@ -24,6 +24,7 @@ #define URI_MQTT "mqtt://" #define URI_WS "ws://" #define URI_WSS "wss://" +#define URI_UNIX "unix://" enum MQTTAsync_threadStates { @@ -89,6 +90,7 @@ typedef struct typedef struct MQTTAsync_struct { char* serverURI; + int unixsock; int ssl; int websocket; Clients* c; diff --git a/src/MQTTClient.c b/src/MQTTClient.c index b26af8d1..32d59c3b 100644 --- a/src/MQTTClient.c +++ b/src/MQTTClient.c @@ -80,6 +80,7 @@ #define URI_MQTT "mqtt://" #define URI_WS "ws://" #define URI_WSS "wss://" +#define URI_UNIX "unix://" #include "VersionInfo.h" #include "WebSocket.h" @@ -294,6 +295,7 @@ typedef struct { char* serverURI; const char* currentServerURI; /* when using HA options, set the currently used serverURI */ + int unixsock; #if defined(OPENSSL) int ssl; #endif @@ -403,6 +405,9 @@ int MQTTClient_createWithOptions(MQTTClient* handle, const char* serverURI, cons && strncmp(URI_SSL, serverURI, strlen(URI_SSL)) != 0 && strncmp(URI_MQTTS, serverURI, strlen(URI_MQTTS)) != 0 && strncmp(URI_WSS, serverURI, strlen(URI_WSS)) != 0 +#endif +#if defined(UNIXSOCK) + && strncmp(URI_UNIX, serverURI, strlen(URI_UNIX)) != 0 #endif ) { @@ -483,6 +488,13 @@ int MQTTClient_createWithOptions(MQTTClient* handle, const char* serverURI, cons goto exit; #endif } +#if defined(UNIXSOCK) + else if (strncmp(URI_UNIX, serverURI, strlen(URI_UNIX)) == 0) + { + serverURI += strlen(URI_UNIX); + m->unixsock = 1; + } +#endif m->serverURI = MQTTStrdup(serverURI); ListAppend(handles, m, sizeof(MQTTClients)); @@ -1235,17 +1247,17 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c Log(TRACE_MIN, -1, "Connecting to serverURI %s with MQTT version %d", serverURI, MQTTVersion); #if defined(OPENSSL) #if defined(__GNUC__) && defined(__linux__) - rc = MQTTProtocol_connect(serverURI, m->c, m->ssl, m->websocket, MQTTVersion, connectProperties, willProperties, + rc = MQTTProtocol_connect(serverURI, m->c, m->unixsock, m->ssl, m->websocket, MQTTVersion, connectProperties, willProperties, millisecsTimeout - MQTTTime_elapsed(start)); #else - rc = MQTTProtocol_connect(serverURI, m->c, m->ssl, m->websocket, MQTTVersion, connectProperties, willProperties); + rc = MQTTProtocol_connect(serverURI, m->c, m->unixsock, m->ssl, m->websocket, MQTTVersion, connectProperties, willProperties); #endif #else #if defined(__GNUC__) && defined(__linux__) - rc = MQTTProtocol_connect(serverURI, m->c, m->websocket, MQTTVersion, connectProperties, willProperties, + rc = MQTTProtocol_connect(serverURI, m->c, m->unixsock, m->websocket, MQTTVersion, connectProperties, willProperties, millisecsTimeout - MQTTTime_elapsed(start)); #else - rc = MQTTProtocol_connect(serverURI, m->c, m->websocket, MQTTVersion, connectProperties, willProperties); + rc = MQTTProtocol_connect(serverURI, m->c, m->unixsock, m->websocket, MQTTVersion, connectProperties, willProperties); #endif #endif if (rc == SOCKET_ERROR) @@ -1897,6 +1909,13 @@ MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions* m->ssl = 1; m->websocket = 1; } +#endif +#if defined(UNIXSOCK) + else if (strncmp(URI_UNIX, serverURI, strlen(URI_UNIX)) == 0) + { + serverURI += strlen(URI_UNIX); + m->unixsock = 1; + } #endif rc = MQTTClient_connectURI(handle, options, serverURI, connectProperties, willProperties); if (rc.reasonCode == MQTTREASONCODE_SUCCESS) diff --git a/src/MQTTProtocolOut.c b/src/MQTTProtocolOut.c index 6e0abc82..b1d1470a 100644 --- a/src/MQTTProtocolOut.c +++ b/src/MQTTProtocolOut.c @@ -202,27 +202,33 @@ exit: /** * MQTT outgoing connect processing for a client - * @param ip_address the TCP address:port to connect to + * @param address The address of the server. For TCP this is in the form + * 'address:port; for a UNIX socket it's the path to the + * socket file, etc. * @param aClient a structure with all MQTT data needed - * @param int ssl - * @param int MQTTVersion the MQTT version to connect with (3 or 4) - * @param long timeout how long to wait for a new socket to be created + * @param unixsock Whether the address if for a UNIX-domain socket + * @param ssl Whether we're connecting with SSL/TLS + * @param websocket Whether we should use a websocket for the connection + * @param MQTTVersion the MQTT version to connect with (3, 4, or 5) + * @param connectProperties The connection properties + * @param willProperties Properties for the LWT + * @param timeout how long to wait for a new socket to be created * @return return code */ #if defined(OPENSSL) #if defined(__GNUC__) && defined(__linux__) -int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int ssl, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* address, Clients* aClient, int unixsock, int ssl, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties, long timeout) #else -int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int ssl, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* address, Clients* aClient, int unixsock, int ssl, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties) #endif #else #if defined(__GNUC__) && defined(__linux__) -int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* address, Clients* aClient, int unixsock, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties, long timeout) #else -int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* address, Clients* aClient, int unixsock, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties) #endif #endif @@ -318,22 +324,28 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket rc = Socket_new(aClient->net.https_proxy, addr_len, port, &(aClient->net.socket)); #endif } +#endif +#if defined(UNIXSOCK) + else if (unixsock) { + addr_len = strlen(address); + rc = Socket_unix_new(address, addr_len, &(aClient->net.socket)); + } #endif else { #if defined(OPENSSL) - addr_len = MQTTProtocol_addressPort(ip_address, &port, NULL, ssl ? + addr_len = MQTTProtocol_addressPort(address, &port, NULL, ssl ? (websocket ? WSS_DEFAULT_PORT : SECURE_MQTT_DEFAULT_PORT) : (websocket ? WS_DEFAULT_PORT : MQTT_DEFAULT_PORT) ); #else - addr_len = MQTTProtocol_addressPort(ip_address, &port, NULL, websocket ? WS_DEFAULT_PORT : MQTT_DEFAULT_PORT); + addr_len = MQTTProtocol_addressPort(address, &port, NULL, websocket ? WS_DEFAULT_PORT : MQTT_DEFAULT_PORT); #endif #if defined(__GNUC__) && defined(__linux__) if (timeout < 0) rc = -1; else - rc = Socket_new(ip_address, addr_len, port, &(aClient->net.socket), timeout); + rc = Socket_new(address, addr_len, port, &(aClient->net.socket), timeout); #else - rc = Socket_new(ip_address, addr_len, port, &(aClient->net.socket)); + rc = Socket_new(address, addr_len, port, &(aClient->net.socket)); #endif } if (rc == EINPROGRESS || rc == EWOULDBLOCK) @@ -345,14 +357,14 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket { if (aClient->net.https_proxy) { aClient->connect_state = PROXY_CONNECT_IN_PROGRESS; - rc = Proxy_connect( &aClient->net, 1, ip_address); + rc = Proxy_connect( &aClient->net, 1, address); } - if (rc == 0 && SSLSocket_setSocketForSSL(&aClient->net, aClient->sslopts, ip_address, addr_len) == 1) + if (rc == 0 && SSLSocket_setSocketForSSL(&aClient->net, aClient->sslopts, address, addr_len) == 1) { rc = aClient->sslopts->struct_version >= 3 ? - SSLSocket_connect(aClient->net.ssl, aClient->net.socket, ip_address, + SSLSocket_connect(aClient->net.ssl, aClient->net.socket, address, aClient->sslopts->verify, aClient->sslopts->ssl_error_cb, aClient->sslopts->ssl_error_context) : - SSLSocket_connect(aClient->net.ssl, aClient->net.socket, ip_address, + SSLSocket_connect(aClient->net.ssl, aClient->net.socket, address, aClient->sslopts->verify, NULL, NULL); if (rc == TCPSOCKET_INTERRUPTED) aClient->connect_state = SSL_IN_PROGRESS; /* SSL connect called - wait for completion */ @@ -365,14 +377,14 @@ int MQTTProtocol_connect(const char* ip_address, Clients* aClient, int websocket if (aClient->net.http_proxy) { #endif aClient->connect_state = PROXY_CONNECT_IN_PROGRESS; - rc = Proxy_connect( &aClient->net, 0, ip_address); + rc = Proxy_connect( &aClient->net, 0, address); } if ( websocket ) { #if defined(OPENSSL) - rc = WebSocket_connect(&aClient->net, ssl, ip_address); + rc = WebSocket_connect(&aClient->net, ssl, address); #else - rc = WebSocket_connect(&aClient->net, 0, ip_address); + rc = WebSocket_connect(&aClient->net, 0, address); #endif if ( rc == TCPSOCKET_INTERRUPTED ) aClient->connect_state = WEBSOCKET_IN_PROGRESS; /* Websocket connect called - wait for completion */ diff --git a/src/MQTTProtocolOut.h b/src/MQTTProtocolOut.h index adc95abf..da144196 100644 --- a/src/MQTTProtocolOut.h +++ b/src/MQTTProtocolOut.h @@ -40,18 +40,18 @@ size_t MQTTProtocol_addressPort(const char* uri, int* port, const char **topic, void MQTTProtocol_reconnect(const char* ip_address, Clients* client); #if defined(OPENSSL) #if defined(__GNUC__) && defined(__linux__) -int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int ssl, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int unixsock, int ssl, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties, long timeout); #else -int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int ssl, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int unixsock, int ssl, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties); #endif #else #if defined(__GNUC__) && defined(__linux__) -int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int unixsock, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties, long timeout); #else -int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int websocket, int MQTTVersion, +int MQTTProtocol_connect(const char* ip_address, Clients* acClients, int unixsock, int websocket, int MQTTVersion, MQTTProperties* connectProperties, MQTTProperties* willProperties); #endif #endif diff --git a/src/Socket.c b/src/Socket.c index 34599adf..8331e8a6 100644 --- a/src/Socket.c +++ b/src/Socket.c @@ -44,6 +44,10 @@ #include "Heap.h" +#if defined(UNIXSOCK) +#include +#endif + #if defined(USE_SELECT) int isReady(int socket, fd_set* read_set, fd_set* write_set); int Socket_continueWrites(fd_set* pwset, SOCKET* socket, mutex_type mutex); @@ -1101,6 +1105,7 @@ exit: /** * Create a new socket and TCP connect to an address/port * @param addr the address string + * @param assr_len the length of the address string * @param port the TCP port * @param sock returns the new socket * @param timeout the timeout in milliseconds @@ -1231,13 +1236,13 @@ int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* sock) return codes from send, for testing only! */ #if defined(SMALL_TCP_BUFFER_TESTING) - if (1) - { - int optsend = 100; //2 * 1440; - printf("Setting optsend to %d\n", optsend); - if (setsockopt(*sock, SOL_SOCKET, SO_SNDBUF, (void*)&optsend, sizeof(optsend)) != 0) - Log(LOG_ERROR, -1, "Could not set SO_SNDBUF for socket %d", *sock); - } + if (1) + { + int optsend = 100; //2 * 1440; + printf("Setting optsend to %d\n", optsend); + if (setsockopt(*sock, SOL_SOCKET, SO_SNDBUF, (void*)&optsend, sizeof(optsend)) != 0) + Log(LOG_ERROR, -1, "Could not set SO_SNDBUF for socket %d", *sock); + } #endif Log(TRACE_MIN, -1, "New socket %d for %s, port %d", *sock, addr, port); if (Socket_addSocket(*sock) == SOCKET_ERROR) @@ -1294,6 +1299,58 @@ exit: return rc; } +#if defined(UNIXSOCK) +/** + * Create a new socket and TCP connect to an address/port + * @param addr the address string, which is a file path + * @param assr_len the length of the address string + * @param sock returns the new socket + * @return completion code 0=good, SOCKET_ERROR=fail + */ +int Socket_unix_new(const char* addr, size_t addr_len, SOCKET* sock) +{ + struct sockaddr_un address; + int rc = SOCKET_ERROR; + + FUNC_ENTRY; + + if (addr_len >= sizeof(address.sun_path)) { + rc = PAHO_MEMORY_ERROR; + } + else { + address.sun_family = AF_UNIX; + memcpy(&address.sun_path, addr, addr_len); + address.sun_path[addr_len] = '\0'; + + *sock = socket(AF_UNIX, SOCK_STREAM, 0); + if (*sock == INVALID_SOCKET) + rc = Socket_error("socket", *sock); + else + { +#if defined(NOSIGPIPE) + int opt = 1; + if (setsockopt(*sock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0) + Log(LOG_ERROR, -1, "Could not set SO_NOSIGPIPE for socket %d", *sock); +#endif + Log(TRACE_MIN, -1, "New UNIX socket %d for %s", *sock, addr); + if (Socket_addSocket(*sock) == SOCKET_ERROR) + rc = Socket_error("addSocket", *sock); + else + { + /* this will complete immediately, even though we are non-blocking */ + rc = connect(*sock, (struct sockaddr*)&address, sizeof(address)); + if (rc == SOCKET_ERROR) + rc = Socket_error("connect", *sock); + } + } + } + +exit: + FUNC_EXIT_RC(rc); + return rc; +} +#endif + static Socket_writeContinue* writecontinue = NULL; void Socket_setWriteContinueCallback(Socket_writeContinue* mywritecontinue) diff --git a/src/Socket.h b/src/Socket.h index e5b9fe63..6e52ab8f 100644 --- a/src/Socket.h +++ b/src/Socket.h @@ -149,6 +149,7 @@ int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* socket, long #else int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* socket); #endif +int Socket_unix_new(const char* addr, size_t addr_len, SOCKET* sock); int Socket_noPendingWrites(SOCKET socket); char* Socket_getpeer(SOCKET sock); diff --git a/src/samples/MQTTAsync_publish.c b/src/samples/MQTTAsync_publish.c index 7695a745..0a4a4c6d 100644 --- a/src/samples/MQTTAsync_publish.c +++ b/src/samples/MQTTAsync_publish.c @@ -146,7 +146,10 @@ int main(int argc, char* argv[]) MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer; int rc; - if ((rc = MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTASYNC_SUCCESS) + const char* uri = (argc > 1) ? argv[1] : ADDRESS; + printf("Using server at %s\n", uri); + + if ((rc = MQTTAsync_create(&client, uri, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTASYNC_SUCCESS) { printf("Failed to create client object, return code %d\n", rc); exit(EXIT_FAILURE); diff --git a/src/samples/MQTTAsync_subscribe.c b/src/samples/MQTTAsync_subscribe.c index ae329116..29a504e3 100644 --- a/src/samples/MQTTAsync_subscribe.c +++ b/src/samples/MQTTAsync_subscribe.c @@ -137,7 +137,10 @@ int main(int argc, char* argv[]) int rc; int ch; - if ((rc = MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) + const char* uri = (argc > 1) ? argv[1] : ADDRESS; + printf("Using server at %s\n", uri); + + if ((rc = MQTTAsync_create(&client, uri, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTASYNC_SUCCESS) { printf("Failed to create client, return code %d\n", rc); diff --git a/src/samples/MQTTClient_publish.c b/src/samples/MQTTClient_publish.c index 399fd339..0cf088d5 100644 --- a/src/samples/MQTTClient_publish.c +++ b/src/samples/MQTTClient_publish.c @@ -34,7 +34,10 @@ int main(int argc, char* argv[]) MQTTClient_deliveryToken token; int rc; - if ((rc = MQTTClient_create(&client, ADDRESS, CLIENTID, + const char* uri = (argc > 1) ? argv[1] : ADDRESS; + printf("Using server at %s\n", uri); + + if ((rc = MQTTClient_create(&client, uri, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) { printf("Failed to create client, return code %d\n", rc); diff --git a/src/samples/MQTTClient_publish_async.c b/src/samples/MQTTClient_publish_async.c index a2b45e87..1e2b50cd 100644 --- a/src/samples/MQTTClient_publish_async.c +++ b/src/samples/MQTTClient_publish_async.c @@ -65,7 +65,10 @@ int main(int argc, char* argv[]) MQTTClient_deliveryToken token; int rc; - if ((rc = MQTTClient_create(&client, ADDRESS, CLIENTID, + const char* uri = (argc > 1) ? argv[1] : ADDRESS; + printf("Using server at %s\n", uri); + + if ((rc = MQTTClient_create(&client, uri, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) { printf("Failed to create client, return code %d\n", rc); diff --git a/src/samples/MQTTClient_subscribe.c b/src/samples/MQTTClient_subscribe.c index 8872dd13..1905eb3d 100644 --- a/src/samples/MQTTClient_subscribe.c +++ b/src/samples/MQTTClient_subscribe.c @@ -57,7 +57,10 @@ int main(int argc, char* argv[]) MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer; int rc; - if ((rc = MQTTClient_create(&client, ADDRESS, CLIENTID, + const char* uri = (argc > 1) ? argv[1] : ADDRESS; + printf("Using server at %s\n", uri); + + if ((rc = MQTTClient_create(&client, uri, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) { printf("Failed to create client, return code %d\n", rc);