This is an automated email from the ASF dual-hosted git repository.
xiaoxiang781216 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nuttx-apps.git
The following commit(s) were added to refs/heads/master by this push:
new 3762a64bc examples/mqttc: add a subscriber client implementation
example
3762a64bc is described below
commit 3762a64bc6ef6781fc76e05275df1de072d97f5f
Author: Mihai Pacuraru <[email protected]>
AuthorDate: Wed Jul 8 14:30:00 2026 +0300
examples/mqttc: add a subscriber client implementation example
Implement the subscriber client that optionally integrates
the TLS layer and can reconnect to the broker if the
application is running in unstable network environments.
The client is compliant with all 3 levels of QoS.
Signed-off-by: Mihai Pacuraru <[email protected]>
---
examples/mqttc/CMakeLists.txt | 14 +-
examples/mqttc/Kconfig | 24 ++
examples/mqttc/Makefile | 10 +
examples/mqttc/cert.inc | 30 ++
examples/mqttc/mqttc_pub.c | 32 +-
examples/mqttc/{mqttc_pub.c => mqttc_sub.c} | 454 +++++++++++++++++++---------
6 files changed, 399 insertions(+), 165 deletions(-)
diff --git a/examples/mqttc/CMakeLists.txt b/examples/mqttc/CMakeLists.txt
index 567712558..6e4ee010a 100644
--- a/examples/mqttc/CMakeLists.txt
+++ b/examples/mqttc/CMakeLists.txt
@@ -23,11 +23,23 @@
if(CONFIG_EXAMPLES_MQTTC)
nuttx_add_application(
NAME
- ${CONFIG_EXAMPLES_MQTTC_PROGNAME}
+ ${CONFIG_EXAMPLES_MQTTC_PUB_PROGNAME}
STACKSIZE
${CONFIG_EXAMPLES_MQTTC_STACKSIZE}
MODULE
${CONFIG_EXAMPLES_MQTTC}
SRCS
mqttc_pub.c)
+
+ if(CONFIG_EXAMPLES_MQTTC_SUB)
+ nuttx_add_application(
+ NAME
+ ${CONFIG_EXAMPLES_MQTTC_SUB_PROGNAME}
+ STACKSIZE
+ ${CONFIG_EXAMPLES_MQTTC_STACKSIZE}
+ MODULE
+ ${CONFIG_EXAMPLES_MQTTC}
+ SRCS
+ mqttc_sub.c)
+ endif()
endif()
diff --git a/examples/mqttc/Kconfig b/examples/mqttc/Kconfig
index adf91b726..61a2e6ba1 100644
--- a/examples/mqttc/Kconfig
+++ b/examples/mqttc/Kconfig
@@ -16,6 +16,17 @@ config EXAMPLES_MQTTC_PROGNAME
string "Program name"
default "mqttc_pub"
+config EXAMPLES_MQTTC_SUB
+ bool "Enable MQTT-C Subscriber Client"
+ default n
+ ---help---
+ Enable a simple MQTT-C subscriber example alongside publisher
+
+config EXAMPLES_MQTTC_SUB_PROGNAME
+ string "Subscriber program name"
+ default "mqttc_sub"
+ depends on EXAMPLES_MQTTC_SUB
+
config EXAMPLES_MQTTC_STACKSIZE
int "Task's stack size"
default 8192
@@ -28,6 +39,19 @@ config EXAMPLES_MQTTC_RXSIZE
int "RX Buffer size"
default 256
+config MQTT_USE_MBEDTLS
+ bool
+ default n
+
+config EXAMPLES_MQTTC_MBEDTLS
+ bool "Enable MQTT over TLS (mbedTLS)"
+ default n
+ depends on NETUTILS_MQTTC_WITH_MBEDTLS
+ select MQTT_USE_MBEDTLS
+ ---help---
+ Enable TLS/SSL support for MQTT connections using mbedTLS.
+ Requires NETUTILS_MQTTC_WITH_MBEDTLS to be enabled.
+
config EXAMPLES_MQTTC_ALLOW_UNVERIFIED_TLS
bool "Allow unverified TLS"
default n
diff --git a/examples/mqttc/Makefile b/examples/mqttc/Makefile
index 9aacf67f6..45a2d9f04 100644
--- a/examples/mqttc/Makefile
+++ b/examples/mqttc/Makefile
@@ -25,10 +25,20 @@ include $(APPDIR)/Make.defs
PROGNAME = $(CONFIG_EXAMPLES_MQTTC_PROGNAME)
PRIORITY = SCHED_PRIORITY_DEFAULT
STACKSIZE = $(CONFIG_EXAMPLES_MQTTC_STACKSIZE)
+
+ifeq ($(CONFIG_EXAMPLES_MQTTC_SUB),y)
+ PROGNAME += $(CONFIG_EXAMPLES_MQTTC_SUB_PROGNAME)
+ PRIORITY += SCHED_PRIORITY_DEFAULT
+ STACKSIZE += $(CONFIG_EXAMPLES_MQTTC_STACKSIZE)
+endif
+
MODULE = $(CONFIG_EXAMPLES_MQTTC)
# MQTT-C example source code
MAINSRC = mqttc_pub.c
+ifeq ($(CONFIG_EXAMPLES_MQTTC_SUB), y)
+ MAINSRC += mqttc_sub.c
+endif
include $(APPDIR)/Application.mk
diff --git a/examples/mqttc/cert.inc b/examples/mqttc/cert.inc
new file mode 100644
index 000000000..542602db5
--- /dev/null
+++ b/examples/mqttc/cert.inc
@@ -0,0 +1,30 @@
+/* Broker CA certificate (PEM), embedded in the firmware so the subscriber
+ * can authenticate the broker without copying a certificate file onto the
+ * board.
+ * Replace the body below if the broker's CA changes; init_conn() parses
+ * it whenever no CA file is supplied on the command line via -c.
+ */
+
+# define TEST_CA_CRT_RSA_SHA256_PEM \
+ "-----BEGIN CERTIFICATE-----\r\n" \
+ "MIIDYTCCAkmgAwIBAgIUAk1VwJ7il2775wUnEgBZSvW77dUwDQYJKoZIhvcNAQEL\r\n" \
+ "BQAwQDELMAkGA1UEBhMCUk8xEjAQBgNVBAgMCUJ1Y2hhcmVzdDEOMAwGA1UECgwF\r\n" \
+ "TXlPcmcxDTALBgNVBAMMBE15Q0EwHhcNMjYwNjA5MDA0MjExWhcNMzYwNjA2MDA0\r\n" \
+ "MjExWjBAMQswCQYDVQQGEwJSTzESMBAGA1UECAwJQnVjaGFyZXN0MQ4wDAYDVQQK\r\n" \
+ "DAVNeU9yZzENMAsGA1UEAwwETXlDQTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCC\r\n" \
+ "AQoCggEBAMGfLamBuio0mYqL8TDLqcxqCDhELmrmYxccz8/BAJ3xG0bbqAg0o2CM\r\n" \
+ "2nUcDWeDTTOFTEAOVKhqqbu7jzQA+/XRQaptYAzt13N0pMUFlKZQ0QVefAH11mph\r\n" \
+ "M2u0exWZObRQxUyfzzJG1UZQ4xGqXUekg+2LMSS09g/QAsCreu8O+xzAK9XUA6np\r\n" \
+ "8lebaIqxFHabrjlbHgan5qKP0+EncMRN/5Lb11MDzfjyIusoFWPZF8rF2x9D+UFv\r\n" \
+ "hZuL9Z40jz3cRlVUxuVtUq1tna2CCwWhwl5gwAQ/b0rkhe+5/b3cgSs3VbIdWHic\r\n" \
+ "nwM4Yq4K/nAetYiOxtjKY5zLIvRopd0CAwEAAaNTMFEwHQYDVR0OBBYEFFR43w+x\r\n" \
+ "ri1ssaAyJzQGDH/dP2e0MB8GA1UdIwQYMBaAFFR43w+xri1ssaAyJzQGDH/dP2e0\r\n" \
+ "MA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBADU/lBjTGb7RlgGq\r\n" \
+ "HYCktoWZBNkJFPVTPrC1mwzJX/vSaz9QUpRA5dpprSl84QKvWNxO5fPEMmwK2Wrw\r\n" \
+ "x6tdE2aWb5x+3W0302JnaUPNhDF26Z5+LK4CoRHjbT2J3FrDg2u6/3Si+RpNSHYh\r\n" \
+ "0OCdApzCyzqWdmvb7VD+u+isSh9CEUcaBOCq/72GNqdIjnSsgMGNHp9Ld6+ctwJO\r\n" \
+ "sQkj0M3MaciNSPZ3y0w/GqeWNQaK3Rk5pSaqrc9RQS42KtJuGOqMiBMG1Ib6b7D4\r\n" \
+ "bJSEDBzhO1M0o9vwr9xYaCUfbQzuuYJ3al4weRl0E4B63o2OFUjMOILIO3LeuuRD\r\n" \
+ "+HXdOE0=\r\n" \
+ "-----END CERTIFICATE-----\r\n"
+/* END FILE */
diff --git a/examples/mqttc/mqttc_pub.c b/examples/mqttc/mqttc_pub.c
index 306df6dcc..d979fb24a 100644
--- a/examples/mqttc/mqttc_pub.c
+++ b/examples/mqttc/mqttc_pub.c
@@ -24,6 +24,7 @@
* Included Files
****************************************************************************/
+#include <nuttx/compiler.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
@@ -42,39 +43,12 @@
# include <mbedtls/net_sockets.h>
# include <mbedtls/ssl.h>
+#include "cert.inc"
+
/****************************************************************************
* Pre-processor Definitions
****************************************************************************/
-/* This is taken from mbedtls/tests/data_files/test-ca-sha256.crt. */
-
-/* BEGIN FILE string macro TEST_CA_CRT_RSA_SHA256_PEM
- * mbedtls/tests/data_files/test-ca-sha256.crt
- */
-
-# define TEST_CA_CRT_RSA_SHA256_PEM \
- "-----BEGIN CERTIFICATE-----\r\n" \
- "MIIDQTCCAimgAwIBAgIBAzANBgkqhkiG9w0BAQsFADA7MQswCQYDVQQGEwJOTDER\r\n" \
- "MA8GA1UECgwIUG9sYXJTU0wxGTAXBgNVBAMMEFBvbGFyU1NMIFRlc3QgQ0EwHhcN\r\n" \
- "MTkwMjEwMTQ0NDAwWhcNMjkwMjEwMTQ0NDAwWjA7MQswCQYDVQQGEwJOTDERMA8G\r\n" \
- "A1UECgwIUG9sYXJTU0wxGTAXBgNVBAMMEFBvbGFyU1NMIFRlc3QgQ0EwggEiMA0G\r\n" \
- "CSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDA3zf8F7vglp0/ht6WMn1EpRagzSHx\r\n" \
- "mdTs6st8GFgIlKXsm8WL3xoemTiZhx57wI053zhdcHgH057Zk+i5clHFzqMwUqny\r\n" \
- "50BwFMtEonILwuVA+T7lpg6z+exKY8C4KQB0nFc7qKUEkHHxvYPZP9al4jwqj+8n\r\n" \
- "YMPGn8u67GB9t+aEMr5P+1gmIgNb1LTV+/Xjli5wwOQuvfwu7uJBVcA0Ln0kcmnL\r\n" \
- "R7EUQIN9Z/SG9jGr8XmksrUuEvmEF/Bibyc+E1ixVA0hmnM3oTDPb5Lc9un8rNsu\r\n" \
- "KNF+AksjoBXyOGVkCeoMbo4bF6BxyLObyavpw/LPh5aPgAIynplYb6LVAgMBAAGj\r\n" \
- "UDBOMAwGA1UdEwQFMAMBAf8wHQYDVR0OBBYEFLRa5KWz3tJS9rnVppUP6z68x/3/\r\n" \
- "MB8GA1UdIwQYMBaAFLRa5KWz3tJS9rnVppUP6z68x/3/MA0GCSqGSIb3DQEBCwUA\r\n" \
- "A4IBAQA4qFSCth2q22uJIdE4KGHJsJjVEfw2/xn+MkTvCMfxVrvmRvqCtjE4tKDl\r\n" \
- "oK4MxFOek07oDZwvtAT9ijn1hHftTNS7RH9zd/fxNpfcHnMZXVC4w4DNA1fSANtW\r\n" \
- "5sY1JB5Je9jScrsLSS+mAjyv0Ow3Hb2Bix8wu7xNNrV5fIf7Ubm+wt6SqEBxu3Kb\r\n" \
- "+EfObAT4huf3czznhH3C17ed6NSbXwoXfby7stWUDeRJv08RaFOykf/Aae7bY5PL\r\n" \
- "yTVrkAnikMntJ9YI+hNNYt3inqq11A5cN0+rVTst8UKCxzQ4GpvroSwPKTFkbMw4\r\n" \
- "/anT1dVxr/BtwJfiESoK3/4CeXR1\r\n" \
- "-----END CERTIFICATE-----\r\n"
-/* END FILE */
-
# define EXTRA_OPT "c:"
#else
# define EXTRA_OPT ""
diff --git a/examples/mqttc/mqttc_pub.c b/examples/mqttc/mqttc_sub.c
similarity index 56%
copy from examples/mqttc/mqttc_pub.c
copy to examples/mqttc/mqttc_sub.c
index 306df6dcc..d7ad1858a 100644
--- a/examples/mqttc/mqttc_pub.c
+++ b/examples/mqttc/mqttc_sub.c
@@ -1,5 +1,5 @@
/****************************************************************************
- * apps/examples/mqttc/mqttc_pub.c
+ * apps/examples/mqttc/mqttc_sub.c
*
* SPDX-License-Identifier: Apache-2.0
*
@@ -24,14 +24,17 @@
* Included Files
****************************************************************************/
+#include <sys/types.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/socket.h>
#include <fcntl.h>
#include <netdb.h>
+#include <arpa/inet.h>
#include <errno.h>
#include <mqtt.h>
+#include <netutils/netlib.h>
#ifdef MQTT_USE_MBEDTLS
# include <inttypes.h>
@@ -42,44 +45,32 @@
# include <mbedtls/net_sockets.h>
# include <mbedtls/ssl.h>
+# include "cert.inc"
+
/****************************************************************************
* Pre-processor Definitions
****************************************************************************/
-/* This is taken from mbedtls/tests/data_files/test-ca-sha256.crt. */
-
-/* BEGIN FILE string macro TEST_CA_CRT_RSA_SHA256_PEM
- * mbedtls/tests/data_files/test-ca-sha256.crt
- */
-
-# define TEST_CA_CRT_RSA_SHA256_PEM \
- "-----BEGIN CERTIFICATE-----\r\n" \
- "MIIDQTCCAimgAwIBAgIBAzANBgkqhkiG9w0BAQsFADA7MQswCQYDVQQGEwJOTDER\r\n" \
- "MA8GA1UECgwIUG9sYXJTU0wxGTAXBgNVBAMMEFBvbGFyU1NMIFRlc3QgQ0EwHhcN\r\n" \
- "MTkwMjEwMTQ0NDAwWhcNMjkwMjEwMTQ0NDAwWjA7MQswCQYDVQQGEwJOTDERMA8G\r\n" \
- "A1UECgwIUG9sYXJTU0wxGTAXBgNVBAMMEFBvbGFyU1NMIFRlc3QgQ0EwggEiMA0G\r\n" \
- "CSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDA3zf8F7vglp0/ht6WMn1EpRagzSHx\r\n" \
- "mdTs6st8GFgIlKXsm8WL3xoemTiZhx57wI053zhdcHgH057Zk+i5clHFzqMwUqny\r\n" \
- "50BwFMtEonILwuVA+T7lpg6z+exKY8C4KQB0nFc7qKUEkHHxvYPZP9al4jwqj+8n\r\n" \
- "YMPGn8u67GB9t+aEMr5P+1gmIgNb1LTV+/Xjli5wwOQuvfwu7uJBVcA0Ln0kcmnL\r\n" \
- "R7EUQIN9Z/SG9jGr8XmksrUuEvmEF/Bibyc+E1ixVA0hmnM3oTDPb5Lc9un8rNsu\r\n" \
- "KNF+AksjoBXyOGVkCeoMbo4bF6BxyLObyavpw/LPh5aPgAIynplYb6LVAgMBAAGj\r\n" \
- "UDBOMAwGA1UdEwQFMAMBAf8wHQYDVR0OBBYEFLRa5KWz3tJS9rnVppUP6z68x/3/\r\n" \
- "MB8GA1UdIwQYMBaAFLRa5KWz3tJS9rnVppUP6z68x/3/MA0GCSqGSIb3DQEBCwUA\r\n" \
- "A4IBAQA4qFSCth2q22uJIdE4KGHJsJjVEfw2/xn+MkTvCMfxVrvmRvqCtjE4tKDl\r\n" \
- "oK4MxFOek07oDZwvtAT9ijn1hHftTNS7RH9zd/fxNpfcHnMZXVC4w4DNA1fSANtW\r\n" \
- "5sY1JB5Je9jScrsLSS+mAjyv0Ow3Hb2Bix8wu7xNNrV5fIf7Ubm+wt6SqEBxu3Kb\r\n" \
- "+EfObAT4huf3czznhH3C17ed6NSbXwoXfby7stWUDeRJv08RaFOykf/Aae7bY5PL\r\n" \
- "yTVrkAnikMntJ9YI+hNNYt3inqq11A5cN0+rVTst8UKCxzQ4GpvroSwPKTFkbMw4\r\n" \
- "/anT1dVxr/BtwJfiESoK3/4CeXR1\r\n" \
- "-----END CERTIFICATE-----\r\n"
-/* END FILE */
-
# define EXTRA_OPT "c:"
#else
# define EXTRA_OPT ""
#endif
+/* Seconds to wait between failed connection attempts in the reconnect
+ * callback.
+ */
+
+#define MQTTC_RECONNECT_DELAY_S 2
+
+/* Network interface to wait on before the first connection attempt, and the
+ * polling parameters for that wait. Change MQTTC_NETIF to match the board's
+ * network device (e.g. "eth0" for wired Ethernet).
+ */
+
+#define MQTTC_NETIF "wlan0"
+#define MQTTC_NET_POLL_MS 200
+#define MQTTC_NET_WAIT_MS 30000
+
/****************************************************************************
* Private Types
****************************************************************************/
@@ -93,9 +84,9 @@ struct mqttc_cfg_s
FAR const char *id;
FAR const char *user;
FAR const char *pass;
-#ifdef MQTT_USE_MBEDTLS
- FAR const char *ca_file;
-#endif
+ #ifdef MQTT_USE_MBEDTLS
+ FAR const char *ca_file;
+ #endif
uint32_t tmo;
uint8_t flags;
uint8_t qos;
@@ -104,34 +95,124 @@ struct mqttc_cfg_s
struct mqtt_conn_context_s
{
struct mqtt_client client;
-#ifdef MQTT_USE_MBEDTLS
- mbedtls_net_context net_ctx;
- mbedtls_ssl_context ssl_ctx;
- mbedtls_ssl_config ssl_conf;
- mbedtls_x509_crt ca_crt;
- mbedtls_entropy_context entropy;
- mbedtls_ctr_drbg_context ctr_drbg;
-#endif
+ #ifdef MQTT_USE_MBEDTLS
+ mbedtls_net_context net_ctx;
+ mbedtls_ssl_context ssl_ctx;
+ mbedtls_ssl_config ssl_conf;
+ mbedtls_x509_crt ca_crt;
+ mbedtls_entropy_context entropy;
+ mbedtls_ctr_drbg_context ctr_drbg;
+ #endif
uint8_t sendbuf[CONFIG_EXAMPLES_MQTTC_TXSIZE];
uint8_t recvbuf[CONFIG_EXAMPLES_MQTTC_RXSIZE];
};
+/* State handed to the reconnect callback so it can rebuild the transport
+ * connection and reconfigure the MQTT session from scratch.
+ */
+
+struct mqttc_reconnect_s
+{
+ FAR const struct mqttc_cfg_s *cfg;
+ FAR struct mqtt_conn_context_s *conn;
+};
+
/****************************************************************************
* Private Function Prototypes
****************************************************************************/
static FAR void *client_refresher(FAR void *data);
-static void parsearg(int argc, FAR char *argv[], FAR struct mqttc_cfg_s *cfg,
- FAR int *n);
+static void parsearg(int argc, FAR char *argv[],
+ FAR struct mqttc_cfg_s *cfg);
static int init_conn(FAR const struct mqttc_cfg_s *cfg,
FAR struct mqtt_conn_context_s *ctx,
FAR mqtt_pal_socket_handle *handle);
static void close_conn(FAR struct mqtt_conn_context_s *ctx);
+static void reconnect_client(FAR struct mqtt_client *client,
+ FAR void **state);
+static void wait_for_network(void);
/****************************************************************************
* Private Functions
****************************************************************************/
+#ifdef MQTT_USE_MBEDTLS
+
+/****************************************************************************
+ * Name: mqttc_net_recv
+ *
+ * Description:
+ * mbedTLS BIO receive callback for non-blocking sockets. The stock
+ * mbedtls_net_recv() relies on fcntl(F_GETFL) reporting O_NONBLOCK to
+ * classify EAGAIN as "would block". On NuttX, F_SETFL applies the
+ * non-blocking mode via FIONBIO but does not retain O_NONBLOCK in the
+ * file status flags, so F_GETFL never reports it. The stock callback
+ * therefore turns a benign EAGAIN into MBEDTLS_ERR_NET_RECV_FAILED. This
+ * wrapper inspects errno directly and returns MBEDTLS_ERR_SSL_WANT_READ
+ * so callers (e.g. MQTT-C's mqtt_sync) keep retrying instead of aborting.
+ *
+ ****************************************************************************/
+
+static int mqttc_net_recv(FAR void *ctx, FAR unsigned char *buf, size_t len)
+{
+ int fd = ((FAR mbedtls_net_context *)ctx)->fd;
+ int ret;
+
+ ret = recv(fd, buf, len, 0);
+ if (ret < 0)
+ {
+ if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
+ {
+ return MBEDTLS_ERR_SSL_WANT_READ;
+ }
+
+ if (errno == EPIPE || errno == ECONNRESET)
+ {
+ return MBEDTLS_ERR_NET_CONN_RESET;
+ }
+
+ return MBEDTLS_ERR_NET_RECV_FAILED;
+ }
+
+ return ret;
+}
+
+/****************************************************************************
+ * Name: mqttc_net_send
+ *
+ * Description:
+ * mbedTLS BIO send callback, mirroring mqttc_net_recv() for the same
+ * non-blocking/EAGAIN reason on the transmit side.
+ *
+ ****************************************************************************/
+
+static int mqttc_net_send(FAR void *ctx, FAR const unsigned char *buf,
+ size_t len)
+{
+ int fd = ((FAR mbedtls_net_context *)ctx)->fd;
+ int ret;
+
+ ret = send(fd, buf, len, 0);
+ if (ret < 0)
+ {
+ if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
+ {
+ return MBEDTLS_ERR_SSL_WANT_WRITE;
+ }
+
+ if (errno == EPIPE || errno == ECONNRESET)
+ {
+ return MBEDTLS_ERR_NET_CONN_RESET;
+ }
+
+ return MBEDTLS_ERR_NET_SEND_FAILED;
+ }
+
+ return ret;
+}
+
+#endif /* MQTT_USE_MBEDTLS */
+
/****************************************************************************
* Name: client_refresher
*
@@ -152,6 +233,34 @@ static FAR void *client_refresher(FAR void *data)
return NULL;
}
+/****************************************************************************
+ * Name: publish_callback
+ *
+ * Description:
+ * Print received publish message.
+ *
+ ****************************************************************************/
+
+void publish_callback(void** unused, struct mqtt_response_publish *published)
+{
+ /* this is a byte stream, we have to convert it to C string */
+
+ char message[CONFIG_EXAMPLES_MQTTC_RXSIZE];
+
+ /* convert to C string (null terminated) */
+
+ size_t message_size = published->application_message_size;
+ if (message_size >= CONFIG_EXAMPLES_MQTTC_RXSIZE)
+ {
+ message_size = CONFIG_EXAMPLES_MQTTC_RXSIZE - 1;
+ }
+
+ memcpy(message, published->application_message, message_size);
+ message[message_size] = '\0';
+
+ printf("Received: %s\n", message);
+}
+
/****************************************************************************
* Name: parsearg
*
@@ -161,11 +270,11 @@ static FAR void *client_refresher(FAR void *data)
****************************************************************************/
static void parsearg(int argc, FAR char *argv[],
- FAR struct mqttc_cfg_s *cfg, FAR int *n)
+ FAR struct mqttc_cfg_s *cfg)
{
int opt;
- while ((opt = getopt(argc, argv, "h:p:m:t:n:q:" EXTRA_OPT)) != ERROR)
+ while ((opt = getopt(argc, argv, "h:p:t:q:" EXTRA_OPT)) != ERROR)
{
switch (opt)
{
@@ -177,40 +286,31 @@ static void parsearg(int argc, FAR char *argv[],
cfg->port = optarg;
break;
- case 'm':
- cfg->msg = optarg;
- break;
-
case 't':
cfg->topic = optarg;
break;
- case 'n':
- *n = strtol(optarg, NULL, 10);
- break;
-
case 'q':
switch (strtol(optarg, NULL, 10))
{
case 0L:
- cfg->qos = MQTT_PUBLISH_QOS_0;
+ cfg->qos = 0;
break;
case 1L:
- cfg->qos = MQTT_PUBLISH_QOS_1;
+ cfg->qos = 1;
break;
case 2L:
- cfg->qos = MQTT_PUBLISH_QOS_2;
+ cfg->qos = 2;
break;
}
break;
-
#ifdef MQTT_USE_MBEDTLS
- case 'c':
+ case 'c':
cfg->ca_file = optarg;
break;
#endif
- default:
+ default:
fprintf(stderr, "ERROR: Unrecognized option\n");
break;
}
@@ -269,8 +369,8 @@ static int init_conn(FAR const struct mqttc_cfg_s *cfg,
else
{
ret = mbedtls_x509_crt_parse(ca_crt,
- (FAR const unsigned char *)TEST_CA_CRT_RSA_SHA256_PEM,
- sizeof(TEST_CA_CRT_RSA_SHA256_PEM));
+ (FAR const unsigned char *)TEST_CA_CRT_RSA_SHA256_PEM,
+ sizeof(TEST_CA_CRT_RSA_SHA256_PEM));
}
if (ret != 0)
@@ -325,7 +425,7 @@ static int init_conn(FAR const struct mqttc_cfg_s *cfg,
}
mbedtls_ssl_set_bio(ssl_ctx, net_ctx,
- mbedtls_net_send, mbedtls_net_recv, NULL);
+ mqttc_net_send, mqttc_net_recv, NULL);
for (; ; )
{
@@ -485,76 +585,170 @@ static void close_conn(FAR struct mqtt_conn_context_s
*conn)
}
/****************************************************************************
- * Public Functions
+ * Name: wait_for_network
+ *
+ * Description:
+ * Block until the network interface has a usable IPv4 address (i.e. the
+ * link is up and DHCP has completed) or a timeout elapses. This avoids
+ * the spurious first-attempt connect() failure that happens when the
+ * example starts before Wi-Fi association/DHCP have finished.
+ *
+ ****************************************************************************/
+
+static void wait_for_network(void)
+{
+ struct in_addr addr;
+ int waited = 0;
+
+ for (; ; )
+ {
+ if (netlib_get_ipv4addr(MQTTC_NETIF, &addr) == 0 &&
+ addr.s_addr != INADDR_ANY &&
+ addr.s_addr != htonl(INADDR_LOOPBACK))
+ {
+ return;
+ }
+
+ if (waited >= MQTTC_NET_WAIT_MS)
+ {
+ return;
+ }
+
+ usleep(MQTTC_NET_POLL_MS * 1000);
+ waited += MQTTC_NET_POLL_MS;
+ }
+}
+
+/****************************************************************************
+ * Name: reconnect_client
+ *
+ * Description:
+ * Reconnect callback registered with mqtt_init_reconnect(). MQTT-C calls
+ * this whenever the client enters an error state, and once at start-up
+ * (with client->error == MQTT_ERROR_INITIAL_RECONNECT) to set up the
+ * initial session. It tears down the previous connection, re-establishes
+ * the transport (TCP + optional TLS), reinitializes the client and then
+ * reconfigures the session (CONNECT + SUBSCRIBE).
+ *
****************************************************************************/
-int main(int argc, FAR char *argv[])
+static void reconnect_client(FAR struct mqtt_client *client,
+ FAR void **state)
{
- struct mqtt_conn_context_s mqtt_conn;
+ FAR struct mqttc_reconnect_s *rc =
+ *((FAR struct mqttc_reconnect_s **)state);
+ FAR const struct mqttc_cfg_s *cfg = rc->cfg;
+ FAR struct mqtt_conn_context_s *conn = rc->conn;
mqtt_pal_socket_handle socketfd;
- int timeout = 100;
enum MQTTErrors mqtterr;
- pthread_attr_t attr;
- pthread_t thrdid;
- int n = 1;
- struct mqttc_cfg_s mqtt_cfg =
+
+ /* Tear down the previous connection, unless this is the initial set-up
+ * call (in which case there is nothing to clean up yet).
+ */
+
+ if (client->error != MQTT_ERROR_INITIAL_RECONNECT)
{
- .host = "broker.hivemq.com",
-#ifndef MQTT_USE_MBEDTLS
- .port = "1883",
-#else
- .port = "8883",
-#endif
- .topic = "test",
- .msg = "test",
- .flags = MQTT_CONNECT_CLEAN_SESSION,
- .tmo = 400,
- .id = NULL,
- .user = NULL,
- .pass = NULL,
- .qos = MQTT_PUBLISH_QOS_0
- };
+ printf("reconnect_client: reconnecting after error \"%s\"\n",
+ mqtt_error_str(client->error));
+ close_conn(conn);
+ }
- parsearg(argc, argv, &mqtt_cfg, &n);
+ /* (Re)establish the transport connection.
+ *
+ * mqtt_sync() unconditionally proceeds to recv/send on client->socketfd
+ * after this callback returns. Until mqtt_reinit() runs, socketfd is the
+ * sentinel value -1 set by mqtt_init_reconnect(); in the mbedTLS build
+ * that is a pointer (0xffffffff) that mbedtls_ssl_read() would dereference
+ * and crash on. We must therefore not return until a connection has
+ * actually been established, so retry with a short back-off on failure.
+ */
- if (init_conn(&mqtt_cfg, &mqtt_conn, &socketfd) < 0)
+ while (init_conn(cfg, conn, &socketfd) < 0)
{
- return -1;
+ printf("ERROR! init_conn() failed; retrying in %d s\n",
+ MQTTC_RECONNECT_DELAY_S);
+ sleep(MQTTC_RECONNECT_DELAY_S);
}
- mqtterr = mqtt_init(&mqtt_conn.client, socketfd,
- mqtt_conn.sendbuf, sizeof(mqtt_conn.sendbuf),
- mqtt_conn.recvbuf, sizeof(mqtt_conn.recvbuf),
- NULL);
+ printf("Success: Connected to broker!\n");
+
+ /* Reinitialize the client with the fresh socket/SSL handle. */
+
+ mqtt_reinit(client, socketfd,
+ conn->sendbuf, sizeof(conn->sendbuf),
+ conn->recvbuf, sizeof(conn->recvbuf));
+
+ /* Send the connection request to the broker. */
+
+ mqtterr = mqtt_connect(client, cfg->id, NULL, NULL, 0, NULL, NULL,
+ cfg->flags, cfg->tmo);
if (mqtterr != MQTT_OK)
{
- printf("ERROR! mqtt_init() failed.\n");
- goto err_with_conn;
+ printf("ERROR! mqtt_connect() failed: %s\n", mqtt_error_str(mqtterr));
+ return;
}
- mqtterr = mqtt_connect(&mqtt_conn.client, mqtt_cfg.id,
- NULL, /* Will topic */
- NULL, /* Will message */
- 0, /* Will message size */
- mqtt_cfg.user, /* User name */
- mqtt_cfg.pass, /* Password */
- mqtt_cfg.flags, mqtt_cfg.tmo);
+ /* (Re)subscribe to the configured topic. */
+ mqtterr = mqtt_subscribe(client, cfg->topic, cfg->qos);
if (mqtterr != MQTT_OK)
{
- printf("ERROR! mqtt_connect() failed\n");
- goto err_with_conn;
+ printf("ERROR! mqtt_subscribe() failed: %s\n",
+ mqtt_error_str(mqtterr));
}
+}
- if (mqtt_conn.client.error != MQTT_OK)
+int main(int argc, FAR char * argv[])
+{
+ struct mqtt_conn_context_s mqtt_conn =
{
- printf("error: %s\n", mqtt_error_str(mqtt_conn.client.error));
- goto err_with_conn;
- }
- else
+ 0
+ };
+
+ struct mqttc_reconnect_s reconnect_state;
+ int timeout = 100;
+ enum MQTTErrors mqtterr;
+ pthread_attr_t attr;
+ pthread_t thrdid;
+ struct mqttc_cfg_s mqtt_cfg =
{
- printf("Success: Connected to broker!\n");
- }
+ .host = "broker.hivemq.com",
+ #ifndef MQTT_USE_MBEDTLS
+ .port = "1883",
+ #else
+ .port = "8883",
+ #endif
+ .topic = "test",
+ .msg = "NULL",
+ .flags = MQTT_CONNECT_CLEAN_SESSION,
+ .tmo = 400,
+ .id = NULL,
+ .user = NULL,
+ .pass = NULL,
+ .qos = MQTT_PUBLISH_QOS_0
+ };
+
+ parsearg(argc, argv, &mqtt_cfg);
+
+ /* Wait for the network to come up so the first connection attempt does not
+ * fail with ENETUNREACH
+ */
+
+ wait_for_network();
+
+ /* Bundle the configuration and connection context for the reconnect
+ * callback. mqtt_init_reconnect() leaves the client in the
+ * MQTT_ERROR_INITIAL_RECONNECT state; the first mqtt_sync() (driven by the
+ * refresher thread) invokes reconnect_client(), which establishes the
+ * transport, calls mqtt_reinit() and then mqtt_connect()/mqtt_subscribe().
+ */
+
+ reconnect_state.cfg = &mqtt_cfg;
+ reconnect_state.conn = &mqtt_conn;
+
+ mqtt_init_reconnect(&mqtt_conn.client,
+ reconnect_client, &reconnect_state,
+ publish_callback);
/* Start a thread to refresh the client (handle egress and ingress client
* traffic)
@@ -574,7 +768,10 @@ int main(int argc, FAR char *argv[])
goto err_with_conn;
}
- /* Wait for MQTT ACK or time-out */
+ /* Wait for the initial MQTT connection to be acknowledged or time-out.
+ * On time-out we keep going anyway: the reconnect callback will continue
+ * retrying in the background.
+ */
while (!mqtt_conn.client.event_connect && --timeout > 0)
{
@@ -583,32 +780,21 @@ int main(int argc, FAR char *argv[])
if (timeout == 0)
{
- goto err_with_thrd;
+ printf("WARNING! Initial connection timed out; "
+ "reconnect will keep retrying.\n");
}
-
- while (n--)
+ else
{
- mqtterr = mqtt_publish(&mqtt_conn.client, mqtt_cfg.topic,
- mqtt_cfg.msg, strlen(mqtt_cfg.msg),
- mqtt_cfg.qos);
- if (mqtterr != MQTT_OK)
- {
- printf("ERROR! mqtt_publish() failed\n");
- goto err_with_thrd;
- }
+ printf("Success: Connected to broker!\n");
+ }
- if (mqtt_conn.client.error != MQTT_OK)
- {
- printf("error: %s\n", mqtt_error_str(mqtt_conn.client.error));
- goto err_with_thrd;
- }
- else
- {
- printf("Success: Published to broker!\n");
- }
+ printf("Listening for %s.\nType q to exit.\n\n", mqtt_cfg.topic);
- sleep(5);
- }
+ /* Wait for messages */
+
+ while (fgetc(stdin) != 'q');
+
+ /* Disconnect */
printf("\nDisconnecting from %s\n\n", mqtt_cfg.host);
mqtterr = mqtt_disconnect(&mqtt_conn.client);
@@ -625,7 +811,6 @@ int main(int argc, FAR char *argv[])
mqtt_sync(&mqtt_conn.client);
-err_with_thrd:
pthread_cancel(thrdid);
err_with_conn:
@@ -633,4 +818,3 @@ err_with_conn:
return 0;
}
-
