From 26e8417fd2af0d68fb8588ccb62a49ffd2eb3b9b Mon Sep 17 00:00:00 2001 From: Mihai Pacuraru Date: Wed, 8 Jul 2026 14:30:00 +0300 Subject: [PATCH] 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 --- 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_sub.c | 820 ++++++++++++++++++++++++++++++++++ 6 files changed, 900 insertions(+), 30 deletions(-) create mode 100644 examples/mqttc/cert.inc create mode 100644 examples/mqttc/mqttc_sub.c diff --git a/examples/mqttc/CMakeLists.txt b/examples/mqttc/CMakeLists.txt index 56771255847..6e4ee010a84 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 adf91b72692..61a2e6ba19e 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 9aacf67f6c8..45a2d9f044c 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 00000000000..542602db5e4 --- /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 306df6dcc36..d979fb24ab4 100644 --- a/examples/mqttc/mqttc_pub.c +++ b/examples/mqttc/mqttc_pub.c @@ -24,6 +24,7 @@ * Included Files ****************************************************************************/ +#include #include #include #include @@ -42,39 +43,12 @@ # include # include +#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_sub.c b/examples/mqttc/mqttc_sub.c new file mode 100644 index 00000000000..d7ad1858ad3 --- /dev/null +++ b/examples/mqttc/mqttc_sub.c @@ -0,0 +1,820 @@ +/**************************************************************************** + * apps/examples/mqttc/mqttc_sub.c + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The + * ASF licenses this file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + * + ****************************************************************************/ + +/**************************************************************************** + * Included Files + ****************************************************************************/ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef MQTT_USE_MBEDTLS +# include + +# include +# include +# include +# include +# include + +# include "cert.inc" + +/**************************************************************************** + * Pre-processor Definitions + ****************************************************************************/ + +# 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 + ****************************************************************************/ + +struct mqttc_cfg_s +{ + FAR const char *host; + FAR const char *port; + FAR const char *topic; + FAR const char *msg; + FAR const char *id; + FAR const char *user; + FAR const char *pass; + #ifdef MQTT_USE_MBEDTLS + FAR const char *ca_file; + #endif + uint32_t tmo; + uint8_t flags; + uint8_t qos; +}; + +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 + 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); +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 + * + * Description: + * The client's refresher. This function triggers back-end routines to + * handle ingress/egress traffic to the broker. + * + ****************************************************************************/ + +static FAR void *client_refresher(FAR void *data) +{ + while (1) + { + mqtt_sync((FAR struct mqtt_client *)data); + usleep(100000U); + } + + 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 + * + * Description: + * Parse command line arguments. + * + ****************************************************************************/ + +static void parsearg(int argc, FAR char *argv[], + FAR struct mqttc_cfg_s *cfg) +{ + int opt; + + while ((opt = getopt(argc, argv, "h:p:t:q:" EXTRA_OPT)) != ERROR) + { + switch (opt) + { + case 'h': + cfg->host = optarg; + break; + + case 'p': + cfg->port = optarg; + break; + + case 't': + cfg->topic = optarg; + break; + + case 'q': + switch (strtol(optarg, NULL, 10)) + { + case 0L: + cfg->qos = 0; + break; + case 1L: + cfg->qos = 1; + break; + case 2L: + cfg->qos = 2; + break; + } + break; +#ifdef MQTT_USE_MBEDTLS + case 'c': + cfg->ca_file = optarg; + break; +#endif + + default: + fprintf(stderr, "ERROR: Unrecognized option\n"); + break; + } + } +} + +/**************************************************************************** + * Name: init_conn + * + * Description: + * Resolve server's name and try to establish a connection. + * + ****************************************************************************/ + +static int init_conn(FAR const struct mqttc_cfg_s *cfg, + FAR struct mqtt_conn_context_s *conn, + FAR mqtt_pal_socket_handle *socketfd) +{ +#ifdef MQTT_USE_MBEDTLS + FAR mbedtls_net_context *net_ctx = &conn->net_ctx; + FAR mbedtls_ssl_context *ssl_ctx = &conn->ssl_ctx; + FAR mbedtls_ssl_config *ssl_conf = &conn->ssl_conf; + FAR mbedtls_x509_crt *ca_crt = &conn->ca_crt; + FAR mbedtls_entropy_context *entropy = &conn->entropy; + FAR mbedtls_ctr_drbg_context *ctr_drbg = &conn->ctr_drbg; + uint32_t result; + + FAR const unsigned char *additional = (FAR const unsigned char *)"MQTT-C"; + size_t additional_len = 6; +#else + struct addrinfo hints; + FAR struct addrinfo *servinfo; + FAR struct addrinfo *itr; + int fd; +#endif + int ret; + + printf("Connecting to %s:%s...\n", cfg->host, cfg->port); + +#ifdef MQTT_USE_MBEDTLS + mbedtls_entropy_init(entropy); + mbedtls_ctr_drbg_init(ctr_drbg); + ret = mbedtls_ctr_drbg_seed(ctr_drbg, mbedtls_entropy_func, entropy, + additional, additional_len); + if (ret != 0) + { + printf("ERROR! mbedtls_ctr_drbg_seed() failed: %d\n", ret); + goto err_with_ctr_drbg; + } + + mbedtls_x509_crt_init(ca_crt); + if (cfg->ca_file != NULL) + { + ret = mbedtls_x509_crt_parse_file(ca_crt, cfg->ca_file); + } + 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)); + } + + if (ret != 0) + { + printf("ERROR! mbedtls_x509_crt_parse_file() failed: %d\n", ret); + goto err_with_x509_crt; + } + + mbedtls_ssl_config_init(ssl_conf); + ret = mbedtls_ssl_config_defaults(ssl_conf, MBEDTLS_SSL_IS_CLIENT, + MBEDTLS_SSL_TRANSPORT_STREAM, + MBEDTLS_SSL_PRESET_DEFAULT); + if (ret != 0) + { + printf("ERROR! mbedtls_ssl_config_defaults() failed: %d\n", ret); + goto err_with_ssl_conf; + } + + mbedtls_ssl_conf_ca_chain(ssl_conf, ca_crt, NULL); + mbedtls_ssl_conf_authmode(ssl_conf, MBEDTLS_SSL_VERIFY_OPTIONAL); + mbedtls_ssl_conf_rng(ssl_conf, mbedtls_ctr_drbg_random, ctr_drbg); + + mbedtls_net_init(net_ctx); + ret = mbedtls_net_connect(net_ctx, cfg->host, cfg->port, + MBEDTLS_NET_PROTO_TCP); + if (ret != 0) + { + printf("ERROR! mbedtls_net_connect() failed: %d\n", ret); + goto err_with_net; + } + + ret = mbedtls_net_set_nonblock(net_ctx); + if (ret != 0) + { + printf("ERROR! mbedtls_net_set_nonblock() failed: %d\n", ret); + goto err_with_net; + } + + mbedtls_ssl_init(ssl_ctx); + ret = mbedtls_ssl_setup(ssl_ctx, ssl_conf); + if (ret != 0) + { + printf("ERROR! mbedtls_ssl_setup() failed: %d\n", ret); + goto err_with_ssl; + } + + ret = mbedtls_ssl_set_hostname(ssl_ctx, cfg->host); + if (ret != 0) + { + printf("ERROR! mbedtls_ssl_set_hostname() failed: %d\n", ret); + goto err_with_ssl; + } + + mbedtls_ssl_set_bio(ssl_ctx, net_ctx, + mqttc_net_send, mqttc_net_recv, NULL); + + for (; ; ) + { + uint32_t want = 0; + ret = mbedtls_ssl_handshake(ssl_ctx); + if (ret == MBEDTLS_ERR_SSL_WANT_READ) + { + want |= MBEDTLS_NET_POLL_READ; + } + else if (ret == MBEDTLS_ERR_SSL_WANT_WRITE) + { + want |= MBEDTLS_NET_POLL_WRITE; + } + else + { + break; + } + + ret = mbedtls_net_poll(net_ctx, want, -1); + if (ret < 0) + { + printf("ERROR! mbedtls_net_poll() failed: %d\n", ret); + goto err_with_ssl; + } + } + + if (ret != 0) + { + printf("ERROR! mbedtls_ssl_handshake() failed: %d\n", ret); + goto err_with_ssl; + } + + result = mbedtls_ssl_get_verify_result(ssl_ctx); + if (result != 0) + { + if (result == 0xffffffff) + { + printf("ERROR! mbedtls_ssl_get_verify_result() failed\n"); + goto err_with_ssl; + } + else + { +#if defined(MBEDTLS_X509_REMOVE_INFO) + const char *buf = ""; +#else + char buf[512]; + mbedtls_x509_crt_verify_info(buf, sizeof(buf), "\t", result); +#endif + printf("Certificate verification failed (%0" PRIx32 ")\n%s\n", + result, buf); + +#ifndef CONFIG_EXAMPLES_MQTTC_ALLOW_UNVERIFIED_TLS + goto err_with_ssl; +#endif + } + } + + *socketfd = ssl_ctx; + + return 0; + +err_with_ssl: + mbedtls_ssl_free(ssl_ctx); +err_with_net: + mbedtls_net_free(net_ctx); +err_with_ssl_conf: + mbedtls_ssl_config_free(ssl_conf); +err_with_x509_crt: + mbedtls_x509_crt_free(ca_crt); +err_with_ctr_drbg: + mbedtls_ctr_drbg_free(ctr_drbg); + mbedtls_entropy_free(entropy); + + return -1; +#else + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; + + ret = getaddrinfo(cfg->host, cfg->port, &hints, &servinfo); + if (ret != 0) + { + printf("ERROR! getaddrinfo() failed: %s\n", gai_strerror(ret)); + return -1; + } + + itr = servinfo; + do + { + fd = socket(itr->ai_family, itr->ai_socktype, itr->ai_protocol); + if (fd < 0) + { + continue; + } + + ret = connect(fd, itr->ai_addr, itr->ai_addrlen); + if (ret == 0) + { + break; + } + + close(fd); + fd = -1; + } + while ((itr = itr->ai_next) != NULL); + + freeaddrinfo(servinfo); + + if (fd < 0) + { + printf("ERROR! Couldn't create socket\n"); + return -1; + } + + ret = fcntl(fd, F_GETFL, 0); + if (ret < 0) + { + close(fd); + printf("ERROR! fcntl() F_GETFL failed, errno: %d\n", errno); + return -1; + } + + ret = fcntl(fd, F_SETFL, ret | O_NONBLOCK); + if (ret < 0) + { + close(fd); + printf("ERROR! fcntl() F_SETFL failed, errno: %d\n", errno); + return -1; + } + + *socketfd = fd; + + return 0; +#endif +} + +/**************************************************************************** + * Name: close_conn + * + * Description: + * Shut down connection to server established by init_conn. + * + ****************************************************************************/ + +static void close_conn(FAR struct mqtt_conn_context_s *conn) +{ +#ifndef MQTT_USE_MBEDTLS + close(conn->client.socketfd); +#else + mbedtls_net_free(&conn->net_ctx); + mbedtls_ssl_free(&conn->ssl_ctx); + mbedtls_ssl_config_free(&conn->ssl_conf); + mbedtls_x509_crt_free(&conn->ca_crt); + mbedtls_ctr_drbg_free(&conn->ctr_drbg); + mbedtls_entropy_free(&conn->entropy); +#endif +} + +/**************************************************************************** + * 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). + * + ****************************************************************************/ + +static void reconnect_client(FAR struct mqtt_client *client, + FAR void **state) +{ + 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; + enum MQTTErrors mqtterr; + + /* 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) + { + printf("reconnect_client: reconnecting after error \"%s\"\n", + mqtt_error_str(client->error)); + close_conn(conn); + } + + /* (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. + */ + + while (init_conn(cfg, conn, &socketfd) < 0) + { + printf("ERROR! init_conn() failed; retrying in %d s\n", + MQTTC_RECONNECT_DELAY_S); + sleep(MQTTC_RECONNECT_DELAY_S); + } + + 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_connect() failed: %s\n", mqtt_error_str(mqtterr)); + return; + } + + /* (Re)subscribe to the configured topic. */ + + mqtterr = mqtt_subscribe(client, cfg->topic, cfg->qos); + if (mqtterr != MQTT_OK) + { + printf("ERROR! mqtt_subscribe() failed: %s\n", + mqtt_error_str(mqtterr)); + } +} + +int main(int argc, FAR char * argv[]) +{ + struct mqtt_conn_context_s mqtt_conn = + { + 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 = + { + .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) + */ + + if (pthread_attr_init(&attr) != 0) + { + printf("ERROR! pthread_attr_init() failed.\n"); + goto err_with_conn; + } + + pthread_attr_setstacksize(&attr, CONFIG_EXAMPLES_MQTTC_STACKSIZE); + + if (pthread_create(&thrdid, &attr, client_refresher, &mqtt_conn.client)) + { + printf("ERROR! pthread_create() failed.\n"); + goto err_with_conn; + } + + /* 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) + { + usleep(10000); + } + + if (timeout == 0) + { + printf("WARNING! Initial connection timed out; " + "reconnect will keep retrying.\n"); + } + else + { + printf("Success: Connected to broker!\n"); + } + + printf("Listening for %s.\nType q to exit.\n\n", mqtt_cfg.topic); + + /* Wait for messages */ + + while (fgetc(stdin) != 'q'); + + /* Disconnect */ + + printf("\nDisconnecting from %s\n\n", mqtt_cfg.host); + mqtterr = mqtt_disconnect(&mqtt_conn.client); + if (mqtterr != MQTT_OK) + { + printf("ERROR! mqtt_disconnect() failed\n"); + } + + /* Force sending the DISCONNECT, the thread will be canceled before getting + * the chance to sync this last packet. + * Note however that close() would cleanly close the connection but only + * through TCP (i.e. no MQTT DISCONNECT packet). + */ + + mqtt_sync(&mqtt_conn.client); + + pthread_cancel(thrdid); + +err_with_conn: + close_conn(&mqtt_conn); + + return 0; +}