diff options
Diffstat (limited to 'src/main.c')
| -rw-r--r-- | src/main.c | 603 |
1 files changed, 603 insertions, 0 deletions
diff --git a/src/main.c b/src/main.c new file mode 100644 index 0000000..570c01d --- /dev/null +++ b/src/main.c @@ -0,0 +1,603 @@ +/* + * Copyright (c) 2018 Nordic Semiconductor ASA + * + * SPDX-License-Identifier: LicenseRef-Nordic-5-Clause + */ + +#include <zephyr/kernel.h> +#include <stdio.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" + +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); + client->password = NULL; + client->user_name = NULL; + 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; +} + +void main(void) +{ + int err; + uint32_t connect_attempt = 0; + + LOG_INF("The MQTT simple sample started"); + +#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) { + 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; +} |