If you need urgent consulting help click here

MQTT

Overview
Sample usage
Using MQTT with TLS
API Reference

Overview 

MQTT (Message Queuing Telemetry Transport) is an application layer protocol which works on top of the TCP/IP stack. It is a lightweight publish/subscribe messaging transport for machine-to-machine communication. For more information about the protocol itself, see http://mqtt.org/.

Zephyr provides an MQTT client library built on top of BSD sockets API. The library is configurable at a per-client basis, with support for MQTT versions 3.1.0 and 3.1.1. The Zephyr MQTT implementation can be used with either plain sockets communicating over TCP, or with secure sockets communicating over TLS. See BSD Sockets for more information about Zephyr sockets.

MQTT clients require an MQTT server to connect to. Such a server, called an MQTT Broker, is responsible for managing client subscriptions and distributing messages published by clients. There are many implementations of MQTT brokers, one of them being Eclipse Mosquitto. See https://mosquitto.org/ for more information about the Eclipse Mosquitto project.

Sample usage 

To create an MQTT client, a client context structure and buffers need to be defined:

/* Buffers for MQTT client. */
static uint8_t rx_buffer[256];
static uint8_t tx_buffer[256];

/* MQTT client context */
static struct mqtt_client client_ctx;

Multiple MQTT client instances can be created in the application and managed independently. Additionally, a structure for MQTT Broker address information is needed. This structure must be accessible throughout the lifespan of the MQTT client and can be shared among MQTT clients:

/* MQTT Broker address information. */
static struct sockaddr_storage broker;

An MQTT client library will notify MQTT events to the application through a callback function created to handle respective events:

void mqtt_evt_handler(struct mqtt_client *client,
                      const struct mqtt_evt *evt)
{
   switch (evt->type) {
      /* Handle events here. */
   }
}

For a list of possible events, see API Reference.

The client context structure needs to be initialized and set up before it can be used. An example configuration for TCP transport is shown below:

mqtt_client_init(&client_ctx);

/* MQTT client configuration */
client_ctx.broker = &broker;
client_ctx.evt_cb = mqtt_evt_handler;
client_ctx.client_id.utf8 = (uint8_t *)"zephyr_mqtt_client";
client_ctx.client_id.size = sizeof("zephyr_mqtt_client") - 1;
client_ctx.password = NULL;
client_ctx.user_name = NULL;
client_ctx.protocol_version = MQTT_VERSION_3_1_1;
client_ctx.transport.type = MQTT_TRANSPORT_NON_SECURE;

/* MQTT buffers configuration */
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);

After the configuration is set up, the MQTT client can connect to the MQTT broker. Call the mqtt_connect function, which will create the appropriate socket, establish a TCP/TLS connection, and send an MQTT CONNECT message. When notified, the application should call the mqtt_input function to process the response received. Note, that mqtt_input is a non-blocking function, therefore the application should use socket poll to wait for the response. If the connection was successful, MQTT_EVT_CONNACK will be notified to the application through the callback function.

rc = mqtt_connect(&client_ctx);
if (rc != 0) {
   return rc;
}

fds[0].fd = client_ctx.transport.tcp.sock;
fds[0].events = ZSOCK_POLLIN;
poll(fds, 1, K_MSEC(5000));

mqtt_input(&client_ctx);

if (!connected) {
   mqtt_abort(&client_ctx);
}

In the above code snippet, the MQTT callback function should set the connected flag upon a successful connection. If the connection fails at the MQTT level or a timeout occurs, the connection will be aborted, and the underlying socket closed.

After the connection is established, an application needs to call mqtt_input and mqtt_live functions periodically to process incoming data and upkeep the connection. If an MQTT message is received, an MQTT callback function will be called and an appropriate event notified.

The connection can be closed by calling the mqtt_disconnect function.

Zephyr provides sample code utilizing the MQTT client API. See MQTT Publisher for more information.

Using MQTT with TLS 

The Zephyr MQTT library can be used with TLS transport for secure communication by selecting a secure transport type (MQTT_TRANSPORT_SECURE) and some additional configuration information:

client_ctx.transport.type = MQTT_TRANSPORT_SECURE;

struct mqtt_sec_config *tls_config = &client_ctx.transport.tls.config;

tls_config->peer_verify = TLS_PEER_VERIFY_REQUIRED;
tls_config->cipher_list = NULL;
tls_config->sec_tag_list = m_sec_tags;
tls_config->sec_tag_count = ARRAY_SIZE(m_sec_tags);
tls_config->hostname = MQTT_BROKER_HOSTNAME;

In this sample code, the m_sec_tags array holds a list of tags, referencing TLS credentials that the MQTT library should use for authentication. We do not specify cipher_list, to allow the use of all cipher suites available in the system. We set hostname field to broker hostname, which is required for server authentication. Finally, we enforce peer certificate verification by setting the peer_verify field.

Note, that TLS credentials referenced by the m_sec_tags array must be registered in the system first. For more information on how to do that, refer to secure sockets documentation.

An example of how to use TLS with MQTT is also present in MQTT Publisher.

API Reference 

group mqtt_socket

MQTT Client Implementation.

MQTT Client’s Application interface is defined in this header.

Note

The implementation assumes TCP module is enabled.

Note

By default the implementation uses MQTT version 3.1.1.

Defines

MQTT_UTF8_LITERAL(literal)

Initialize UTF-8 encoded string from C literal string.

Use it as follows:

struct mqtt_utf8 password = MQTT_UTF8_LITERAL(“my_pass”);

Parameters

literal – [in] Literal string from which to generate mqtt_utf8 object.

Typedefs

typedef void (*mqtt_evt_cb_t)(struct mqtt_client *client, const struct mqtt_evt *evt)

Asynchronous event notification callback registered by the application.

Param client: [in] Identifies the client for which the event is notified.
Param evt: [in] Event description along with result and associated parameters (if any).

Enums

enum mqtt_evt_type

MQTT Asynchronous Events notified to the application from the module through the callback registered by the application.

Values:

enumerator MQTT_EVT_CONNACK: Acknowledgment of connection request. Event result accompanying the event indicates whether the connection failed or succeeded.

enumerator MQTT_EVT_DISCONNECT: Disconnection Event. MQTT Client Reference is no longer valid once this event is received for the client.

enumerator MQTT_EVT_PUBLISH: Publish event received when message is published on a topic client is subscribed to.

Note

PUBLISH event structure only contains payload size, the payload data parameter should be ignored. Payload content has to be read manually with mqtt_read_publish_payload function.

enumerator MQTT_EVT_PUBACK: Acknowledgment for published message with QoS 1.

enumerator MQTT_EVT_PUBREC: Reception confirmation for published message with QoS 2.

enumerator MQTT_EVT_PUBREL: Release of published message with QoS 2.

enumerator MQTT_EVT_PUBCOMP: Confirmation to a publish release message with QoS 2.

enumerator MQTT_EVT_SUBACK: Acknowledgment to a subscribe request.

enumerator MQTT_EVT_UNSUBACK: Acknowledgment to a unsubscribe request.

enumerator MQTT_EVT_PINGRESP: Ping Response from server.

enum mqtt_version

MQTT version protocol level.

Values:

enumerator MQTT_VERSION_3_1_0 = 3: Protocol level for 3.1.0.

enumerator MQTT_VERSION_3_1_1 = 4: Protocol level for 3.1.1.

enum mqtt_qos

MQTT Quality of Service types.

Values:

enumerator MQTT_QOS_0_AT_MOST_ONCE = 0x00: Lowest Quality of Service, no acknowledgment needed for published message.

enumerator MQTT_QOS_1_AT_LEAST_ONCE = 0x01: Medium Quality of Service, if acknowledgment expected for published message, duplicate messages permitted.

enumerator MQTT_QOS_2_EXACTLY_ONCE = 0x02: Highest Quality of Service, acknowledgment expected and message shall be published only once. Message not published to interested parties unless client issues a PUBREL.

enum mqtt_conn_return_code

MQTT CONNACK return codes.

Values:

enumerator MQTT_CONNECTION_ACCEPTED = 0x00: Connection accepted.

enumerator MQTT_UNACCEPTABLE_PROTOCOL_VERSION = 0x01: The Server does not support the level of the MQTT protocol requested by the Client.

enumerator MQTT_IDENTIFIER_REJECTED = 0x02: The Client identifier is correct UTF-8 but not allowed by the Server.

enumerator MQTT_SERVER_UNAVAILABLE = 0x03: The Network Connection has been made but the MQTT service is unavailable.

enumerator MQTT_BAD_USER_NAME_OR_PASSWORD = 0x04: The data in the user name or password is malformed.

enumerator MQTT_NOT_AUTHORIZED = 0x05: The Client is not authorized to connect.

enum mqtt_suback_return_code

MQTT SUBACK return codes.

Values:

enumerator MQTT_SUBACK_SUCCESS_QoS_0 = 0x00: Subscription with QoS 0 succeeded.

enumerator MQTT_SUBACK_SUCCESS_QoS_1 = 0x01: Subscription with QoS 1 succeeded.

enumerator MQTT_SUBACK_SUCCESS_QoS_2 = 0x02: Subscription with QoS 2 succeeded.

enumerator MQTT_SUBACK_FAILURE = 0x80: Subscription for a topic failed.

enum mqtt_transport_type

MQTT transport type.

Values:

enumerator MQTT_TRANSPORT_NON_SECURE: Use non secure TCP transport for MQTT connection.

enumerator MQTT_TRANSPORT_NUM: Shall not be used as a transport type. Indicator of maximum transport types possible.

Functions

void mqtt_client_init(struct mqtt_client *client)

Initializes the client instance.

Note

Shall be called to initialize client structure, before setting any client parameters and before connecting to broker.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.

int mqtt_connect(struct mqtt_client *client)

API to request new MQTT client connection.

Note

This memory is assumed to be resident until mqtt_disconnect is called.

Note

Any subsequent changes to parameters like broker address, user name, device id, etc. have no effect once MQTT connection is established.

Note

Default protocol revision used for connection request is 3.1.1. Please set client.protocol_version = MQTT_VERSION_3_1_0 to use protocol 3.1.0.

Note

Please modify CONFIG_MQTT_KEEPALIVE time to override default of 1 minute.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_publish(struct mqtt_client *client, const struct mqtt_publish_param *param)

API to publish messages on topics.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.
param – [in] Parameters to be used for the publish message. Shall not be NULL.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_publish_qos1_ack(struct mqtt_client *client, const struct mqtt_puback_param *param)

API used by client to send acknowledgment on receiving QoS1 publish message. Should be called on reception of MQTT_EVT_PUBLISH with QoS level MQTT_QOS_1_AT_LEAST_ONCE.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.
param – [in] Identifies message being acknowledged.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_publish_qos2_receive(struct mqtt_client *client, const struct mqtt_pubrec_param *param)

API used by client to send acknowledgment on receiving QoS2 publish message. Should be called on reception of MQTT_EVT_PUBLISH with QoS level MQTT_QOS_2_EXACTLY_ONCE.

Parameters

client – [in] Identifies client instance for which the procedure is requested. Shall not be NULL.
param – [in] Identifies message being acknowledged.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_publish_qos2_release(struct mqtt_client *client, const struct mqtt_pubrel_param *param)

API used by client to request release of QoS2 publish message. Should be called on reception of MQTT_EVT_PUBREC.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.
param – [in] Identifies message being released.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_publish_qos2_complete(struct mqtt_client *client, const struct mqtt_pubcomp_param *param)

API used by client to send acknowledgment on receiving QoS2 publish release message. Should be called on reception of MQTT_EVT_PUBREL.

Parameters

client – [in] Identifies client instance for which the procedure is requested. Shall not be NULL.
param – [in] Identifies message being completed.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_subscribe(struct mqtt_client *client, const struct mqtt_subscription_list *param)

API to request subscription of one or more topics on the connection.

Parameters

client – [in] Identifies client instance for which the procedure is requested. Shall not be NULL.
param – [in] Subscription parameters. Shall not be NULL.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_unsubscribe(struct mqtt_client *client, const struct mqtt_subscription_list *param)

API to request unsubscription of one or more topics on the connection.

Note

QoS included in topic description is unused in this API.

Parameters

client – [in] Identifies client instance for which the procedure is requested. Shall not be NULL.
param – [in] Parameters describing topics being unsubscribed from. Shall not be NULL.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_ping(struct mqtt_client *client)

API to send MQTT ping. The use of this API is optional, as the library handles the connection keep-alive on it’s own, see mqtt_live.

Parameters

client – [in] Identifies client instance for which procedure is requested.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_disconnect(struct mqtt_client *client)

API to disconnect MQTT connection.

Parameters

client – [in] Identifies client instance for which procedure is requested.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_abort(struct mqtt_client *client)

API to abort MQTT connection. This will close the corresponding transport without closing the connection gracefully at the MQTT level (with disconnect message).

Parameters

client – [in] Identifies client instance for which procedure is requested.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_live(struct mqtt_client *client)

This API should be called periodically for the client to be able to keep the connection alive by sending Ping Requests if need be.

Note

Application shall ensure that the periodicity of calling this function makes it possible to respect the Keep Alive time agreed with the broker on connection. mqtt_connect for details on Keep Alive time.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_keepalive_time_left(const struct mqtt_client *client)

Helper function to determine when next keep alive message should be sent. Can be used for instance as a source for poll timeout.

Parameters

client – [in] Client instance for which the procedure is requested.

Returns

Time in milliseconds until next keep alive message is expected to be sent. Function will return -1 if keep alive messages are not enabled.

int mqtt_input(struct mqtt_client *client)

Receive an incoming MQTT packet. The registered callback will be called with the packet content.

Note

In case of PUBLISH message, the payload has to be read separately with mqtt_read_publish_payload function. The size of the payload to read is provided in the publish event structure.

Note

This is a non-blocking call.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.

Returns

0 or a negative error code (errno.h) indicating reason of failure.

int mqtt_read_publish_payload(struct mqtt_client *client, void *buffer, size_t length)

Read the payload of the received PUBLISH message. This function should be called within the MQTT event handler, when MQTT PUBLISH message is notified.

Note

This is a non-blocking call.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.
buffer – [out] Buffer where payload should be stored.
length – [in] Length of the buffer, in bytes.

Returns

Number of bytes read or a negative error code (errno.h) indicating reason of failure.

int mqtt_read_publish_payload_blocking(struct mqtt_client *client, void *buffer, size_t length)

Blocking version of mqtt_read_publish_payload function.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.
buffer – [out] Buffer where payload should be stored.
length – [in] Length of the buffer, in bytes.

Returns

Number of bytes read or a negative error code (errno.h) indicating reason of failure.

int mqtt_readall_publish_payload(struct mqtt_client *client, uint8_t *buffer, size_t length)

Blocking version of mqtt_read_publish_payload function which runs until the required number of bytes are read.

Parameters

client – [in] Client instance for which the procedure is requested. Shall not be NULL.
buffer – [out] Buffer where payload should be stored.
length – [in] Number of bytes to read.

Returns

0 if success, otherwise a negative error code (errno.h) indicating reason of failure.

struct mqtt_utf8

#include <mqtt.h>

Abstracts UTF-8 encoded strings.

Public Members

const uint8_t *utf8: Pointer to UTF-8 string.

uint32_t size: Size of UTF string, in bytes.

struct mqtt_binstr

#include <mqtt.h>

Abstracts binary strings.

Public Members

uint8_t *data: Pointer to binary stream.

uint32_t len: Length of binary stream.

struct mqtt_topic

#include <mqtt.h>

Abstracts MQTT UTF-8 encoded topic that can be subscribed to or published.

Public Members

struct mqtt_utf8 topic: Topic on to be published or subscribed to.

uint8_t qos: Quality of service requested for the subscription. mqtt_qos for details.

struct mqtt_publish_message

#include <mqtt.h>

Parameters for a publish message.

Public Members

struct mqtt_topic topic: Topic on which data was published.

struct mqtt_binstr payload: Payload on the topic published.

struct mqtt_connack_param

#include <mqtt.h>

Parameters for a connection acknowledgment (CONNACK).

Public Members

uint8_t session_present_flag: The Session Present flag enables a Client to establish whether the Client and Server have a consistent view about whether there is already stored Session state.

enum mqtt_conn_return_code return_code: The appropriate non-zero Connect return code indicates if the Server is unable to process a connection request for some reason.

struct mqtt_puback_param: #include <mqtt.h>

Parameters for MQTT publish acknowledgment (PUBACK).

struct mqtt_pubrec_param: #include <mqtt.h>

Parameters for MQTT publish receive (PUBREC).

struct mqtt_pubrel_param: #include <mqtt.h>

Parameters for MQTT publish release (PUBREL).

struct mqtt_pubcomp_param: #include <mqtt.h>

Parameters for MQTT publish complete (PUBCOMP).

struct mqtt_suback_param: #include <mqtt.h>

Parameters for MQTT subscription acknowledgment (SUBACK).

struct mqtt_unsuback_param: #include <mqtt.h>

Parameters for MQTT unsubscribe acknowledgment (UNSUBACK).

struct mqtt_publish_param

#include <mqtt.h>

Parameters for a publish message.

Public Members

struct mqtt_publish_message message: Messages including topic, QoS and its payload (if any) to be published.

uint16_t message_id: Message id used for the publish message. Redundant for QoS 0.

uint8_t dup_flag: Duplicate flag. If 1, it indicates the message is being retransmitted. Has no meaning with QoS 0.

uint8_t retain_flag: Retain flag. If 1, the message shall be stored persistently by the broker.

struct mqtt_subscription_list

#include <mqtt.h>

List of topics in a subscription request.

Public Members

struct mqtt_topic *list: Array containing topics along with QoS for each.

uint16_t list_count: Number of topics in the subscription list

uint16_t message_id: Message id used to identify subscription request.

union mqtt_evt_param

#include <mqtt.h>

Defines event parameters notified along with asynchronous events to the application.

Public Members

struct mqtt_connack_param connack: Parameters accompanying MQTT_EVT_CONNACK event.

struct mqtt_publish_param publish: Parameters accompanying MQTT_EVT_PUBLISH event.

Note

PUBLISH event structure only contains payload size, the payload data parameter should be ignored. Payload content has to be read manually with mqtt_read_publish_payload function.

struct mqtt_puback_param puback: Parameters accompanying MQTT_EVT_PUBACK event.

struct mqtt_pubrec_param pubrec: Parameters accompanying MQTT_EVT_PUBREC event.

struct mqtt_pubrel_param pubrel: Parameters accompanying MQTT_EVT_PUBREL event.

struct mqtt_pubcomp_param pubcomp: Parameters accompanying MQTT_EVT_PUBCOMP event.

struct mqtt_suback_param suback: Parameters accompanying MQTT_EVT_SUBACK event.

struct mqtt_unsuback_param unsuback: Parameters accompanying MQTT_EVT_UNSUBACK event.

struct mqtt_evt

#include <mqtt.h>

Defines MQTT asynchronous event notified to the application.

Public Members

enum mqtt_evt_type type: Identifies the event.

union mqtt_evt_param param: Contains parameters (if any) accompanying the event.

int result: Event result. 0 or a negative error code (errno.h) indicating reason of failure.

struct mqtt_sec_config

#include <mqtt.h>

TLS configuration for secure MQTT transports.

Public Members

int peer_verify: Indicates the preference for peer verification.

uint32_t cipher_count: Indicates the number of entries in the cipher list.

int *cipher_list: Indicates the list of ciphers to be used for the session. May be NULL to use the default ciphers.

uint32_t sec_tag_count: Indicates the number of entries in the sec tag list.

sec_tag_t *sec_tag_list: Indicates the list of security tags to be used for the session.

const char *hostname: Peer hostname for ceritificate verification. May be NULL to skip hostname verification.

int cert_nocopy: Indicates the preference for copying certificates to the heap.

struct mqtt_transport

#include <mqtt.h>

MQTT transport specific data.

Public Members

enum mqtt_transport_type type: Transport type selection for client instance. mqtt_transport_type for possible values. MQTT_TRANSPORT_MAX is not a valid type.

int sock: Socket descriptor.

struct mqtt_internal

#include <mqtt.h>

MQTT internal state.

Public Members

struct sys_mutex mutex: Internal. Mutex to protect access to the client instance.

uint32_t last_activity: Internal. Wall clock value (in milliseconds) of the last activity that occurred. Needed for periodic PING.

uint32_t state: Internal. Client’s state in the connection.

uint32_t rx_buf_datalen: Internal. Packet length read so far.

uint32_t remaining_payload: Internal. Remaining payload length to read.

struct mqtt_client

#include <mqtt.h>

MQTT Client definition to maintain information relevant to the client.

Public Members

struct mqtt_internal internal: MQTT client internal state.

struct mqtt_transport transport: MQTT transport configuration and data.

struct mqtt_utf8 client_id: Unique client identification to be used for the connection.

const void *broker: Broker details, for example, address, port. Address type should be compatible with transport used.

struct mqtt_utf8 *user_name: User name (if any) to be used for the connection. NULL indicates no user name.

struct mqtt_utf8 *password: Password (if any) to be used for the connection. Note that if password is provided, user name shall also be provided. NULL indicates no password.

struct mqtt_topic *will_topic: Will topic and QoS. Can be NULL.

struct mqtt_utf8 *will_message: Will message. Can be NULL. Non NULL value valid only if will topic is not NULL.

mqtt_evt_cb_t evt_cb: Application callback registered with the module to get MQTT events.

uint8_t *rx_buf: Receive buffer used for MQTT packet reception in RX path.

uint32_t rx_buf_size: Size of receive buffer.

uint8_t *tx_buf: Transmit buffer used for creating MQTT packet in TX path.

uint32_t tx_buf_size: Size of transmit buffer.

uint16_t keepalive: Keepalive interval for this client in seconds. Default is CONFIG_MQTT_KEEPALIVE.

uint8_t protocol_version: MQTT protocol version.

int8_t unacked_ping: Unanswered PINGREQ count on this connection.

uint8_t will_retain: Will retain flag, 1 if will message shall be retained persistently.

uint8_t clean_session: Clean session flag indicating a fresh (1) or a retained session (0). Default is CONFIG_MQTT_CLEAN_SESSION.

MQTT

Overview

Sample usage

Using MQTT with TLS

API Reference

Overview 

Sample usage 

Using MQTT with TLS 

API Reference 