I've been working on developing an application based on the MQTT_Simple example with TWIM & TWIS added to it. The application seems to have a 60 second delay before the "while (1)" loop in the "do_connect:" section of the main function and I can't identify where the delay is located in the code. The delay is skipped and the program goes through the while loop when an MQTT message is received, but if an MQTT message is not received there is the 60 second delay. I would like to continuously run additional lines of code independent of receiving MQTT messages or this delay. Any advice would be greatly appreciated. My code is as follows:
#include <zephyr/kernel.h>
#include <stdio.h>
#include <stdint.h>
#include <zephyr/drivers/uart.h>
#include <string.h>
#include <zephyr/random/rand32.h>
#include <zephyr/net/mqtt.h>
#include <zephyr/net/socket.h>
#include <zephyr/shell/shell.h>
#include <hw_id.h>
#if defined(CONFIG_NRF_MODEM_LIB)
#include <nrf_modem_at.h>
#endif /* CONFIG_NRF_MODEM_LIB */
#include <modem/lte_lc.h>
#include <zephyr/logging/log.h>
#if defined(CONFIG_MODEM_KEY_MGMT)
#include <modem/modem_key_mgmt.h>
#endif
#include <dk_buttons_and_leds.h>
#include "certificates.h"
#include <nrfx_twim.h>
#include <nrfx_twis.h>
/**
* @defgroup nrfx_twim_twis_non_blocking_example Non-blocking TWIM with TWIS example
* @{
* @ingroup nrfx_twim_twis_examples
*
* @brief Example showing the functionality of nrfx_twim and nrfx_twis drivers operating in the
* non-blocking mode.
*
* @details Application initializes nrfx_twim and nrfx_twis drivers, then sends the specified message
* ( @ref MSG_TO_SEND ) from master to slave and back using TWIM and TWIS peripherals.
* To perform this operation different buffers are used and data is transferred as follows:
* @ref m_tx_buffer_master -> @ref m_rx_buffer_slave -> @ref m_rx_buffer_master
* The @ref twim_handler() and @ref twis_handler() are executed with relevant log messages.
*/
#define LOOPBACK_PIN_1A 3
#define LOOPBACK_PIN_1B 31
#define LOOPBACK_PIN_2A 4
#define LOOPBACK_PIN_2B 30
/** @brief Symbol specifying pin number of master SCL. */
#define MASTER_SCL_PIN LOOPBACK_PIN_1A
/** @brief Symbol specifying pin number of master SDA. */
#define MASTER_SDA_PIN LOOPBACK_PIN_2A
/** @brief Symbol specifying pin number of slave SCL. */
#define SLAVE_SCL_PIN LOOPBACK_PIN_1B
/** @brief Symbol specifying pin number of slave SDA. */
#define SLAVE_SDA_PIN LOOPBACK_PIN_2B
#define TWIM_INST_IDX 1
#define TWIS_INST_IDX 2
/** @brief Symbol specifying slave address on TWI bus. */
#define SLAVE_ADDR 0x0B
/** @brief Symbol specifying message to be sent via TWIM data transfer.*/
#define MSG_TO_SEND "Nordic Semiconductor"
/** @brief Structure containing TWIS driver instance. */
static nrfx_twis_t twis_inst = NRFX_TWIS_INSTANCE(TWIS_INST_IDX);
/** @brief Structure containing TWIM driver instance. */
static nrfx_twim_t twim_inst = NRFX_TWIM_INSTANCE(TWIM_INST_IDX);
/** @brief TWIM transmit buffer initialized with @ref MSG_TO_SEND. */
static uint8_t m_tx_buffer_master[] = MSG_TO_SEND;
/** @brief TWIS recieve buffer for a message from TWIM. */
static uint8_t m_rx_buffer_slave[sizeof(MSG_TO_SEND)];
/** @brief TWIM recieve buffer for a message from TWIS. */
static uint8_t m_rx_buffer_master[sizeof(MSG_TO_SEND)];
LOG_MODULE_REGISTER(mqtt_simple, CONFIG_MQTT_SIMPLE_LOG_LEVEL);
/* Buffers for MQTT client. */
static uint8_t rx_buffer[CONFIG_MQTT_MESSAGE_BUFFER_SIZE];
static uint8_t tx_buffer[CONFIG_MQTT_MESSAGE_BUFFER_SIZE];
static uint8_t payload_buf[CONFIG_MQTT_PAYLOAD_BUFFER_SIZE];
/* The mqtt client struct */
static struct mqtt_client client;
/* MQTT Broker details. */
static struct sockaddr_storage broker;
/* File descriptor */
static struct pollfd fds;
#if defined(CONFIG_MQTT_LIB_TLS)
static int certificates_provision(void)
{
int err = 0;
LOG_INF("Provisioning certificates");
#if defined(CONFIG_NRF_MODEM_LIB) && defined(CONFIG_MODEM_KEY_MGMT)
err = modem_key_mgmt_write(CONFIG_MQTT_TLS_SEC_TAG,
MODEM_KEY_MGMT_CRED_TYPE_CA_CHAIN,
CA_CERTIFICATE,
strlen(CA_CERTIFICATE));
if (err) {
LOG_ERR("Failed to provision CA certificate: %d", err);
return err;
}
#elif defined(CONFIG_BOARD_QEMU_X86) && defined(CONFIG_NET_SOCKETS_SOCKOPT_TLS)
err = tls_credential_add(CONFIG_MQTT_TLS_SEC_TAG,
TLS_CREDENTIAL_CA_CERTIFICATE,
CA_CERTIFICATE,
sizeof(CA_CERTIFICATE));
if (err) {
LOG_ERR("Failed to register CA certificate: %d", err);
return err;
}
#endif
return err;
}
#endif /* defined(CONFIG_MQTT_LIB_TLS) */
/**@brief Function to print strings without null-termination
*/
static void data_print(uint8_t *prefix, uint8_t *data, size_t len)
{
char buf[len + 1];
memcpy(buf, data, len);
buf[len] = 0;
LOG_INF("%s%s", (char *)prefix, (char *)buf);
}
/**@brief Function to publish data on the configured topic
*/
static int data_publish(struct mqtt_client *c, enum mqtt_qos qos,
uint8_t *data, size_t len)
{
struct mqtt_publish_param param;
param.message.topic.qos = qos;
param.message.topic.topic.utf8 = CONFIG_MQTT_PUB_TOPIC;
param.message.topic.topic.size = strlen(CONFIG_MQTT_PUB_TOPIC);
param.message.payload.data = data;
param.message.payload.len = len;
param.message_id = sys_rand32_get();
param.dup_flag = 0;
param.retain_flag = 0;
data_print("Publishing: ", data, len);
LOG_INF("to topic: %s len: %u",
CONFIG_MQTT_PUB_TOPIC,
(unsigned int)strlen(CONFIG_MQTT_PUB_TOPIC));
return mqtt_publish(c, ¶m);
}
/**@brief Function to subscribe to the configured topic
*/
static int subscribe(void)
{
struct mqtt_topic subscribe_topic = {
.topic = {
.utf8 = CONFIG_MQTT_SUB_TOPIC,
.size = strlen(CONFIG_MQTT_SUB_TOPIC)
},
.qos = MQTT_QOS_1_AT_LEAST_ONCE
};
const struct mqtt_subscription_list subscription_list = {
.list = &subscribe_topic,
.list_count = 1,
.message_id = 1234
};
LOG_INF("Subscribing to: %s len %u", CONFIG_MQTT_SUB_TOPIC,
(unsigned int)strlen(CONFIG_MQTT_SUB_TOPIC));
return mqtt_subscribe(&client, &subscription_list);
}
/**@brief Function to read the published payload.
*/
static int publish_get_payload(struct mqtt_client *c, size_t length)
{
int ret;
int err = 0;
/* Return an error if the payload is larger than the payload buffer.
* Note: To allow new messages, we have to read the payload before returning.
*/
if (length > sizeof(payload_buf)) {
err = -EMSGSIZE;
}
/* Truncate payload until it fits in the payload buffer. */
while (length > sizeof(payload_buf)) {
ret = mqtt_read_publish_payload_blocking(
c, payload_buf, (length - sizeof(payload_buf)));
if (ret == 0) {
return -EIO;
} else if (ret < 0) {
return ret;
}
length -= ret;
}
ret = mqtt_readall_publish_payload(c, payload_buf, length);
if (ret) {
return ret;
}
return err;
}
/**@brief MQTT client event handler
*/
void mqtt_evt_handler(struct mqtt_client *const c,
const struct mqtt_evt *evt)
{
int err;
switch (evt->type) {
case MQTT_EVT_CONNACK:
if (evt->result != 0) {
LOG_ERR("MQTT connect failed: %d", evt->result);
break;
}
LOG_INF("MQTT client connected");
subscribe();
break;
case MQTT_EVT_DISCONNECT:
LOG_INF("MQTT client disconnected: %d", evt->result);
break;
case MQTT_EVT_PUBLISH: {
const struct mqtt_publish_param *p = &evt->param.publish;
LOG_INF("MQTT PUBLISH result=%d len=%d",
evt->result, p->message.payload.len);
err = publish_get_payload(c, p->message.payload.len);
if (p->message.topic.qos == MQTT_QOS_1_AT_LEAST_ONCE) {
const struct mqtt_puback_param ack = {
.message_id = p->message_id
};
/* Send acknowledgment. */
mqtt_publish_qos1_ack(&client, &ack);
}
if (err >= 0) {
data_print("Received: ", payload_buf,
p->message.payload.len);
/* Echo back received data */
data_publish(&client, MQTT_QOS_1_AT_LEAST_ONCE,
payload_buf, p->message.payload.len);
} else if (err == -EMSGSIZE) {
LOG_ERR("Received payload (%d bytes) is larger than the payload buffer "
"size (%d bytes).",
p->message.payload.len, sizeof(payload_buf));
} else {
LOG_ERR("publish_get_payload failed: %d", err);
LOG_INF("Disconnecting MQTT client...");
err = mqtt_disconnect(c);
if (err) {
LOG_ERR("Could not disconnect: %d", err);
}
}
} break;
case MQTT_EVT_PUBACK:
if (evt->result != 0) {
LOG_ERR("MQTT PUBACK error: %d", evt->result);
break;
}
LOG_INF("PUBACK packet id: %u", evt->param.puback.message_id);
break;
case MQTT_EVT_SUBACK:
if (evt->result != 0) {
LOG_ERR("MQTT SUBACK error: %d", evt->result);
break;
}
LOG_INF("SUBACK packet id: %u", evt->param.suback.message_id);
break;
case MQTT_EVT_PINGRESP:
if (evt->result != 0) {
LOG_ERR("MQTT PINGRESP error: %d", evt->result);
}
break;
default:
LOG_INF("Unhandled MQTT event type: %d", evt->type);
break;
}
}
/**@brief Resolves the configured hostname and
* initializes the MQTT broker structure
*/
static int broker_init(void)
{
int err;
struct addrinfo *result;
struct addrinfo *addr;
struct addrinfo hints = {
.ai_family = AF_INET,
.ai_socktype = SOCK_STREAM
};
err = getaddrinfo(CONFIG_MQTT_BROKER_HOSTNAME, NULL, &hints, &result);
if (err) {
LOG_ERR("getaddrinfo failed: %d", err);
return -ECHILD;
}
addr = result;
/* Look for address of the broker. */
while (addr != NULL) {
/* IPv4 Address. */
if (addr->ai_addrlen == sizeof(struct sockaddr_in)) {
struct sockaddr_in *broker4 =
((struct sockaddr_in *)&broker);
char ipv4_addr[NET_IPV4_ADDR_LEN];
broker4->sin_addr.s_addr =
((struct sockaddr_in *)addr->ai_addr)
->sin_addr.s_addr;
broker4->sin_family = AF_INET;
broker4->sin_port = htons(CONFIG_MQTT_BROKER_PORT);
inet_ntop(AF_INET, &broker4->sin_addr.s_addr,
ipv4_addr, sizeof(ipv4_addr));
LOG_INF("IPv4 Address found %s", ipv4_addr);
break;
} else {
LOG_ERR("ai_addrlen = %u should be %u or %u",
(unsigned int)addr->ai_addrlen,
(unsigned int)sizeof(struct sockaddr_in),
(unsigned int)sizeof(struct sockaddr_in6));
}
addr = addr->ai_next;
}
/* Free the address. */
freeaddrinfo(result);
return err;
}
#define RANDOM_LEN 10
#define CLIENT_ID_LEN sizeof(CONFIG_BOARD) + 1 + RANDOM_LEN
/* Function to get the client id */
static const uint8_t* client_id_get(void)
{
static uint8_t client_id[MAX(sizeof(CONFIG_MQTT_CLIENT_ID),
CLIENT_ID_LEN)];
if (strlen(CONFIG_MQTT_CLIENT_ID) > 0) {
snprintf(client_id, sizeof(client_id), "%s",
CONFIG_MQTT_CLIENT_ID);
goto exit;
}
char hw_id_buf[HW_ID_LEN] = {0};
int err = hw_id_get(hw_id_buf, ARRAY_SIZE(hw_id_buf));
if (!err) {
snprintf(client_id, sizeof(client_id), "nrf-%s",
hw_id_buf);
goto exit;
}
LOG_ERR("failed to retrieve HW ID, err: %d", err);
uint32_t id = sys_rand32_get();
snprintf(client_id, sizeof(client_id), "%s-%010u", CONFIG_BOARD, id);
exit:
LOG_DBG("client_id = %s", (char *)client_id);
return client_id;
}
/**@brief Initialize the MQTT client structure
*/
static int client_init(struct mqtt_client *client)
{
int err;
mqtt_client_init(client);
err = broker_init();
if (err) {
LOG_ERR("Failed to initialize broker connection");
return err;
}
/* MQTT client configuration */
client->broker = &broker;
client->evt_cb = mqtt_evt_handler;
client->client_id.utf8 = client_id_get();
client->client_id.size = strlen(client->client_id.utf8);
struct mqtt_utf8 pass,name;
pass.size = (uint32_t)strlen("****************");
pass.utf8 =(uint8_t*)"****************";
printk("pass: %d,%s\n",pass.size,pass.utf8);
name.size = (uint32_t) strlen("****************");
name.utf8 = (uint8_t*)"****************";
printk("name: %d,%s\n",name.size,name.utf8);
client->password = &pass;
client->user_name = &name;
client->protocol_version = MQTT_VERSION_3_1_1;
/* MQTT buffers configuration */
client->rx_buf = rx_buffer;
client->rx_buf_size = sizeof(rx_buffer);
client->tx_buf = tx_buffer;
client->tx_buf_size = sizeof(tx_buffer);
/* MQTT transport configuration */
#if defined(CONFIG_MQTT_LIB_TLS)
struct mqtt_sec_config *tls_cfg = &(client->transport).tls.config;
static sec_tag_t sec_tag_list[] = { CONFIG_MQTT_TLS_SEC_TAG };
LOG_INF("TLS enabled");
client->transport.type = MQTT_TRANSPORT_SECURE;
tls_cfg->peer_verify = CONFIG_MQTT_TLS_PEER_VERIFY;
tls_cfg->cipher_count = 0;
tls_cfg->cipher_list = NULL;
tls_cfg->sec_tag_count = ARRAY_SIZE(sec_tag_list);
tls_cfg->sec_tag_list = sec_tag_list;
tls_cfg->hostname = CONFIG_MQTT_BROKER_HOSTNAME;
#if defined(CONFIG_NRF_MODEM_LIB)
tls_cfg->session_cache = IS_ENABLED(CONFIG_MQTT_TLS_SESSION_CACHING) ?
TLS_SESSION_CACHE_ENABLED :
TLS_SESSION_CACHE_DISABLED;
#else
/* TLS session caching is not supported by the Zephyr network stack */
tls_cfg->session_cache = TLS_SESSION_CACHE_DISABLED;
#endif
#else
client->transport.type = MQTT_TRANSPORT_NON_SECURE;
#endif
return err;
}
/**@brief Initialize the file descriptor structure used by poll.
*/
static int fds_init(struct mqtt_client *c)
{
if (c->transport.type == MQTT_TRANSPORT_NON_SECURE) {
fds.fd = c->transport.tcp.sock;
} else {
#if defined(CONFIG_MQTT_LIB_TLS)
fds.fd = c->transport.tls.sock;
#else
return -ENOTSUP;
#endif
}
fds.events = POLLIN;
return 0;
}
#if defined(CONFIG_DK_LIBRARY)
static void button_handler(uint32_t button_states, uint32_t has_changed)
{
if (has_changed & button_states &
BIT(CONFIG_BUTTON_EVENT_BTN_NUM - 1)) {
int ret;
ret = data_publish(&client,
MQTT_QOS_1_AT_LEAST_ONCE,
CONFIG_BUTTON_EVENT_PUBLISH_MSG,
sizeof(CONFIG_BUTTON_EVENT_PUBLISH_MSG)-1);
if (ret) {
LOG_ERR("Publish failed: %d", ret);
}
}
}
#endif
static int shell_mqtt_publish(const struct shell *shell, size_t argc, char **argv)
{
ARG_UNUSED(argc);
ARG_UNUSED(argv);
int ret;
ret = data_publish(&client,
MQTT_QOS_1_AT_LEAST_ONCE,
CONFIG_BUTTON_EVENT_PUBLISH_MSG,
sizeof(CONFIG_BUTTON_EVENT_PUBLISH_MSG) - 1);
if (ret) {
LOG_ERR("Publish failed: %d", ret);
}
return ret;
}
SHELL_STATIC_SUBCMD_SET_CREATE(mqtt_sub,
SHELL_CMD(publish, NULL, "Publish data to configured publish topic",
shell_mqtt_publish),
SHELL_SUBCMD_SET_END /* Array terminated. */
);
SHELL_CMD_REGISTER(mqtt, &mqtt_sub, "MQTT operations", NULL);
/**@brief Configures modem to provide LTE link. Blocks until link is
* successfully established.
*/
static int modem_configure(void)
{
#if defined(CONFIG_LTE_LINK_CONTROL)
/* Turn off LTE power saving features for a more responsive demo. Also,
* request power saving features before network registration. Some
* networks rejects timer updates after the device has registered to the
* LTE network.
*/
LOG_INF("Disabling PSM and eDRX");
lte_lc_psm_req(false);
lte_lc_edrx_req(false);
if (IS_ENABLED(CONFIG_LTE_AUTO_INIT_AND_CONNECT)) {
/* Do nothing, modem is already turned on
* and connected.
*/
} else {
int err;
LOG_INF("LTE Link Connecting...");
err = lte_lc_init_and_connect();
if (err) {
LOG_INF("Failed to establish LTE connection: %d", err);
return err;
}
LOG_INF("LTE Link Connected!");
}
#endif /* defined(CONFIG_LTE_LINK_CONTROL) */
return 0;
}
/**
* @brief Function for handling TWIM driver events.
*
* @param[in] p_event Event information structure.
* @param[in] p_context General purpose parameter set during initialization of the TWIM.
* This parameter can be used to pass additional information to the
* handler function. In this example @p p_context is used to pass address
* of TWI transfer descriptor structure.
*/
static void twim_handler(nrfx_twim_evt_t const * p_event, void * p_context)
{
nrfx_err_t status;
(void)status;
nrfx_twim_xfer_desc_t * twim_desc = p_context;
static bool m_master_done = false;
if (p_event->type == NRFX_TWIM_EVT_DONE)
{
LOG_INF("--> Master event: done - transfer completed");
twim_desc->type = NRFX_TWIM_XFER_RX;
twim_desc->p_primary_buf = m_rx_buffer_master;
if (!m_master_done)
{
status = nrfx_twim_xfer(&twim_inst, twim_desc, 0);
NRFX_ASSERT(status == NRFX_SUCCESS);
m_master_done = true;
}
else
{
LOG_INF("Content of master RX buffer: %s", m_rx_buffer_master);
}
}
else
{
LOG_INF("--> MASTER handler, event: %d.", p_event->type);
}
}
/**
* @brief Function for handling TWIS driver events.
*
* @param[in] p_event Event information structure.
*/
static void twis_handler(nrfx_twis_evt_t const * p_event)
{
nrfx_err_t status;
(void)status;
switch (p_event->type)
{
case NRFX_TWIS_EVT_WRITE_DONE:
LOG_INF("--> Slave event: write done.");
break;
case NRFX_TWIS_EVT_READ_DONE:
LOG_INF("--> Slave event: read done.");
break;
case NRFX_TWIS_EVT_WRITE_REQ:
status = nrfx_twis_rx_prepare(&twis_inst, m_rx_buffer_slave, sizeof(m_rx_buffer_slave));
NRFX_ASSERT(status == NRFX_SUCCESS);
LOG_INF("--> Slave event: write request");
break;
case NRFX_TWIS_EVT_READ_REQ:
status = nrfx_twis_tx_prepare(&twis_inst, m_rx_buffer_slave, sizeof(m_rx_buffer_slave));
NRFX_ASSERT(status == NRFX_SUCCESS);
LOG_INF("--> Slave event: read request");
break;
default:
LOG_INF("--> SLAVE event: %d.", p_event->type);
}
}
void main(void)
{
int err;
uint32_t connect_attempt = 0;
LOG_INF("The MQTT simple sample started");
nrfx_err_t status;
(void)status;
LOG_INF("Starting nrfx_twim_twis non-blocking example.");
nrfx_twim_xfer_desc_t twim_xfer_desc = NRFX_TWIM_XFER_DESC_TX(SLAVE_ADDR,
m_tx_buffer_master,
sizeof(m_tx_buffer_master));
nrfx_twim_config_t twim_config = NRFX_TWIM_DEFAULT_CONFIG(MASTER_SCL_PIN, MASTER_SDA_PIN);
status = nrfx_twim_init(&twim_inst, &twim_config, twim_handler, &twim_xfer_desc);
nrfx_twis_config_t twis_config = NRFX_TWIS_DEFAULT_CONFIG(SLAVE_SCL_PIN,
SLAVE_SDA_PIN,
SLAVE_ADDR);
status = nrfx_twis_init(&twis_inst, &twis_config, twis_handler);
#if defined(__ZEPHYR__)
#define TWIM_INST NRFX_CONCAT_2(NRF_TWIM, TWIM_INST_IDX)
#define TWIM_INST_HANDLER NRFX_CONCAT_3(nrfx_twim_, TWIM_INST_IDX, _irq_handler)
IRQ_DIRECT_CONNECT(NRFX_IRQ_NUMBER_GET(TWIM_INST), IRQ_PRIO_LOWEST, TWIM_INST_HANDLER, 0);
#define TWIS_INST NRFX_CONCAT_2(NRF_TWIS, TWIS_INST_IDX)
#define TWIS_INST_HANDLER NRFX_CONCAT_3(nrfx_twis_, TWIS_INST_IDX, _irq_handler)
IRQ_DIRECT_CONNECT(NRFX_IRQ_NUMBER_GET(TWIS_INST), IRQ_PRIO_LOWEST, TWIS_INST_HANDLER, 0);
#endif
nrfx_twim_enable(&twim_inst);
nrfx_twis_enable(&twis_inst);
status = nrfx_twim_xfer(&twim_inst, &twim_xfer_desc, 0);
#if defined(CONFIG_MQTT_LIB_TLS)
err = certificates_provision();
if (err != 0) {
LOG_ERR("Failed to provision certificates");
return;
}
#endif /* defined(CONFIG_MQTT_LIB_TLS) */
do {
err = modem_configure();
if (err) {
LOG_INF("Retrying in %d seconds",
CONFIG_LTE_CONNECT_RETRY_DELAY_S);
k_sleep(K_SECONDS(CONFIG_LTE_CONNECT_RETRY_DELAY_S));
}
} while (err);
err = client_init(&client);
if (err != 0) {
LOG_ERR("client_init: %d", err);
return;
}
#if defined(CONFIG_DK_LIBRARY)
dk_buttons_init(button_handler);
#endif
do_connect:
if (connect_attempt++ > 0) {
LOG_INF("Reconnecting in %d seconds...",
CONFIG_MQTT_RECONNECT_DELAY_S);
k_sleep(K_SECONDS(CONFIG_MQTT_RECONNECT_DELAY_S));
}
err = mqtt_connect(&client);
if (err != 0) {
LOG_ERR("mqtt_connect %d", err);
goto do_connect;
}
err = fds_init(&client);
if (err != 0) {
LOG_ERR("fds_init: %d", err);
return;
}
while (1) {
LOG_INF("Messagetest");
LOG_INF("Message to send: %s", m_tx_buffer_master);
LOG_INF("Content of master RX buffer: %s", m_rx_buffer_master);
err = poll(&fds, 1, mqtt_keepalive_time_left(&client));
if (err < 0) {
LOG_ERR("poll: %d", errno);
break;
}
err = mqtt_live(&client);
if ((err != 0) && (err != -EAGAIN)) {
LOG_ERR("ERROR: mqtt_live: %d", err);
break;
}
if ((fds.revents & POLLIN) == POLLIN) {
err = mqtt_input(&client);
if (err != 0) {
LOG_ERR("mqtt_input: %d", err);
break;
}
}
if ((fds.revents & POLLERR) == POLLERR) {
LOG_ERR("POLLERR");
break;
}
if ((fds.revents & POLLNVAL) == POLLNVAL) {
LOG_ERR("POLLNVAL");
break;
}
}
LOG_INF("Disconnecting MQTT client...");
err = mqtt_disconnect(&client);
if (err) {
LOG_ERR("Could not disconnect MQTT client: %d", err);
}
goto do_connect;
}
