From b3b9164352b1fc276f0c1388be5e6c44f16906f2 Mon Sep 17 00:00:00 2001 From: Michal Hanus Date: Mon, 2 Jun 2025 09:46:16 +0200 Subject: final1 --- prilohy/nrfdemo_source/src/mqtt_connection.c | 402 +++++++++++++++++++++++++++ 1 file changed, 402 insertions(+) create mode 100644 prilohy/nrfdemo_source/src/mqtt_connection.c (limited to 'prilohy/nrfdemo_source/src/mqtt_connection.c') diff --git a/prilohy/nrfdemo_source/src/mqtt_connection.c b/prilohy/nrfdemo_source/src/mqtt_connection.c new file mode 100644 index 0000000..2410bca --- /dev/null +++ b/prilohy/nrfdemo_source/src/mqtt_connection.c @@ -0,0 +1,402 @@ +#include +#include +#include + +#include +#include +#include +#include +#include + +#include +#include "mqtt_connection.h" + +#if NCS_VERSION_NUMBER < 0x20600 +#include +#else +#include +#endif + +/* 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]; + +/* MQTT Broker details. */ +static struct sockaddr_storage broker; + +LOG_MODULE_DECLARE(mqtt_app); + +/**@brief Function to get the payload of recived data. + */ +static int get_received_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 Function to subscribe to the configured topic + */ +/* Define the function subscribe() to subscribe to a specific topic. */ +static int subscribe(struct mqtt_client *const c) +{ + 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(c, &subscription_list); +} + +/**@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); +} + +int publish_sensor(struct mqtt_client *c, uint8_t* topic, int32_t val1, int32_t val2) +{ + uint8_t buf[20], sgn = (val1 < 0 && val2 < 0) ? '-' : '+'; + snprintf(buf, 20, "%c%i.%03i\0", sgn, labs(val1), labs(val2)/1000); + publish_wrap(c, topic, buf, MIN(20, strlen(buf))); +} + +int publish_int(struct mqtt_client *c, uint8_t* topic, int32_t val) +{ + uint8_t buf[20]; + snprintf(buf, 20, "%d", val); + publish_wrap(c, topic, buf, strlen(buf)); +} + +uint16_t mqtt_cycle = 0; + +int publish_wrap(struct mqtt_client *c, uint8_t* topic, uint8_t *data, size_t len) +{ + struct mqtt_publish_param param; + + param.message.topic.qos = MQTT_QOS_1_AT_LEAST_ONCE; + param.message.topic.topic.utf8 = topic; + param.message.topic.topic.size = strlen(topic); + param.message.payload.data = data; + param.message.payload.len = len; + param.message_id = mqtt_cycle++; + param.dup_flag = 0; + param.retain_flag = 0; + + data_print("Publishing: ", data, len); + LOG_INF("to topic: %s len: %u", + topic, + (unsigned int)strlen(topic)); + + return mqtt_publish(c, ¶m); +} + +/**@brief Function to publish data on the configured topic + */ +/* Define the function data_publish() to publish data */ +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 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: + /* Subscribe to the topic CONFIG_MQTT_SUB_TOPIC when we have a successful connection */ + if (evt->result != 0) { + LOG_ERR("MQTT connect failed: %d", evt->result); + break; + } + + LOG_INF("MQTT client connected"); + subscribe(c); + break; + + case MQTT_EVT_DISCONNECT: + LOG_INF("MQTT client disconnected: %d", evt->result); + break; + + case MQTT_EVT_PUBLISH: + /* Listen to published messages received from the broker and extract the message */ + { + /* Extract the payload */ + const struct mqtt_publish_param *p = &evt->param.publish; + //Print the length of the recived message + LOG_INF("MQTT PUBLISH result=%d len=%d", + evt->result, p->message.payload.len); + + //Extract the data of the recived message + err = get_received_payload(c, p->message.payload.len); + + //Send acknowledgment to the broker on receiving QoS1 publish message + 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(c, &ack); + } + + /* On successful extraction of data */ + if (err >= 0) { + data_print("Received: ", payload_buf, p->message.payload.len); + // Control the LED + if(strncmp(payload_buf,CONFIG_TURN_LED_ON_CMD,sizeof(CONFIG_TURN_LED_ON_CMD)-1) == 0){ + dk_set_led_on(LED_CONTROL_OVER_MQTT); + } + else if(strncmp(payload_buf,CONFIG_TURN_LED_OFF_CMD,sizeof(CONFIG_TURN_LED_OFF_CMD)-1) == 0){ + dk_set_led_off(LED_CONTROL_OVER_MQTT); + } + /* On failed extraction of data */ + // Payload buffer is smaller than the received data + } 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)); + // Failed to extract data, disconnect + } else { + LOG_ERR("get_received_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", (char *)(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; +} + +/* 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 imei_buf[CGSN_RESPONSE_LENGTH + 1]; + int err; + + err = nrf_modem_at_cmd(imei_buf, sizeof(imei_buf), "AT+CGSN"); + if (err) { + LOG_ERR("Failed to obtain IMEI, error: %d", err); + goto exit; + } + + imei_buf[IMEI_LEN] = '\0'; + + snprintf(client_id, sizeof(client_id), "nrf-%.*s", IMEI_LEN, imei_buf); + +exit: + LOG_DBG("client_id = %s", (char *)(client_id)); + + return client_id; +} + + +/**@brief Initialize the MQTT client structure + */ +/* Define the function client_init() to initialize the MQTT client instance. */ +int client_init(struct mqtt_client *client) +{ + int err; + /* Initializes the client instance. */ + mqtt_client_init(client); + + /* Resolves the configured hostname and initializes the MQTT broker structure */ + 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); + + /* We are not using TLS */ + client->transport.type = MQTT_TRANSPORT_NON_SECURE; + + return err; +} + +/**@brief Initialize the file descriptor structure used by poll. + */ +int fds_init(struct mqtt_client *c, struct pollfd *fds) +{ + if (c->transport.type == MQTT_TRANSPORT_NON_SECURE) { + fds->fd = c->transport.tcp.sock; + } else { + return -ENOTSUP; + } + + fds->events = POLLIN; + + return 0; +} \ No newline at end of file -- cgit v1.2.3