MQTT Client

MQTT client v3.1.1 implementation, based on callback (non-netconn) connection API.

MQTT application example code

/*
 * MQTT client example with ESP device using asynchronous callbacks
 *
 * Once device is connected to network,
 * it will try to connect to mosquitto test server and start the MQTT.
 *
 * If successfully connected, it will publish data to "lwesp_topic" topic every x seconds.
 *
 * To check if data are sent, you can use mqtt-spy PC software to inspect
 * test.mosquitto.org server and subscribe to publishing topic
 */
#include "lwesp/apps/lwesp_mqtt_client.h"
#include "lwesp/lwesp.h"
#include "lwesp/lwesp_timeout.h"
#include "mqtt_client.h"

static lwesp_mqtt_client_p  mqtt_client;    /*!< MQTT client structure */
static char mqtt_client_id[13];             /*!< Client ID is structured from ESP station MAC address */

/**
 * \brief           Connection information for MQTT CONNECT packet
 */
static const lwesp_mqtt_client_info_t
mqtt_client_info = {
    .id = mqtt_client_id,                       /* The only required field for connection! */

    .keep_alive = 10,
    // .user = "test_username",
    // .pass = "test_password",
};

static void prv_mqtt_cb(lwesp_mqtt_client_p client, lwesp_mqtt_evt_t* evt);
static void prv_example_do_connect(lwesp_mqtt_client_p client);
static uint32_t retries = 0;

/**
 * \brief           Custom callback function for ESP events
 * \param[in]       evt: ESP event callback function
 */
static lwespr_t
prv_mqtt_lwesp_cb(lwesp_evt_t* evt) {
    switch (lwesp_evt_get_type(evt)) {
#if LWESP_CFG_MODE_STATION
        case LWESP_EVT_WIFI_GOT_IP: {
            prv_example_do_connect(mqtt_client);/* Start connection after we have a connection to network client */
            break;
        }
#endif /* LWESP_CFG_MODE_STATION */
        default:
            break;
    }
    return lwespOK;
}

/**
 * \brief           MQTT client thread
 * \param[in]       arg: User argument
 */
void
mqtt_client_thread(void const* arg) {
    lwesp_mac_t mac;

    LWESP_UNUSED(arg);

    /* Register new callback for general events from ESP stack */
    lwesp_evt_register(prv_mqtt_lwesp_cb);

    /* Get station MAC to format client ID */
    if (lwesp_sta_getmac(&mac, NULL, NULL, 1) == lwespOK) {
        snprintf(mqtt_client_id, sizeof(mqtt_client_id), "%02X%02X%02X%02X%02X%02X",
                 (unsigned)mac.mac[0], (unsigned)mac.mac[1], (unsigned)mac.mac[2],
                 (unsigned)mac.mac[3], (unsigned)mac.mac[4], (unsigned)mac.mac[5]
                );
    } else {
        strcpy(mqtt_client_id, "unknown");
    }
    printf("MQTT Client ID: %s\r\n", mqtt_client_id);

    /*
     * Create a new client with 256 bytes of RAW TX data
     * and 128 bytes of RAW incoming data
     * 
     * If station is already connected to access point,
     * try to connect immediately, otherwise it
     * will get connected from callback function instead 
     */
    mqtt_client = lwesp_mqtt_client_new(256, 128);  /* Create new MQTT client */
    if (lwesp_sta_is_joined()) {                /* If ESP is already joined to network */
        prv_example_do_connect(mqtt_client);    /* Start connection to MQTT server */
    }                                           

    /* Make dummy delay of thread */
    while (1) {
        lwesp_delay(1000);
    }
}

/**
 * \brief           Timeout callback for MQTT events
 * \param[in]       arg: User argument
 */
static void
prv_mqtt_timeout_cb(void* arg) {
    static char tx_data[20];
    static uint32_t num = 10;
    lwesp_mqtt_client_p client = arg;
    lwespr_t res;
    
    if (lwesp_mqtt_client_is_connected(client)) {
        sprintf(tx_data, "R: %u, N: %u", (unsigned)retries, (unsigned)num);
        if ((res = lwesp_mqtt_client_publish(client, "lwesp_topic", tx_data, LWESP_U16(strlen(tx_data)), LWESP_MQTT_QOS_EXACTLY_ONCE, 0, (void*)((uintptr_t)num))) == lwespOK) {
            printf("Publishing %d...\r\n", (int)num);
            num++;
        } else {
            printf("Cannot publish...: %d\r\n", (int)res);
        }
    }
    lwesp_timeout_add(10000, prv_mqtt_timeout_cb, arg);
}

/**
 * \brief           MQTT event callback function
 * \param[in]       client: MQTT client where event occurred
 * \param[in]       evt: Event type and data
 */
static void
prv_mqtt_cb(lwesp_mqtt_client_p client, lwesp_mqtt_evt_t* evt) {
    switch (lwesp_mqtt_client_evt_get_type(client, evt)) {
        /*
         * Connect event
         * Called if user successfully connected to MQTT server
         * or even if connection failed for some reason
         */
        case LWESP_MQTT_EVT_CONNECT: {            /* MQTT connect event occurred */
            lwesp_mqtt_conn_status_t status = lwesp_mqtt_client_evt_connect_get_status(client, evt);

            if (status == LWESP_MQTT_CONN_STATUS_ACCEPTED) {
                printf("MQTT accepted!\r\n");
                /*
                 * Once we are accepted by server,
                 * it is time to subscribe to different topics
                 * We will subscrive to "mqtt_lwesp_example_topic" topic,
                 * and will also set the same name as subscribe argument for callback later
                 */
                lwesp_mqtt_client_subscribe(client, "lwesp_topic", LWESP_MQTT_QOS_EXACTLY_ONCE, "lwesp_topic");

                /* Start timeout timer after 5000ms and call mqtt_timeout_cb function */
                lwesp_timeout_add(5000, prv_mqtt_timeout_cb, client);
            } else {
                printf("MQTT server connection was not successful: %d\r\n", (int)status);

                /* Try to connect all over again */
                prv_example_do_connect(client);
            }
            break;
        }

        /*
         * Subscribe event just happened.
         * Here it is time to check if it was successful or failed attempt
         */
        case LWESP_MQTT_EVT_SUBSCRIBE: {
            const char* arg = lwesp_mqtt_client_evt_subscribe_get_argument(client, evt);  /* Get user argument */
            lwespr_t res = lwesp_mqtt_client_evt_subscribe_get_result(client, evt); /* Get result of subscribe event */

            if (res == lwespOK) {
                printf("Successfully subscribed to %s topic\r\n", arg);
                if (!strcmp(arg, "lwesp_topic")) {   /* Check topic name we were subscribed */
                    /* Subscribed to "lwesp_topic" topic */

                    /*
                     * Now publish an even on example topic
                     * and set QoS to minimal value which does not guarantee message delivery to received
                     */
                    lwesp_mqtt_client_publish(client, "lwesp_topic", "test_data", 9, LWESP_MQTT_QOS_AT_MOST_ONCE, 0, (void*)1);
                }
            }
            break;
        }

        /* Message published event occurred */
        case LWESP_MQTT_EVT_PUBLISH: {
            uint32_t val = (uint32_t)(uintptr_t)lwesp_mqtt_client_evt_publish_get_argument(client, evt);/* Get user argument, which is in fact our custom number */

            printf("Publish event, user argument on message was: %d\r\n", (int)val);
            break;
        }

        /*
         * A new message was published to us
         * and now it is time to read the data
         */
        case LWESP_MQTT_EVT_PUBLISH_RECV: {
            const char* topic = lwesp_mqtt_client_evt_publish_recv_get_topic(client, evt);
            size_t topic_len = lwesp_mqtt_client_evt_publish_recv_get_topic_len(client, evt);
            const uint8_t* payload = lwesp_mqtt_client_evt_publish_recv_get_payload(client, evt);
            size_t payload_len = lwesp_mqtt_client_evt_publish_recv_get_payload_len(client, evt);

            LWESP_UNUSED(payload);
            LWESP_UNUSED(payload_len);
            LWESP_UNUSED(topic);
            LWESP_UNUSED(topic_len);
            break;
        }

        /* Client is fully disconnected from MQTT server */
        case LWESP_MQTT_EVT_DISCONNECT: {
            printf("MQTT client disconnected!\r\n");
            prv_example_do_connect(client);         /* Connect to server all over again */
            break;
        }

        default:
            break;
    }
}

/**
 * \brief           Make a connection to MQTT server in non-blocking mode
 * Act only if client ready to connect and not already connected 
 */
static void
prv_example_do_connect(lwesp_mqtt_client_p client) {
    if (client == NULL
        || lwesp_mqtt_client_is_connected(client)) {
        return;
    }
    printf("Trying to connect to MQTT server\r\n");

    /*
     * Start a simple connection to open source
     * MQTT server on mosquitto.org
     */
    retries++;
    lwesp_timeout_remove(prv_mqtt_timeout_cb);
    lwesp_mqtt_client_connect(mqtt_client, "test.mosquitto.org", 1883, prv_mqtt_cb, &mqtt_client_info);
}

group LWESP_APP_MQTT_CLIENT

MQTT client.

Typedefs

typedef struct lwesp_mqtt_client *lwesp_mqtt_client_p

Pointer to lwesp_mqtt_client_t structure.

typedef void (*lwesp_mqtt_evt_fn)(lwesp_mqtt_client_p client, lwesp_mqtt_evt_t *evt)

MQTT event callback function.

Param client

[in] MQTT client

Param evt

[in] MQTT event with type and related data

Enums

enum lwesp_mqtt_qos_t

Quality of service enumeration.

Values:

enumerator LWESP_MQTT_QOS_AT_MOST_ONCE = 0x00

Delivery is not guaranteed to arrive, but can arrive up to 1 time = non-critical packets where losses are allowed

enumerator LWESP_MQTT_QOS_AT_LEAST_ONCE = 0x01

Delivery is quaranteed at least once, but it may be delivered multiple times with the same content

enumerator LWESP_MQTT_QOS_EXACTLY_ONCE = 0x02

Delivery is quaranteed exactly once = very critical packets such as billing informations or similar

enum lwesp_mqtt_state_t

State of MQTT client.

Values:

enumerator LWESP_MQTT_CONN_DISCONNECTED = 0x00

Connection with server is not established

enumerator LWESP_MQTT_CONN_CONNECTING

Client is connecting to server

enumerator LWESP_MQTT_CONN_DISCONNECTING

Client connection is disconnecting from server

enumerator LWESP_MQTT_CONNECTING

MQTT client is connecting… CONNECT command has been sent to server

enumerator LWESP_MQTT_CONNECTED

MQTT is fully connected and ready to send data on topics

enum lwesp_mqtt_evt_type_t

MQTT event types.

Values:

enumerator LWESP_MQTT_EVT_CONNECT

MQTT client connect event

enumerator LWESP_MQTT_EVT_SUBSCRIBE

MQTT client subscribed to specific topic

enumerator LWESP_MQTT_EVT_UNSUBSCRIBE

MQTT client unsubscribed from specific topic

enumerator LWESP_MQTT_EVT_PUBLISH

MQTT client publish message to server event.

Note

When publishing packet with quality of service LWESP_MQTT_QOS_AT_MOST_ONCE, you may not receive event, even if packet was successfully sent, thus do not rely on this event for packet with qos = LWESP_MQTT_QOS_AT_MOST_ONCE

enumerator LWESP_MQTT_EVT_PUBLISH_RECV

MQTT client received a publish message from server

enumerator LWESP_MQTT_EVT_DISCONNECT

MQTT client disconnected from MQTT server

enumerator LWESP_MQTT_EVT_KEEP_ALIVE

MQTT keep-alive event. It gets invoked after client and server exchange successful “keep-alive message”, defined by MQTT protocol

enumerator LWESP_MQTT_EVT_CONN_POLL

Local ESP connection poll event. When connection is active, stack periodically sends polling events to user. This event is propagated to user MQTT space

enum lwesp_mqtt_conn_status_t

List of possible results from MQTT server when executing connect command.

Values:

enumerator LWESP_MQTT_CONN_STATUS_ACCEPTED = 0x00

Connection accepted and ready to use

enumerator LWESP_MQTT_CONN_STATUS_REFUSED_PROTOCOL_VERSION = 0x01

Connection Refused, unacceptable protocol version

enumerator LWESP_MQTT_CONN_STATUS_REFUSED_ID = 0x02

Connection refused, identifier rejected

enumerator LWESP_MQTT_CONN_STATUS_REFUSED_SERVER = 0x03

Connection refused, server unavailable

enumerator LWESP_MQTT_CONN_STATUS_REFUSED_USER_PASS = 0x04

Connection refused, bad user name or password

enumerator LWESP_MQTT_CONN_STATUS_REFUSED_NOT_AUTHORIZED = 0x05

Connection refused, not authorized

enumerator LWESP_MQTT_CONN_STATUS_TCP_FAILED = 0x100

TCP connection to server was not successful

Functions

lwesp_mqtt_client_p lwesp_mqtt_client_new(size_t tx_buff_len, size_t rx_buff_len)

Allocate a new MQTT client structure.

Parameters

• tx_buff_len – [in] Length of raw data output buffer

• rx_buff_len – [in] Length of raw data input buffer

Returns

Pointer to new allocated MQTT client structure or NULL on failure

void lwesp_mqtt_client_delete(lwesp_mqtt_client_p client)

Delete MQTT client structure.

Note

MQTT client must be disconnected first

Parameters

client – [in] MQTT client

lwespr_t lwesp_mqtt_client_connect(lwesp_mqtt_client_p client, const char *host, lwesp_port_t port, lwesp_mqtt_evt_fn evt_fn, const lwesp_mqtt_client_info_t *info)

Connect to MQTT server in non-blocking mode. Function returns immediately and does not wait for server to be connected.

Note

After TCP connection is established, CONNECT packet is automatically sent to server. Application must rely on events coming to event function, passed at connect stage

Parameters

• client – [in] MQTT client

• host – [in] Host address for server

• port – [in] Host port number

• evt_fn – [in] Callback function for all events on this MQTT client

• info – [in] Information structure for connection. It is used after connection is successfully established. Variable must not be a local or changes will be lost with potential faulty operation

Returns

lwespOK on success, member of lwespr_t enumeration otherwise

lwespr_t lwesp_mqtt_client_disconnect(lwesp_mqtt_client_p client)

Disconnect from MQTT server.

Parameters

client – [in] MQTT client

Returns

lwespOK if request sent to queue or member of lwespr_t otherwise

uint8_t lwesp_mqtt_client_is_connected(lwesp_mqtt_client_p client)

Test if client is connected to server and accepted to MQTT protocol.

Note

Function will return error if TCP is connected but MQTT not accepted

Parameters

client – [in] MQTT client

Returns

1 on success, 0 otherwise

lwespr_t lwesp_mqtt_client_subscribe(lwesp_mqtt_client_p client, const char *topic, lwesp_mqtt_qos_t qos, void *arg)

Subscribe to MQTT topic.

Parameters

• client – [in] MQTT client

• topic – [in] Topic name to subscribe to

• qos – [in] Quality of service. This parameter can be a value of lwesp_mqtt_qos_t

• arg – [in] User custom argument used in callback

Returns

lwespOK on success, member of lwespr_t enumeration otherwise

lwespr_t lwesp_mqtt_client_unsubscribe(lwesp_mqtt_client_p client, const char *topic, void *arg)

Unsubscribe from MQTT topic.

Parameters

• client – [in] MQTT client

• topic – [in] Topic name to unsubscribe from

• arg – [in] User custom argument used in callback

Returns

lwespOK on success, member of lwespr_t enumeration otherwise

lwespr_t lwesp_mqtt_client_publish(lwesp_mqtt_client_p client, const char *topic, const void *payload, uint16_t len, lwesp_mqtt_qos_t qos, uint8_t retain, void *arg)

Publish a new message on specific topic.

Parameters

• client – [in] MQTT client

• topic – [in] Topic to send message to

• payload – [in] Message data

• payload_len – [in] Length of payload data

• qos – [in] Quality of service. This parameter can be a value of lwesp_mqtt_qos_t enumeration

• retain – [in] Retian parameter value

• arg – [in] User custom argument used in callback

Returns

lwespOK on success, member of lwespr_t enumeration otherwise

void *lwesp_mqtt_client_get_arg(lwesp_mqtt_client_p client)

Get user argument on client.

Parameters

client – [in] MQTT client handle

Returns

User argument

void lwesp_mqtt_client_set_arg(lwesp_mqtt_client_p client, void *arg)

Set user argument on client.

Parameters

• client – [in] MQTT client handle

• arg – [in] User argument

struct lwesp_mqtt_client_info_t

#include <lwesp_mqtt_client.h>

MQTT client information structure.

Public Members

const char *id

Client unique identifier. It is required and must be set by user

const char *user

Authentication username. Set to NULL if not required

const char *pass

Authentication password, set to NULL if not required

uint16_t keep_alive

Keep-alive parameter in units of seconds. When set to 0, functionality is disabled (not recommended)

const char *will_topic

Will topic

const char *will_message

Will message

lwesp_mqtt_qos_t will_qos

Will topic quality of service

uint8_t use_ssl

Connect to server using SSL connection with AT commands

struct lwesp_mqtt_request_t

#include <lwesp_mqtt_client.h>

MQTT request object.

Public Members

uint8_t status

Entry status flag for in use or pending bit

uint16_t packet_id

Packet ID generated by client on publish

void *arg

User defined argument

uint32_t expected_sent_len

Number of total bytes which must be sent on connection before we can say “packet was sent”.

uint32_t timeout_start_time

Timeout start time in units of milliseconds

struct lwesp_mqtt_evt_t

#include <lwesp_mqtt_client.h>

MQTT event structure for callback function.

Public Members

lwesp_mqtt_evt_type_t type

Event type

lwesp_mqtt_conn_status_t status

Connection status with MQTT

struct lwesp_mqtt_evt_t::[anonymous]::[anonymous] connect

Event for connecting to server

uint8_t is_accepted

Status if client was accepted to MQTT prior disconnect event

struct lwesp_mqtt_evt_t::[anonymous]::[anonymous] disconnect

Event for disconnecting from server

void *arg

User argument for callback function

lwespr_t res

Response status

struct lwesp_mqtt_evt_t::[anonymous]::[anonymous] sub_unsub_scribed

Event for (un)subscribe to/from topics

struct lwesp_mqtt_evt_t::[anonymous]::[anonymous] publish

Published event

const uint8_t *topic

Pointer to topic identifier

size_t topic_len

Length of topic

const void *payload

Topic payload

size_t payload_len

Length of topic payload

uint8_t dup

Duplicate flag if message was sent again

lwesp_mqtt_qos_t qos

Received packet quality of service

struct lwesp_mqtt_evt_t::[anonymous]::[anonymous] publish_recv

Publish received event

union lwesp_mqtt_evt_t::[anonymous] evt

Event data parameters

group LWESP_APP_MQTT_CLIENT_EVT

Event helper functions.

Connect event

Note

Use these functions on LWESP_MQTT_EVT_CONNECT event

lwesp_mqtt_client_evt_connect_get_status(client, evt)

Get connection status.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

Connection status. Member of lwesp_mqtt_conn_status_t

Disconnect event

Note

Use these functions on LWESP_MQTT_EVT_DISCONNECT event

lwesp_mqtt_client_evt_disconnect_is_accepted(client, evt)

Check if MQTT client was accepted by server when disconnect event occurred.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

1 on success, 0 otherwise

Subscribe/unsubscribe event

Note

Use these functions on LWESP_MQTT_EVT_SUBSCRIBE or LWESP_MQTT_EVT_UNSUBSCRIBE events

lwesp_mqtt_client_evt_subscribe_get_argument(client, evt)

Get user argument used on lwesp_mqtt_client_subscribe.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

User argument

lwesp_mqtt_client_evt_subscribe_get_result(client, evt)

Get result of subscribe event.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

lwespOK on success, member of lwespr_t otherwise

lwesp_mqtt_client_evt_unsubscribe_get_argument(client, evt)

Get user argument used on lwesp_mqtt_client_unsubscribe.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

User argument

lwesp_mqtt_client_evt_unsubscribe_get_result(client, evt)

Get result of unsubscribe event.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

lwespOK on success, member of lwespr_t otherwise

Publish receive event

Note

Use these functions on LWESP_MQTT_EVT_PUBLISH_RECV event

lwesp_mqtt_client_evt_publish_recv_get_topic(client, evt)

Get topic from received publish packet.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

Topic name

lwesp_mqtt_client_evt_publish_recv_get_topic_len(client, evt)

Get topic length from received publish packet.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

Topic length

lwesp_mqtt_client_evt_publish_recv_get_payload(client, evt)

Get payload from received publish packet.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

Packet payload

lwesp_mqtt_client_evt_publish_recv_get_payload_len(client, evt)

Get payload length from received publish packet.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

Payload length

lwesp_mqtt_client_evt_publish_recv_is_duplicate(client, evt)

Check if packet is duplicated.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

1 if duplicated, 0 otherwise

lwesp_mqtt_client_evt_publish_recv_get_qos(client, evt)

Get received quality of service.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

Member of lwesp_mqtt_qos_t enumeration

Publish event

Note

Use these functions on LWESP_MQTT_EVT_PUBLISH event

lwesp_mqtt_client_evt_publish_get_argument(client, evt)

Get user argument used on lwesp_mqtt_client_publish.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

User argument

lwesp_mqtt_client_evt_publish_get_result(client, evt)

Get result of publish event.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

lwespOK on success, member of lwespr_t otherwise

Defines

lwesp_mqtt_client_evt_get_type(client, evt)

Get MQTT event type.

Parameters

• client – [in] MQTT client

• evt – [in] Event handle

Returns

MQTT Event type, value of lwesp_mqtt_evt_type_t enumeration