Implements most user-facing functions of the MQTT library. More...

#include "iot_config.h"
#include <string.h>
#include "private/iot_mqtt_internal.h"
#include "platform/iot_clock.h"
#include "platform/iot_threads.h"
#include "iot_atomic.h"

Macros
#define	MQTT_LIBRARY_UNINITIALIZED ( ( uint32_t ) 0 )
	Uninitialized value for _initCalled.

#define	MQTT_LIBRARY_INITIALIZED ( ( uint32_t ) 1 )
	Initialized value for _initCalled.

#define	IOT_MQTT_PREVIOUS_SUBSCRIPTIONS_PACKET_ID 1
	Place holder packet identifier used when _IotMqtt_AddSubscriptions is called with previous subscriptions lists. Any non-zero value is acceptable, since this value is never sent out to the broker.

Functions
static bool	_checkInit (void)
	Check if the library is initialized. More...

static bool	_mqttSubscription_setUnsubscribe (const IotLink_t const pSubscriptionLink, void pMatch)
	Set the unsubscribed flag of an MQTT subscription. More...

static void	_mqttSubscription_tryDestroy (void *pData)
	Destroy an MQTT subscription if its reference count is 0. More...

static void	_mqttOperation_tryDestroy (void *pData)
	Decrement the reference count of an MQTT operation and attempt to destroy it. More...

static bool	_createKeepAliveOperation (const IotMqttNetworkInfo_t pNetworkInfo, uint16_t keepAliveSeconds, _mqttConnection_t pMqttConnection)
	Initialize the keep-alive operation for an MQTT connection. More...

static IotNetworkError_t	_createNetworkConnection (const IotMqttNetworkInfo_t pNetworkInfo, IotNetworkConnection_t pNetworkConnection, bool *pCreatedNewNetworkConnection)
	Initialize a network connection, creating it if necessary. More...

static _mqttConnection_t *	_createMqttConnection (bool awsIotMqttMode, const IotMqttNetworkInfo_t *pNetworkInfo, uint16_t keepAliveSeconds)
	Creates a new MQTT connection and initializes its members. More...

static void	_destroyMqttConnection (_mqttConnection_t *pMqttConnection)
	Destroys the members of an MQTT connection. More...

static IotMqttError_t	_subscriptionCommonSetup (IotMqttOperationType_t operation, IotMqttConnection_t mqttConnection, const IotMqttSubscription_t pSubscriptionList, size_t subscriptionCount, uint32_t flags, const IotMqttOperation_t const pOperationReference)
	Common setup function for subscribe and unsubscribe operations. More...

static IotMqttError_t	_subscriptionCreateAndSerialize (IotMqttOperationType_t operation, IotMqttConnection_t mqttConnection, IotMqttSerializeSubscribe_t serializeSubscription, const IotMqttSubscription_t pSubscriptionList, size_t subscriptionCount, uint32_t flags, const IotMqttCallbackInfo_t pCallbackInfo, _mqttOperation_t **ppSubscriptionOperation)
	Utility function for creating and serializing subscription requests. More...

static IotMqttError_t	_sendMqttMessage (_mqttOperation_t *pMqttOperation, uint32_t flags)
	Utility function for sending/scheduling a subscribe, unsubscribe or publish message. More...

static IotMqttError_t	_subscriptionCommon (IotMqttOperationType_t operation, IotMqttConnection_t mqttConnection, IotMqttSerializeSubscribe_t serializeSubscription, const IotMqttSubscription_t pSubscriptionList, size_t subscriptionCount, uint32_t flags, const IotMqttCallbackInfo_t pCallbackInfo, IotMqttOperation_t *const pOperationReference)
	The common component of both IotMqtt_SubscribeAsync and IotMqtt_UnsubscribeAsync. More...

static void	_setOperationReference (IotMqttOperation_t const pOperationReference, _mqttOperation_t pNewOperation)
	Set an operation reference if provided. More...

static IotMqttError_t	_waitForOperation (IotMqttOperation_t operation, uint32_t timeoutMs)
	Wait for an MQTT operation to complete. More...

static IotMqttError_t	_scheduleKeepAlive (IotMqttConnection_t pMqttConnection)
	Utility function for scheduling ping request after connection with with the broker is established. More...

static IotMqttError_t	_sendConnectRequest (_mqttOperation_t *pOperation, uint32_t timeoutMs)
	Utility function for sending connect request. More...

static IotMqttError_t	_addSubscriptions (_mqttOperation_t pOperation, IotMqttConnection_t pMqttConnection, const IotMqttConnectInfo_t pConnectInfo)
	Utility function for adding subscriptions to connect request. More...

static void	_handleConnectFailure (IotMqttConnection_t pMqttConnection, IotNetworkConnection_t pNetworkConnection, const IotMqttNetworkInfo_t pNetworkInfo, _mqttOperation_t pOperation, bool ownNetworkConnection)
	Utility function for handling MQTT connect failure. More...

bool	_IotMqtt_IncrementConnectionReferences (_mqttConnection_t *pMqttConnection)
	Attempt to increment the reference count of an MQTT connection. More...

void	_IotMqtt_DecrementConnectionReferences (_mqttConnection_t *pMqttConnection)
	Decrement the reference count of an MQTT connection. More...

IotMqttError_t	IotMqtt_Init (void)
	One-time initialization function for the MQTT library. More...

void	IotMqtt_Cleanup (void)
	One-time deinitialization function for the MQTT library. More...

IotMqttError_t	IotMqtt_Connect (const IotMqttNetworkInfo_t pNetworkInfo, const IotMqttConnectInfo_t pConnectInfo, uint32_t timeoutMs, IotMqttConnection_t *const pMqttConnection)
	Establish a new MQTT connection. More...

void	IotMqtt_Disconnect (IotMqttConnection_t mqttConnection, uint32_t flags)
	Closes an MQTT connection and frees resources. More...

IotMqttError_t	IotMqtt_SubscribeAsync (IotMqttConnection_t mqttConnection, const IotMqttSubscription_t pSubscriptionList, size_t subscriptionCount, uint32_t flags, const IotMqttCallbackInfo_t pCallbackInfo, IotMqttOperation_t *const pSubscribeOperation)
	Subscribes to the given array of topic filters and optionally receive an asynchronous notification when the subscribe completes. More...

IotMqttError_t	IotMqtt_SubscribeSync (IotMqttConnection_t mqttConnection, const IotMqttSubscription_t *pSubscriptionList, size_t subscriptionCount, uint32_t flags, uint32_t timeoutMs)
	Subscribes to the given array of topic filters with a timeout. More...

IotMqttError_t	IotMqtt_UnsubscribeAsync (IotMqttConnection_t mqttConnection, const IotMqttSubscription_t pSubscriptionList, size_t subscriptionCount, uint32_t flags, const IotMqttCallbackInfo_t pCallbackInfo, IotMqttOperation_t *const pUnsubscribeOperation)
	Unsubscribes from the given array of topic filters and optionally receive an asynchronous notification when the unsubscribe completes. More...

IotMqttError_t	IotMqtt_UnsubscribeSync (IotMqttConnection_t mqttConnection, const IotMqttSubscription_t *pSubscriptionList, size_t subscriptionCount, uint32_t flags, uint32_t timeoutMs)
	Unsubscribes from a given array of topic filters with a timeout. More...

IotMqttError_t	IotMqtt_PublishAsync (IotMqttConnection_t mqttConnection, const IotMqttPublishInfo_t pPublishInfo, uint32_t flags, const IotMqttCallbackInfo_t pCallbackInfo, IotMqttOperation_t *const pPublishOperation)
	Publishes a message to the given topic name and optionally receive an asynchronous notification when the publish completes. More...

IotMqttError_t	IotMqtt_PublishSync (IotMqttConnection_t mqttConnection, const IotMqttPublishInfo_t *pPublishInfo, uint32_t flags, uint32_t timeoutMs)
	Publish a message to the given topic name with a timeout. More...

IotMqttError_t	IotMqtt_Wait (IotMqttOperation_t operation, uint32_t timeoutMs)
	Waits for an operation to complete. More...

const char *	IotMqtt_strerror (IotMqttError_t status)
	Returns a string that describes an IotMqttError_t. More...

const char *	IotMqtt_OperationType (IotMqttOperationType_t operation)
	Returns a string that describes an IotMqttOperationType_t. More...

Variables
static volatile uint32_t	_initCalled = MQTT_LIBRARY_UNINITIALIZED
	Tracks whether IotMqtt_Init has been called. More...

Detailed Description

Implements most user-facing functions of the MQTT library.

Function Documentation

◆ _checkInit()

static bool _checkInit ( void )

static

Check if the library is initialized.

Returns: true if IotMqtt_Init was called; false otherwise.

◆ _mqttSubscription_setUnsubscribe()

static bool _mqttSubscription_setUnsubscribe	(	const IotLink_t *const	pSubscriptionLink,
		void *	pMatch
	)

static

Set the unsubscribed flag of an MQTT subscription.

Parameters

[in]	pSubscriptionLink	Pointer to the link member of an _mqttSubscription_t.
[in]	pMatch	Not used.

Returns: Always returns true.

◆ _mqttSubscription_tryDestroy()

static void _mqttSubscription_tryDestroy ( void * pData )

static

Destroy an MQTT subscription if its reference count is 0.

Parameters

[in] pData The subscription to destroy. This parameter is of type void* for compatibility with free.

◆ _mqttOperation_tryDestroy()

static void _mqttOperation_tryDestroy ( void * pData )

static

Decrement the reference count of an MQTT operation and attempt to destroy it.

Parameters

[in] pData The operation data to destroy. This parameter is of type void* for compatibility with free.

◆ _createKeepAliveOperation()

static bool _createKeepAliveOperation	(	const IotMqttNetworkInfo_t *	pNetworkInfo,
		uint16_t	keepAliveSeconds,
		_mqttConnection_t *	pMqttConnection
	)

static

Initialize the keep-alive operation for an MQTT connection.

Parameters

[in]	pNetworkInfo	User-provided network information for the new connection.
[in]	keepAliveSeconds	User-provided keep-alive interval.
[out]	pMqttConnection	The MQTT connection associated with the keep-alive.

Returns: true if the keep-alive job was successfully created; false otherwise.

◆ _createNetworkConnection()

static IotNetworkError_t _createNetworkConnection	(	const IotMqttNetworkInfo_t *	pNetworkInfo,
		IotNetworkConnection_t *	pNetworkConnection,
		bool *	pCreatedNewNetworkConnection
	)

static

Initialize a network connection, creating it if necessary.

Parameters

[in]	pNetworkInfo	User-provided network information for the connection connection.
[out]	pNetworkConnection	On success, the created and/or initialized network connection.
[out]	pCreatedNewNetworkConnection	On success, `true` if a new network connection was created; `false` if an existing one will be used.

Returns: Any IotNetworkError_t, as defined by the network stack.

◆ _createMqttConnection()

static _mqttConnection_t * _createMqttConnection	(	bool	awsIotMqttMode,
		const IotMqttNetworkInfo_t *	pNetworkInfo,
		uint16_t	keepAliveSeconds
	)

static

Creates a new MQTT connection and initializes its members.

Parameters

[in]	awsIotMqttMode	Specifies if this connection is to an AWS IoT MQTT server.
[in]	pNetworkInfo	User-provided network information for the new connection.
[in]	keepAliveSeconds	User-provided keep-alive interval for the new connection.

Returns: Pointer to a newly-created MQTT connection; NULL on failure.

◆ _destroyMqttConnection()

static void _destroyMqttConnection ( _mqttConnection_t * pMqttConnection )

static

Destroys the members of an MQTT connection.

Parameters

[in] pMqttConnection Which connection to destroy.

◆ _subscriptionCommonSetup()

static IotMqttError_t _subscriptionCommonSetup	(	IotMqttOperationType_t	operation,
		IotMqttConnection_t	mqttConnection,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		const IotMqttOperation_t *const	pOperationReference
	)

static

Common setup function for subscribe and unsubscribe operations.

See IotMqtt_SubscribeAsync or IotMqtt_UnsubscribeAsync for a description of the parameters and return values.

◆ _subscriptionCreateAndSerialize()

static IotMqttError_t _subscriptionCreateAndSerialize	(	IotMqttOperationType_t	operation,
		IotMqttConnection_t	mqttConnection,
		IotMqttSerializeSubscribe_t	serializeSubscription,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		const IotMqttCallbackInfo_t *	pCallbackInfo,
		_mqttOperation_t **	ppSubscriptionOperation
	)

static

Utility function for creating and serializing subscription requests.

See IotMqtt_SubscribeAsync or IotMqtt_UnsubscribeAsync for a description of the parameters and return values.

◆ _sendMqttMessage()

static IotMqttError_t _sendMqttMessage	(	_mqttOperation_t *	pMqttOperation,
		uint32_t	flags
	)

static

Utility function for sending/scheduling a subscribe, unsubscribe or publish message.

Parameters

[in]	pMqttOperation	Reference to MQTT operation.
[in]	flags	Flags which modify the behavior of this function.

Returns: IOT_MQTT_SUCCESS or IOT_MQTT_SCHEDULING_ERROR.

◆ _subscriptionCommon()

static IotMqttError_t _subscriptionCommon	(	IotMqttOperationType_t	operation,
		IotMqttConnection_t	mqttConnection,
		IotMqttSerializeSubscribe_t	serializeSubscription,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		const IotMqttCallbackInfo_t *	pCallbackInfo,
		IotMqttOperation_t *const	pOperationReference
	)

static

The common component of both IotMqtt_SubscribeAsync and IotMqtt_UnsubscribeAsync.

See IotMqtt_SubscribeAsync or IotMqtt_UnsubscribeAsync for a description of the parameters and return values.

◆ _setOperationReference()

static void _setOperationReference	(	IotMqttOperation_t *const	pOperationReference,
		_mqttOperation_t *	pNewOperation
	)

static

Set an operation reference if provided.

Parameters

[out]	pOperationReference	Operation reference provided by the application.
[in]	pNewOperation	Operation to set.

◆ _waitForOperation()

static IotMqttError_t _waitForOperation	(	IotMqttOperation_t	operation,
		uint32_t	timeoutMs
	)

static

Wait for an MQTT operation to complete.

See IotMqtt_Wait for a description of the parameters and return values.

◆ _scheduleKeepAlive()

static IotMqttError_t _scheduleKeepAlive ( IotMqttConnection_t pMqttConnection )

static

Utility function for scheduling ping request after connection with with the broker is established.

Parameters

[in] pMqttConnection MQTT connection reference.

Returns: IOT_MQTT_SUCCESS, IOT_MQTT_SCHEDULING_ERROR.

◆ _sendConnectRequest()

static IotMqttError_t _sendConnectRequest	(	_mqttOperation_t *	pOperation,
		uint32_t	timeoutMs
	)

static

Utility function for sending connect request.

Parameters

[in]	pOperation	CONNECT operation reference.
[in]	timeoutMs	How many milliseconds to wait for CONN_ACK.

Returns: IOT_MQTT_SUCCESS, IOT_MQTT_BAD_PARAMETER, IOT_MQTT_NETWORK_ERROR.

◆ _addSubscriptions()

static IotMqttError_t _addSubscriptions	(	_mqttOperation_t *	pOperation,
		IotMqttConnection_t	pMqttConnection,
		const IotMqttConnectInfo_t *	pConnectInfo
	)

static

Utility function for adding subscriptions to connect request.

Parameters

[in]	pOperation	CONNECT operation reference.
[in]	pMqttConnection	MQTT connection reference.
[in]	pConnectInfo	MQTT connection information.

Returns: IOT_MQTT_SUCCESS, IOT_MQTT_NO_MEMORY

◆ _handleConnectFailure()

static void _handleConnectFailure	(	IotMqttConnection_t	pMqttConnection,
		IotNetworkConnection_t	pNetworkConnection,
		const IotMqttNetworkInfo_t *	pNetworkInfo,
		_mqttOperation_t *	pOperation,
		bool	ownNetworkConnection
	)

static

Utility function for handling MQTT connect failure.

Parameters

[in]	pMqttConnection	MQTT connection reference.
[in]	pNetworkConnection	Network connection reference.
[in]	pNetworkInfo	User-provided network information.
[in]	pOperation	CONNECT operation reference.
[in]	ownNetworkConnection	if true, connection needs to be closed.

◆ _IotMqtt_IncrementConnectionReferences()

bool _IotMqtt_IncrementConnectionReferences ( _mqttConnection_t * pMqttConnection )

Attempt to increment the reference count of an MQTT connection.

Parameters

[in] pMqttConnection The referenced MQTT connection.

Returns: true if the reference count was incremented; false otherwise. The reference count will not be incremented for a disconnected connection.

◆ _IotMqtt_DecrementConnectionReferences()

void _IotMqtt_DecrementConnectionReferences ( _mqttConnection_t * pMqttConnection )

Decrement the reference count of an MQTT connection.

Also destroys an unreferenced MQTT connection.

Parameters

[in] pMqttConnection The referenced MQTT connection.

◆ IotMqtt_Init()

IotMqttError_t IotMqtt_Init ( void )

One-time initialization function for the MQTT library.

This function performs setup of the MQTT library. It must be called once (and only once) before calling any other MQTT function. Calling this function more than once without first calling IotMqtt_Cleanup may result in a crash.

Returns

One of the following:

Warning: No thread-safety guarantees are provided for this function.

See also: IotMqtt_Cleanup

◆ IotMqtt_Cleanup()

void IotMqtt_Cleanup ( void )

One-time deinitialization function for the MQTT library.

This function frees resources taken in IotMqtt_Init. It should be called after closing all MQTT connections to clean up the MQTT library. After this function returns, IotMqtt_Init must be called again before calling any other MQTT function.

Warning: No thread-safety guarantees are provided for this function. Do not call this function if any MQTT connections are open!

See also: IotMqtt_Init

◆ IotMqtt_Connect()

IotMqttError_t IotMqtt_Connect	(	const IotMqttNetworkInfo_t *	pNetworkInfo,
		const IotMqttConnectInfo_t *	pConnectInfo,
		uint32_t	timeoutMs,
		IotMqttConnection_t *const	pMqttConnection
	)

Establish a new MQTT connection.

This function opens a connection between a new MQTT client and an MQTT server (also called a broker). MQTT connections are established on top of transport layer protocols (such as TCP/IP), and optionally, application layer security protocols (such as TLS). The MQTT packet that establishes a connection is called the MQTT CONNECT packet. After IotMqtt_Init, this function must be called before any other MQTT library function.

If pConnectInfo->cleanSession is true, this function establishes a clean MQTT session. Subscriptions and unacknowledged PUBLISH messages will be discarded when the connection is closed.

If pConnectInfo->cleanSession is false, this function establishes (or re-establishes) a persistent MQTT session. The parameters pConnectInfo->pPreviousSubscriptions and pConnectInfo->previousSubscriptionCount may be used to restore subscriptions present in a re-established persistent session. Any restored subscriptions MUST have been present in the persistent session; this function does not send an MQTT SUBSCRIBE packet!

pConnectInfo->pPreviousSubscriptions and pConnectInfo->previousSubscriptionCount can also be used to pass a list of subscriptions to be stored locally without a SUBSCRIBE packet being sent to the broker. These subscriptions are useful to invoke application level callbacks for messages received on unsolicited topics from the broker.

This MQTT library is network agnostic, meaning it has no knowledge of the underlying network protocol carrying the MQTT packets. It interacts with the network through a network abstraction layer, allowing it to be used with many different network stacks. The network abstraction layer is established per-connection, allowing every IotMqttConnection_t to use a different network stack. The parameter pNetworkInterface sets up the network abstraction layer for an MQTT connection; see the documentation on IotMqttNetworkInfo_t for details on its members.

The pConnectInfo parameter provides the contents of the MQTT CONNECT packet. Most members are defined by the MQTT spec.. The pConnectInfo->pWillInfo member provides information on a Last Will and Testament (LWT) message to be published if the MQTT connection is closed without sending a DISCONNECT packet. Unlike other PUBLISH messages, a LWT message payload is limited to 65535 bytes in length. Additionally, the retry interval and limit members of IotMqttPublishInfo_t are ignored for LWT messages. The LWT message is optional; pWillInfo may be NULL.

MQTT keep-alive is configured by IotMqttConnectInfo_t::keepAliveSeconds. Keep-alive is used to detect half-open or otherwise unusable network connections. An MQTT client will send periodic ping requests (PINGREQ) to the server if the connection is idle. The MQTT server must respond to ping requests with a ping response (PINGRESP). The standard does not specify how long the server has to respond to a ping request, noting only a "reasonable amount of time". In this library, the amount of time a server has to respond to a ping request is set with IOT_MQTT_RESPONSE_WAIT_MS.

Unlike IotMqtt_PublishAsync, IotMqtt_SubscribeAsync, and IotMqtt_UnsubscribeAsync, this function is always blocking. Additionally, because the MQTT connection acknowledgement packet (CONNACK packet) does not contain any information on which CONNECT packet it acknowledges, only one CONNECT operation may be in progress at any time. This means that parallel threads making calls to IotMqtt_Connect will be serialized to send their CONNECT packets one-by-one.

Parameters

[in]	pNetworkInfo	Information on the transport-layer network connection to use with the MQTT connection.
[in]	pConnectInfo	MQTT connection setup parameters.
[in]	timeoutMs	If the MQTT server does not accept the connection within this timeout in milliseconds, this function returns IOT_MQTT_TIMEOUT.
[out]	pMqttConnection	Set to a newly-initialized MQTT connection handle if this function succeeds.

Returns

One of the following:

Example

// Callback function to receive messages from the broker on an unsolicited topic.
void unsolicitedMessageCallback( void * pArgument, IotMqttCallbackParam_t * pPublish );
// Parameters to MQTT connect.
IotMqttConnection_t mqttConnection = IOT_MQTT_CONNECTION_INITIALIZER;
IotMqttNetworkInfo_t networkInfo = IOT_MQTT_NETWORK_INFO_INITIALIZER;
IotMqttConnectInfo_t connectInfo = IOT_MQTT_CONNECT_INFO_INITIALIZER;
IotMqttPublishInfo_t willInfo = IOT_MQTT_PUBLISH_INFO_INITIALIZER;
// A local subscription to receive messages from the broker on an unsolicited topic.
IotMqttSubscription_t subscription = IOT_MQTT_SUBSCRIPTION_INITIALIZER;
// Example network abstraction types.
IotNetworkServerInfo_t serverInfo = { ... };
IotNetworkCredentials_t credentialInfo = { ... };
IotNetworkInterface_t networkInterface = { ... };
// Example using a generic network implementation.
networkInfo.createNetworkConnection = true;
networkInfo.u.setup.pNetworkServerInfo = &serverInfo;
networkInfo.u.setup.pNetworkCredentialInfo = &credentialInfo;
networkInfo.pNetworkInterface = &networkInterface;
// Set the members of the connection info (password and username not used).
connectInfo.cleanSession = true;
connectInfo.keepAliveSeconds = 30;
connectInfo.pClientIdentifier = "uniqueclientidentifier";
connectInfo.clientIdentifierLength = 22;
// Set the members of the will info (retain and retry not used).
willInfo.qos = IOT_MQTT_QOS_1;
willInfo.pTopicName = "will/topic/name";
willInfo.topicNameLength = ( uint16_t ) strlen( willInfo.pTopicName );
willInfo.pPayload = "MQTT client unexpectedly disconnected.";
willInfo.payloadLength = strlen( willInfo.pPayload );
// Set the pointer to the will info.
connectInfo.pWillInfo = &willInfo;
// [Optional] Set a local subscription to receive the broker messages on an unsolicited topic.
subscription.qos = IOT_MQTT_QOS_0;
subscription.pTopicFilter = "some/unsolicited/topic";
subscription.topicLength = ( uint16_t ) strlen( subscription.pTopicFilter );
subscription.callback.function = unsolicitedMessageCallback;
connectInfo.pPreviousSubscriptions = &subscription;
connectInfo.previousSubscriptionCount = 1;
// Call CONNECT with a 5 second block time. Should return
// IOT_MQTT_SUCCESS when successful.
IotMqttError_t result = IotMqtt_Connect( &networkInfo,
                                         &connectInfo,
                                         5000,
                                         &mqttConnection );
if( result == IOT_MQTT_SUCCESS )
{
    // Do something with the MQTT connection...
    // Clean up and close the MQTT connection once it's no longer needed.
    IotMqtt_Disconnect( mqttConnection, 0 );
}

◆ IotMqtt_Disconnect()

void IotMqtt_Disconnect	(	IotMqttConnection_t	mqttConnection,
		uint32_t	flags
	)

Closes an MQTT connection and frees resources.

This function closes an MQTT connection and should only be called once the MQTT connection is no longer needed. Its exact behavior depends on the flags parameter.

Normally, flags should be 0. This gracefully shuts down an MQTT connection by sending an MQTT DISCONNECT packet. Any network close function provided when the connection was established will also be called. Note that because the MQTT server will not acknowledge a DISCONNECT packet, the client has no way of knowing if the server received the DISCONNECT packet. In the case where the DISCONNECT packet is lost in transport, any Last Will and Testament (LWT) message established with the connection may be published. However, if the DISCONNECT reaches the MQTT server, the LWT message will be discarded and not published.

Should the underlying network connection become unusable, this function should be called with flags set to IOT_MQTT_FLAG_CLEANUP_ONLY. In this case, no DISCONNECT packet will be sent, though the network close function will still be called. This function will only free the resources used by the MQTT connection; it still must be called even if the network is offline to avoid leaking resources.

IotMqtt_Disconnect modifies mqttConnection, so it shouldn't be used after calling this function.

Parameters

[in]	mqttConnection	The MQTT connection to close and clean up.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags.

◆ IotMqtt_SubscribeAsync()

IotMqttError_t IotMqtt_SubscribeAsync	(	IotMqttConnection_t	mqttConnection,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		const IotMqttCallbackInfo_t *	pCallbackInfo,
		IotMqttOperation_t *const	pSubscribeOperation
	)

Subscribes to the given array of topic filters and optionally receive an asynchronous notification when the subscribe completes.

This function sends an MQTT SUBSCRIBE packet to the server. A SUBSCRIBE packet notifies the server to send any matching PUBLISH messages to this client. A single SUBSCRIBE packet may carry more than one topic filter, hence the parameters to this function include an array of subscriptions.

An MQTT subscription has two pieces:

The subscription topic filter registered with the MQTT server. The MQTT SUBSCRIBE packet sent from this client to server notifies the server to send messages matching the given topic filters to this client.
The callback function that this client will invoke when an incoming message is received. The callback function notifies applications of an incoming PUBLISH message.

The helper function IotMqtt_IsSubscribed can be used to check if a callback function is registered for a particular topic filter.

To modify an already-registered subscription callback, call this function with a new pSubscriptionList. Any topic filters in pSubscriptionList that already have a registered callback will be replaced with the new values in pSubscriptionList.

Attention: QoS 2 subscriptions are currently unsupported. Only 0 or 1 are valid for subscription QoS.

Parameters

[in]	mqttConnection	The MQTT connection to use for the subscription.
[in]	pSubscriptionList	Pointer to the first element in the array of subscriptions.
[in]	subscriptionCount	The number of elements in pSubscriptionList.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags.
[in]	pCallbackInfo	Asynchronous notification of this function's completion (`NULL` to disable).
[out]	pSubscribeOperation	Set to a handle by which this operation may be referenced after this function returns. This reference is invalidated once the subscription operation completes.

Returns

This function will return IOT_MQTT_STATUS_PENDING upon success.

Upon completion of the subscription (either through an IotMqttCallbackInfo_t or IotMqtt_Wait), the status will be one of:

If this function fails before queuing a subscribe operation, it will return one of:

See also: IotMqtt_SubscribeSync for a blocking variant of this function.; IotMqtt_UnsubscribeAsync for the function that removes subscriptions.

Example

#define NUMBER_OF_SUBSCRIPTIONS ...
// Subscription callback function.
void subscriptionCallback( void * pArgument, IotMqttCallbackParam_t * pPublish );
// An initialized and connected MQTT connection.
IotMqttConnection_t mqttConnection;
// Subscription information.
pSubscriptions[ NUMBER_OF_SUBSCRIPTIONS ] = { IOT_MQTT_SUBSCRIPTION_INITIALIZER };
IotMqttOperation_t lastOperation = IOT_MQTT_OPERATION_INITIALIZER;
// Set the subscription information.
for( int i = 0; i < NUMBER_OF_SUBSCRIPTIONS; i++ )
{
    pSubscriptions[ i ].qos = IOT_MQTT_QOS_1;
    pSubscriptions[ i ].pTopicFilter = "some/topic/filter";
    pSubscriptions[ i ].topicLength = ( uint16_t ) strlen( pSubscriptions[ i ].pTopicFilter );
    pSubscriptions[ i ].callback.function = subscriptionCallback;
}
IotMqttError_t result = IotMqtt_SubscribeAsync( mqttConnection,
                                                pSubscriptions,
                                                NUMBER_OF_SUBSCRIPTIONS,
                                                IOT_MQTT_FLAG_WAITABLE,
                                                NULL,
                                                &lastOperation );
// Subscribe returns IOT_MQTT_STATUS_PENDING when successful. Wait up to
// 5 seconds for the operation to complete.
if( result == IOT_MQTT_STATUS_PENDING )
{
    result = IotMqtt_Wait( subscriptionRef, 5000 );
}
// Check that the subscriptions were successful.
if( result == IOT_MQTT_SUCCESS )
{
    // Wait for messages on the subscription topic filters...
    // Unsubscribe once the subscriptions are no longer needed.
    result = IotMqtt_UnsubscribeAsync( mqttConnection,
                                       pSubscriptions,
                                       NUMBER_OF_SUBSCRIPTIONS,
                                       IOT_MQTT_FLAG_WAITABLE,
                                       NULL,
                                       &lastOperation );
    // UNSUBSCRIBE returns IOT_MQTT_STATUS_PENDING when successful.
    // Wait up to 5 seconds for the operation to complete.
    if( result == IOT_MQTT_STATUS_PENDING )
    {
        result = IotMqtt_Wait( lastOperation, 5000 );
    }
}
// Check which subscriptions were rejected by the server.
else if( result == IOT_MQTT_SERVER_REFUSED )
{
    for( int i = 0; i < NUMBER_OF_SUBSCRIPTIONS; i++ )
    {
        if( IotMqtt_IsSubscribed( mqttConnection,
                                  pSubscriptions[ i ].pTopicFilter,
                                  pSubscriptions[ i ].topicFilterLength,
                                  NULL ) == false )
        {
            // This subscription was rejected.
        }
    }
}

◆ IotMqtt_SubscribeSync()

IotMqttError_t IotMqtt_SubscribeSync	(	IotMqttConnection_t	mqttConnection,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		uint32_t	timeoutMs
	)

Subscribes to the given array of topic filters with a timeout.

This function sends an MQTT SUBSCRIBE packet to the server, then waits for a server response to the packet. Internally, this function is a call to IotMqtt_SubscribeAsync followed by IotMqtt_Wait. See IotMqtt_SubscribeAsync for more information about the MQTT SUBSCRIBE operation.

Attention: QoS 2 subscriptions are currently unsupported. Only 0 or 1 are valid for subscription QoS.

Parameters

[in]	mqttConnection	The MQTT connection to use for the subscription.
[in]	pSubscriptionList	Pointer to the first element in the array of subscriptions.
[in]	subscriptionCount	The number of elements in pSubscriptionList.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags. Currently, flags are ignored by this function; this parameter is for future-compatibility.
[in]	timeoutMs	If the MQTT server does not acknowledge the subscriptions within this timeout in milliseconds, this function returns IOT_MQTT_TIMEOUT.

Returns

One of the following:

◆ IotMqtt_UnsubscribeAsync()

IotMqttError_t IotMqtt_UnsubscribeAsync	(	IotMqttConnection_t	mqttConnection,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		const IotMqttCallbackInfo_t *	pCallbackInfo,
		IotMqttOperation_t *const	pUnsubscribeOperation
	)

Unsubscribes from the given array of topic filters and optionally receive an asynchronous notification when the unsubscribe completes.

This function sends an MQTT UNSUBSCRIBE packet to the server. An UNSUBSCRIBE packet removes registered topic filters from the server. After unsubscribing, the server will no longer send messages on these topic filters to the client.

Corresponding subscription callback functions are also removed from the MQTT connection. These subscription callback functions will be removed even if the MQTT UNSUBSCRIBE packet fails to send.

Parameters

[in]	mqttConnection	The MQTT connection used for the subscription.
[in]	pSubscriptionList	Pointer to the first element in the array of subscriptions.
[in]	subscriptionCount	The number of elements in pSubscriptionList.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags.
[in]	pCallbackInfo	Asynchronous notification of this function's completion (`NULL` to disable).
[out]	pUnsubscribeOperation	Set to a handle by which this operation may be referenced after this function returns. This reference is invalidated once the unsubscribe operation completes.

Returns

This function will return IOT_MQTT_STATUS_PENDING upon success.

Upon completion of the unsubscribe (either through an IotMqttCallbackInfo_t or IotMqtt_Wait), the status will be one of:

If this function fails before queuing an unsubscribe operation, it will return one of:

See also: IotMqtt_UnsubscribeSync for a blocking variant of this function.; IotMqtt_SubscribeAsync for the function that adds subscriptions.

◆ IotMqtt_UnsubscribeSync()

IotMqttError_t IotMqtt_UnsubscribeSync	(	IotMqttConnection_t	mqttConnection,
		const IotMqttSubscription_t *	pSubscriptionList,
		size_t	subscriptionCount,
		uint32_t	flags,
		uint32_t	timeoutMs
	)

Unsubscribes from a given array of topic filters with a timeout.

This function sends an MQTT UNSUBSCRIBE packet to the server, then waits for a server response to the packet. Internally, this function is a call to IotMqtt_UnsubscribeAsync followed by IotMqtt_Wait. See IotMqtt_UnsubscribeAsync for more information about the MQTT UNSUBSCRIBE operation.

Parameters

[in]	mqttConnection	The MQTT connection used for the subscription.
[in]	pSubscriptionList	Pointer to the first element in the array of subscriptions.
[in]	subscriptionCount	The number of elements in pSubscriptionList.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags. Flags are currently ignored but reserved for future use.
[in]	timeoutMs	If the MQTT server does not acknowledge the UNSUBSCRIBE within this timeout in milliseconds, this function returns IOT_MQTT_TIMEOUT.

Returns

One of the following:

◆ IotMqtt_PublishAsync()

IotMqttError_t IotMqtt_PublishAsync	(	IotMqttConnection_t	mqttConnection,
		const IotMqttPublishInfo_t *	pPublishInfo,
		uint32_t	flags,
		const IotMqttCallbackInfo_t *	pCallbackInfo,
		IotMqttOperation_t *const	pPublishOperation
	)

Publishes a message to the given topic name and optionally receive an asynchronous notification when the publish completes.

This function sends an MQTT PUBLISH packet to the server. A PUBLISH packet contains a payload and a topic name. Any clients with a subscription on a topic filter matching the PUBLISH topic name will receive a copy of the PUBLISH packet from the server.

If a PUBLISH packet fails to reach the server and it is not a QoS 0 message, it will be retransmitted. See IotMqttPublishInfo_t for a description of the retransmission strategy.

Attention: QoS 2 messages are currently unsupported. Only 0 or 1 are valid for message QoS.

Parameters

[in]	mqttConnection	The MQTT connection to use for the publish.
[in]	pPublishInfo	MQTT publish parameters.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags.
[in]	pCallbackInfo	Asynchronous notification of this function's completion (`NULL` to disable).
[out]	pPublishOperation	Set to a handle by which this operation may be referenced after this function returns. This reference is invalidated once the publish operation completes.

Returns

This function will return IOT_MQTT_STATUS_PENDING upon success for QoS 1 publishes. For a QoS 0 publish it returns IOT_MQTT_SUCCESS upon success.

Upon completion of a QoS 1 publish (either through an IotMqttCallbackInfo_t or IotMqtt_Wait), the status will be one of:

IOT_MQTT_SUCCESS
IOT_MQTT_NETWORK_ERROR
IOT_MQTT_SCHEDULING_ERROR
IOT_MQTT_BAD_RESPONSE
IOT_MQTT_RETRY_NO_RESPONSE (if pPublishInfo->retryMs and pPublishInfo->retryLimit were set).

If this function fails before queuing an publish operation (regardless of QoS), it will return one of:

Note: The parameters pCallbackInfo and pPublishOperation should only be used for QoS 1 publishes. For QoS 0, they should both be NULL.

See also: IotMqtt_PublishSync for a blocking variant of this function.

Example

// An initialized and connected MQTT connection.
IotMqttConnection_t mqttConnection;
// Publish information.
IotMqttPublishInfo_t publishInfo = IOT_MQTT_PUBLISH_INFO_INITIALIZER;
// Set the publish information. QoS 0 example (retain not used):
publishInfo.qos = IOT_MQTT_QOS_0;
publishInfo.pTopicName = "some/topic/name";
publishInfo.topicNameLength = ( uint16_t ) strlen( publishInfo.pTopicName );
publishInfo.pPayload = "payload";
publishInfo.payloadLength = strlen( publishInfo.pPayload );
// QoS 0 publish should return IOT_MQTT_SUCCESS upon success.
IotMqttError_t qos0Result = IotMqtt_PublishAsync( mqttConnection,
                                                  &publishInfo,
                                                  0,
                                                  NULL,
                                                  NULL );
// QoS 1 with retry example (using same topic name and payload as QoS 0 example):
IotMqttOperation_t qos1Operation = IOT_MQTT_OPERATION_INITIALIZER;
publishInfo.qos = IOT_MQTT_QOS_1;
publishInfo.retryMs = 1000; // Retry if no response is received in 1 second.
publishInfo.retryLimit = 5; // Retry up to 5 times.
// QoS 1 publish should return IOT_MQTT_STATUS_PENDING upon success.
IotMqttError_t qos1Result = IotMqtt_PublishAsync( mqttConnection,
                                                  &publishInfo,
                                                  IOT_MQTT_FLAG_WAITABLE,
                                                  NULL,
                                                  &qos1Operation );
// Wait up to 5 seconds for the publish to complete.
if( qos1Result == IOT_MQTT_STATUS_PENDING )
{
    qos1Result = IotMqtt_Wait( qos1Operation, 5000 );
}

◆ IotMqtt_PublishSync()

IotMqttError_t IotMqtt_PublishSync	(	IotMqttConnection_t	mqttConnection,
		const IotMqttPublishInfo_t *	pPublishInfo,
		uint32_t	flags,
		uint32_t	timeoutMs
	)

Publish a message to the given topic name with a timeout.

This function sends an MQTT PUBLISH packet to the server, then waits for a server response to the packet. Internally, this function is a call to IotMqtt_PublishAsync followed by IotMqtt_Wait. See IotMqtt_PublishAsync for more information about the MQTT PUBLISH operation.

Attention: QoS 2 messages are currently unsupported. Only 0 or 1 are valid for message QoS.

Parameters

[in]	mqttConnection	The MQTT connection to use for the publish.
[in]	pPublishInfo	MQTT publish parameters.
[in]	flags	Flags which modify the behavior of this function. See MQTT Function Flags. Currently, flags are ignored by this function; this parameter is for future-compatibility.
[in]	timeoutMs	If the MQTT server does not acknowledge a QoS 1 PUBLISH within this timeout in milliseconds, this function returns IOT_MQTT_TIMEOUT. This parameter is ignored for QoS 0 PUBLISH messages.

Returns

One of the following:

◆ IotMqtt_Wait()

IotMqttError_t IotMqtt_Wait	(	IotMqttOperation_t	operation,
		uint32_t	timeoutMs
	)

Waits for an operation to complete.

This function blocks to wait for a subscribe, unsubscribe, or publish to complete. These operations are by default asynchronous; the function calls queue an operation for processing, and a callback is invoked once the operation is complete.

To use this function, the flag IOT_MQTT_FLAG_WAITABLE must have been set in the operation's function call. Additionally, this function must always be called with any waitable operation to clean up resources.

Regardless of its return value, this function always clean up resources used by the waitable operation. This means reference is invalidated as soon as this function returns, even if it returns IOT_MQTT_TIMEOUT or another error.

Parameters

[in]	operation	Reference to the operation to wait for. The flag IOT_MQTT_FLAG_WAITABLE must have been set for this operation.
[in]	timeoutMs	How many milliseconds to wait before returning IOT_MQTT_TIMEOUT.

Returns: The return value of this function depends on the MQTT operation associated with reference. See IotMqttError_t for possible return values.

Example

// Operation reference and timeout.
IotMqttOperation_t publishOperation = IOT_MQTT_OPERATION_INITIALIZER;
uint32_t timeoutMs = 5000; // 5 seconds
// MQTT operation to wait for.
IotMqttError_t result = IotMqtt_PublishAsync( mqttConnection,
                                              &publishInfo,
                                              IOT_MQTT_FLAG_WAITABLE,
                                              NULL,
                                              &publishOperation );
// Publish should have returned IOT_MQTT_STATUS_PENDING. The call to wait
// returns once the result of the publish is available or the timeout expires.
if( result == IOT_MQTT_STATUS_PENDING )
{
    result = IotMqtt_Wait( publishOperation, timeoutMs );
    // After the call to wait, the result of the publish is known
    // (not IOT_MQTT_STATUS_PENDING).
    assert( result != IOT_MQTT_STATUS_PENDING );
}

◆ IotMqtt_strerror()

const char* IotMqtt_strerror ( IotMqttError_t status )

Returns a string that describes an IotMqttError_t.

Like the POSIX strerror, this function returns a string describing a return code. In this case, the return code is an MQTT library error code, status.

The string returned by this function MUST be treated as read-only: any attempt to modify its contents may result in a crash. Therefore, this function is limited to usage in logging.

Parameters

[in] status The status to describe.

Returns: A read-only string that describes status.

Warning: The string returned by this function must never be modified.

◆ IotMqtt_OperationType()

const char* IotMqtt_OperationType ( IotMqttOperationType_t operation )

Returns a string that describes an IotMqttOperationType_t.

This function returns a string describing an MQTT operation type, operation.

The string returned by this function MUST be treated as read-only: any attempt to modify its contents may result in a crash. Therefore, this function is limited to usage in logging.

Parameters

[in] operation The operation to describe.

Returns: A read-only string that describes operation.

Warning: The string returned by this function must never be modified.

Variable Documentation

◆ _initCalled

volatile uint32_t _initCalled = MQTT_LIBRARY_UNINITIALIZED

static

Tracks whether IotMqtt_Init has been called.

API functions will fail if IotMqtt_Init was not called.

Macros

Functions

Variables

Detailed Description

Function Documentation

◆ _checkInit()

◆ _mqttSubscription_setUnsubscribe()

◆ _mqttSubscription_tryDestroy()

◆ _mqttOperation_tryDestroy()

◆ _createKeepAliveOperation()

◆ _createNetworkConnection()

◆ _createMqttConnection()

◆ _destroyMqttConnection()

◆ _subscriptionCommonSetup()

◆ _subscriptionCreateAndSerialize()

◆ _sendMqttMessage()

◆ _subscriptionCommon()

◆ _setOperationReference()

◆ _waitForOperation()

◆ _scheduleKeepAlive()

◆ _sendConnectRequest()

◆ _addSubscriptions()

◆ _handleConnectFailure()

◆ _IotMqtt_IncrementConnectionReferences()

◆ _IotMqtt_DecrementConnectionReferences()

◆ IotMqtt_Init()

◆ IotMqtt_Cleanup()

◆ IotMqtt_Connect()

◆ IotMqtt_Disconnect()

◆ IotMqtt_SubscribeAsync()

◆ IotMqtt_SubscribeSync()

◆ IotMqtt_UnsubscribeAsync()

◆ IotMqtt_UnsubscribeSync()

◆ IotMqtt_PublishAsync()

◆ IotMqtt_PublishSync()

◆ IotMqtt_Wait()

◆ IotMqtt_strerror()

◆ IotMqtt_OperationType()

Variable Documentation

◆ _initCalled