MQTT Client in Zephyr: Publish and Subscribe

CONFIG_MQTT_LIB=y

#include <zephyr/kernel.h>
#include <zephyr/net/socket.h>
#include <zephyr/net/mqtt.h>
#include <errno.h>
#include <stdbool.h>
#include <string.h>

/* Buffers for MQTT */
static uint8_t rx_buffer[128];
static uint8_t tx_buffer[128];
static struct mqtt_client client_ctx;
static struct sockaddr_in broker;
static bool connected;

static void mqtt_evt_handler(struct mqtt_client *const client,
                             const struct mqtt_evt *evt)
{
    switch (evt->type) {
    case MQTT_EVT_CONNACK:
        if (evt->result == 0) {
            connected = true;
            printk("MQTT session established\n");
        } else {
            printk("Broker rejected connection: %d\n", evt->result);
        }
        break;

    case MQTT_EVT_PUBLISH: {
        size_t remaining = evt->param.publish.message.payload.len;
        uint8_t payload[64];

        printk("Message received: ");
        while (remaining > 0) {
            int len = mqtt_read_publish_payload(client, payload,
                                                 MIN(sizeof(payload), remaining));
            if (len <= 0) {
                break;
            }
            printk("%.*s", len, payload);
            remaining -= len;
        }
        printk("\n");
        break;
    }

    case MQTT_EVT_DISCONNECT:
        connected = false;
        printk("MQTT session closed: %d\n", evt->result);
        break;

    default:
        break;
    }
}

/* Helper function to prepare the client structure */
static int prepare_mqtt_client(void)
{
    mqtt_client_init(&client_ctx);

    /* Lab broker address */
    broker.sin_family = AF_INET;
    broker.sin_port = htons(1883);
    if (zsock_inet_pton(AF_INET, "192.0.2.10", &broker.sin_addr) != 1) {
        return -EINVAL;
    }

    client_ctx.broker = (struct sockaddr *)&broker;
    client_ctx.evt_cb = mqtt_evt_handler;
    client_ctx.client_id.utf8 = (uint8_t *)"zephyr_device_001";
    client_ctx.client_id.size = strlen("zephyr_device_001");
    client_ctx.protocol_version = MQTT_VERSION_3_1_1;
    client_ctx.transport.type = MQTT_TRANSPORT_NON_SECURE;
    
    /* Manual buffers (Mandatory in Zephyr) */
    client_ctx.rx_buf = rx_buffer;
    client_ctx.rx_buf_size = sizeof(rx_buffer);
    client_ctx.tx_buf = tx_buffer;
    client_ctx.tx_buf_size = sizeof(tx_buffer);

    return 0;
}

static int process_mqtt(int timeout_ms)
{
    struct zsock_pollfd fds = {
        .fd = client_ctx.transport.tcp.sock,
        .events = ZSOCK_POLLIN,
    };

    int rc = zsock_poll(&fds, 1, timeout_ms);
    if (rc < 0) {
        return -errno;
    }
    if (rc > 0 && (fds.revents & (ZSOCK_POLLERR | ZSOCK_POLLHUP))) {
        return -ECONNRESET;
    }
    if (rc > 0 && (fds.revents & ZSOCK_POLLIN)) {
        rc = mqtt_input(&client_ctx);
        if (rc != 0) {
            return rc;
        }
    }
    return mqtt_live(&client_ctx);
}

int main(void)
{
    /* ... (Assume Wi-Fi already connected and we have an IP) ... */

    int rc = prepare_mqtt_client();
    if (rc != 0) {
        return rc;
    }

    /* 1. Connection */
    rc = mqtt_connect(&client_ctx);
    if (rc != 0) {
        printk("Error connecting MQTT: %d\n", rc);
        return rc;
    }

    /* mqtt_connect sends CONNECT; we process input until receiving CONNACK. */
    while (!connected) {
        rc = process_mqtt(1000);
        if (rc != 0) {
            return rc;
        }
    }

    /* 2. Subscribe to commands */
    struct mqtt_topic topics[] = {{
        .topic = {
            .utf8 = (uint8_t *)"zephyr/curso/comandos",
            .size = strlen("zephyr/curso/comandos"),
        },
        .qos = MQTT_QOS_0_AT_MOST_ONCE,
    }};
    const struct mqtt_subscription_list subscriptions = {
        .list = topics,
        .list_count = ARRAY_SIZE(topics),
        .message_id = 1,
    };

    rc = mqtt_subscribe(&client_ctx, &subscriptions);
    if (rc != 0) {
        return rc;
    }

    /* 3. Publish telemetry */
    char payload[] = "Hello from Zephyr OS";
    struct mqtt_publish_param param = {
        .message = {
            .topic = {
                .topic = {
                    .utf8 = (uint8_t *)"zephyr/curso/telemetria",
                    .size = strlen("zephyr/curso/telemetria"),
                },
                .qos = MQTT_QOS_0_AT_MOST_ONCE,
            },
            .payload = {
                .data = payload,
                .len = strlen(payload),
            },
        },
        .message_id = 2,
    };

    rc = mqtt_publish(&client_ctx, &param);
    if (rc == 0) {
        printk("Message published successfully.\n");
    }

    /* 4. Process input and keep the connection alive */
    while (connected) {
        rc = process_mqtt(1000);
        if (rc != 0) {
            printk("Error in MQTT session: %d\n", rc);
            break;
        }
    }

    mqtt_disconnect(&client_ctx);
    return rc;
}