Implements an MQTT agent (or daemon task) to enable multithreaded access to coreMQTT. More...

#include <string.h>
#include <stdio.h>
#include <assert.h>
#include "core_mqtt_agent.h"
#include "core_mqtt_agent_command_functions.h"
#include "core_mqtt_agent_default_logging.h"

Functions
static MQTTStatus_t	addAwaitingOperation (MQTTAgentContext_t pAgentContext, uint16_t packetId, MQTTAgentCommand_t pCommand)
	Track an operation by adding it to a list, indicating it is anticipating an acknowledgment. More...

static MQTTAgentAckInfo_t *	getAwaitingOperation (MQTTAgentContext_t *pAgentContext, uint16_t incomingPacketId)
	Retrieve an operation from the list of pending acks, and optionally remove it from the list. More...

static MQTTStatus_t	createCommand (MQTTAgentCommandType_t commandType, const MQTTAgentContext_t pMqttAgentContext, void pMqttInfoParam, MQTTAgentCommandCallback_t commandCompleteCallback, MQTTAgentCommandContext_t pCommandCompleteCallbackContext, MQTTAgentCommand_t pCommand)
	Populate the parameters of a MQTTAgentCommand struct. More...

static MQTTStatus_t	addCommandToQueue (const MQTTAgentContext_t pAgentContext, MQTTAgentCommand_t pCommand, uint32_t blockTimeMs)
	Add a command to the global command queue. More...

static MQTTStatus_t	processCommand (MQTTAgentContext_t pMqttAgentContext, MQTTAgentCommand_t pCommand, bool *pEndLoop)
	Process a MQTTAgentCommand struct. More...

static void	mqttEventCallback (MQTTContext_t pMqttContext, MQTTPacketInfo_t pPacketInfo, MQTTDeserializedInfo_t *pDeserializedInfo)
	Dispatch incoming publishes and acks to their various handler functions. More...

static void	handleAcks (const MQTTAgentContext_t pAgentContext, const MQTTPacketInfo_t pPacketInfo, const MQTTDeserializedInfo_t pDeserializedInfo, MQTTAgentAckInfo_t pAckInfo, uint8_t packetType)
	Mark a command as complete after receiving an acknowledgment packet. More...

static MQTTAgentContext_t *	getAgentFromMQTTContext (MQTTContext_t *pMQTTContext)
	Retrieve a pointer to an agent context given an MQTT context. More...

static MQTTStatus_t	createAndAddCommand (MQTTAgentCommandType_t commandType, const MQTTAgentContext_t pMqttAgentContext, void pMqttInfoParam, MQTTAgentCommandCallback_t commandCompleteCallback, MQTTAgentCommandContext_t *pCommandCompleteCallbackContext, uint32_t blockTimeMs)
	Helper function for creating a command and adding it to the command queue. More...

static void	concludeCommand (const MQTTAgentContext_t pAgentContext, MQTTAgentCommand_t pCommand, MQTTStatus_t returnCode, uint8_t *pSubackCodes)
	Helper function to mark a command as complete and invoke its callback. This function calls the releaseCommand callback. More...

static MQTTStatus_t	resendPublishes (MQTTAgentContext_t *pMqttAgentContext)
	Resend QoS 1 and 2 publishes after resuming a session. More...

static void	clearPendingAcknowledgments (MQTTAgentContext_t *pMqttAgentContext, bool clearOnlySubUnsubEntries)
	Clears the list of pending acknowledgments by invoking each callback with MQTTRecvFailed either for ALL operations in the list OR only for Subscribe/Unsubscribe operations. More...

static bool	validateStruct (const MQTTAgentContext_t pMqttAgentContext, const MQTTAgentCommandInfo_t pCommandInfo)
	Validate an MQTTAgentContext_t and a MQTTAgentCommandInfo_t from API functions. More...

static bool	validateParams (MQTTAgentCommandType_t commandType, const void *pParams)
	Validate the parameters for a CONNECT, SUBSCRIBE, UNSUBSCRIBE or PUBLISH. More...

static bool	isSpaceInPendingAckList (const MQTTAgentContext_t *pAgentContext)
	Called before accepting any PUBLISH or SUBSCRIBE messages to check there is space in the pending ACK list for the outgoing PUBLISH or SUBSCRIBE. More...

MQTTStatus_t	MQTTAgent_Init (MQTTAgentContext_t pMqttAgentContext, const MQTTAgentMessageInterface_t pMsgInterface, const MQTTFixedBuffer_t pNetworkBuffer, const TransportInterface_t pTransportInterface, MQTTGetCurrentTimeFunc_t getCurrentTimeMs, MQTTAgentIncomingPublishCallback_t incomingCallback, void *pIncomingPacketContext)
	Perform any initialization the MQTT agent requires before it can be used. Must be called before any other function. More...

MQTTStatus_t	MQTTAgent_CommandLoop (MQTTAgentContext_t *pMqttAgentContext)
	Process commands from the command queue in a loop. More...

MQTTStatus_t	MQTTAgent_ResumeSession (MQTTAgentContext_t *pMqttAgentContext, bool sessionPresent)
	Resume a session by resending publishes if a session is present in the broker, or clear state information if not. More...

MQTTStatus_t	MQTTAgent_CancelAll (MQTTAgentContext_t *pMqttAgentContext)
	Cancel all enqueued commands and those awaiting acknowledgment while the command loop is not running. More...

MQTTStatus_t	MQTTAgent_Subscribe (const MQTTAgentContext_t pMqttAgentContext, MQTTAgentSubscribeArgs_t pSubscriptionArgs, const MQTTAgentCommandInfo_t *pCommandInfo)
	Add a command to call MQTT_Subscribe() for an MQTT connection. More...

MQTTStatus_t	MQTTAgent_Unsubscribe (const MQTTAgentContext_t pMqttAgentContext, MQTTAgentSubscribeArgs_t pSubscriptionArgs, const MQTTAgentCommandInfo_t *pCommandInfo)
	Add a command to call MQTT_Unsubscribe() for an MQTT connection. More...

MQTTStatus_t	MQTTAgent_Publish (const MQTTAgentContext_t pMqttAgentContext, MQTTPublishInfo_t pPublishInfo, const MQTTAgentCommandInfo_t *pCommandInfo)
	Add a command to call MQTT_Publish() for an MQTT connection. More...

MQTTStatus_t	MQTTAgent_ProcessLoop (const MQTTAgentContext_t pMqttAgentContext, const MQTTAgentCommandInfo_t pCommandInfo)
	Send a message to the MQTT agent purely to trigger an iteration of its loop, which will result in a call to MQTT_ProcessLoop(). This function can be used to wake the MQTT agent task when it is known data may be available on the connected socket. More...

MQTTStatus_t	MQTTAgent_Connect (const MQTTAgentContext_t pMqttAgentContext, MQTTAgentConnectArgs_t pConnectArgs, const MQTTAgentCommandInfo_t *pCommandInfo)
	Add a command to call MQTT_Connect() for an MQTT connection. If a session is resumed with the broker, it will also resend the necessary QoS1/2 publishes. More...

MQTTStatus_t	MQTTAgent_Disconnect (const MQTTAgentContext_t pMqttAgentContext, const MQTTAgentCommandInfo_t pCommandInfo)
	Add a command to disconnect an MQTT connection. More...

MQTTStatus_t	MQTTAgent_Ping (const MQTTAgentContext_t pMqttAgentContext, const MQTTAgentCommandInfo_t pCommandInfo)
	Add a command to call MQTT_Ping() for an MQTT connection. More...

MQTTStatus_t	MQTTAgent_Terminate (const MQTTAgentContext_t pMqttAgentContext, const MQTTAgentCommandInfo_t pCommandInfo)
	Add a termination command to the command queue. More...

Detailed Description

Implements an MQTT agent (or daemon task) to enable multithreaded access to coreMQTT.

Note: Implements an MQTT agent (or daemon task) on top of the coreMQTT MQTT client library. The agent makes coreMQTT usage thread safe by being the only task (or thread) in the system that is allowed to access the native coreMQTT API - and in so doing, serializes all access to coreMQTT even when multiple tasks are using the same MQTT connection.

The agent provides an equivalent API for each coreMQTT API. Whereas coreMQTT APIs are prefixed "MQTT_", the agent APIs are prefixed "MQTTAgent_". For example, that agent's MQTTAgent_Publish() API is the thread safe equivalent to coreMQTT's MQTT_Publish() API.

Function Documentation

◆ addAwaitingOperation()

static MQTTStatus_t addAwaitingOperation	(	MQTTAgentContext_t *	pAgentContext,
		uint16_t	packetId,
		MQTTAgentCommand_t *	pCommand
	)

static

Track an operation by adding it to a list, indicating it is anticipating an acknowledgment.

Parameters

[in]	pAgentContext	Agent context for the MQTT connection.
[in]	packetId	Packet ID of pending ack.
[in]	pCommand	Pointer to command that is expecting an ack.

Returns

Returns one of the following:

MQTTSuccess if an entry was added for the to the list.
MQTTStateCollision if there already exists an entry for the same packet ID in the list.
MQTTNoMemory if there is no space available in the list for adding a new entry.

◆ getAwaitingOperation()

static MQTTAgentAckInfo_t * getAwaitingOperation	(	MQTTAgentContext_t *	pAgentContext,
		uint16_t	incomingPacketId
	)

static

Retrieve an operation from the list of pending acks, and optionally remove it from the list.

Parameters

[in]	pAgentContext	Agent context for the MQTT connection.
[in]	incomingPacketId	Packet ID of incoming ack.

Returns: Pointer to stored information about the operation awaiting the ack. Returns NULL if the packet ID is zero or original command does not exist.

◆ createCommand()

static MQTTStatus_t createCommand	(	MQTTAgentCommandType_t	commandType,
		const MQTTAgentContext_t *	pMqttAgentContext,
		void *	pMqttInfoParam,
		MQTTAgentCommandCallback_t	commandCompleteCallback,
		MQTTAgentCommandContext_t *	pCommandCompleteCallbackContext,
		MQTTAgentCommand_t *	pCommand
	)

static

Populate the parameters of a MQTTAgentCommand struct.

Parameters

[in]	commandType	Type of command. For example, publish or subscribe.
[in]	pMqttAgentContext	Pointer to MQTT context to use for command.
[in]	pMqttInfoParam	Pointer to MQTTPublishInfo_t or MQTTSubscribeInfo_t.
[in]	commandCompleteCallback	Callback for when command completes.
[in]	pCommandCompleteCallbackContext	Context and necessary structs for command.
[out]	pCommand	Pointer to initialized command.

Returns: MQTTSuccess if all necessary fields for the command are passed, else an enumerated error code.

◆ addCommandToQueue()

static MQTTStatus_t addCommandToQueue	(	const MQTTAgentContext_t *	pAgentContext,
		MQTTAgentCommand_t *	pCommand,
		uint32_t	blockTimeMs
	)

static

Add a command to the global command queue.

Parameters

[in]	pAgentContext	Agent context for the MQTT connection.
[in]	pCommand	Pointer to command to copy to queue.
[in]	blockTimeMs	The maximum amount of time to milliseconds to wait in the Blocked state (so not consuming any CPU time) for the command to be posted to the queue should the queue already be full.

Returns: MQTTSuccess if the command was added to the queue, else an enumerated error code.

◆ processCommand()

static MQTTStatus_t processCommand	(	MQTTAgentContext_t *	pMqttAgentContext,
		MQTTAgentCommand_t *	pCommand,
		bool *	pEndLoop
	)

static

Process a MQTTAgentCommand struct.

Note: This agent does not check existing subscriptions before sending a SUBSCRIBE or UNSUBSCRIBE packet. If a subscription already exists, then a SUBSCRIBE packet will be sent anyway, and if multiple tasks are subscribed to a topic filter, then they will all be unsubscribed after an UNSUBSCRIBE.

Parameters

[in]	pMqttAgentContext	Agent context for MQTT connection.
[in]	pCommand	Pointer to command to process.
[out]	pEndLoop	Whether the command loop should terminate.

Returns: status of MQTT library API call.

◆ mqttEventCallback()

static void mqttEventCallback	(	MQTTContext_t *	pMqttContext,
		MQTTPacketInfo_t *	pPacketInfo,
		MQTTDeserializedInfo_t *	pDeserializedInfo
	)

static

Dispatch incoming publishes and acks to their various handler functions.

Parameters

[in]	pMqttContext	MQTT Context
[in]	pPacketInfo	Pointer to incoming packet.
[in]	pDeserializedInfo	Pointer to deserialized information from the incoming packet.

◆ handleAcks()

static void handleAcks	(	const MQTTAgentContext_t *	pAgentContext,
		const MQTTPacketInfo_t *	pPacketInfo,
		const MQTTDeserializedInfo_t *	pDeserializedInfo,
		MQTTAgentAckInfo_t *	pAckInfo,
		uint8_t	packetType
	)

static

Mark a command as complete after receiving an acknowledgment packet.

Parameters

[in]	pAgentContext	Agent context for the MQTT connection.
[in]	pPacketInfo	Pointer to incoming packet.
[in]	pDeserializedInfo	Pointer to deserialized information from the incoming packet.
[in]	pAckInfo	Pointer to stored information for the original operation resulting in the received packet.
[in]	packetType	The type of the incoming packet, either SUBACK, UNSUBACK, PUBACK, or PUBCOMP.

◆ getAgentFromMQTTContext()

static MQTTAgentContext_t * getAgentFromMQTTContext ( MQTTContext_t * pMQTTContext )

static

Retrieve a pointer to an agent context given an MQTT context.

Parameters

[in] pMQTTContext MQTT Context to search for.

Returns: Pointer to agent context, or NULL.

◆ createAndAddCommand()

static MQTTStatus_t createAndAddCommand	(	MQTTAgentCommandType_t	commandType,
		const MQTTAgentContext_t *	pMqttAgentContext,
		void *	pMqttInfoParam,
		MQTTAgentCommandCallback_t	commandCompleteCallback,
		MQTTAgentCommandContext_t *	pCommandCompleteCallbackContext,
		uint32_t	blockTimeMs
	)

static

Helper function for creating a command and adding it to the command queue.

Parameters

[in]	commandType	Type of command.
[in]	pMqttAgentContext	Handle of the MQTT connection to use.
[in]	pCommandCompleteCallbackContext	Context and necessary structs for command.
[in]	commandCompleteCallback	Callback for when command completes.
[in]	pMqttInfoParam	Pointer to command argument.
[in]	blockTimeMs	Maximum amount of time in milliseconds to wait (in the Blocked state, so not consuming any CPU time) for the command to be posted to the MQTT agent should the MQTT agent's event queue be full.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

◆ concludeCommand()

static void concludeCommand	(	const MQTTAgentContext_t *	pAgentContext,
		MQTTAgentCommand_t *	pCommand,
		MQTTStatus_t	returnCode,
		uint8_t *	pSubackCodes
	)

static

Helper function to mark a command as complete and invoke its callback. This function calls the releaseCommand callback.

Parameters

[in]	pAgentContext	Agent context for the MQTT connection.
[in]	pCommand	Command to complete.
[in]	returnCode	Return status of command.
[in]	pSubackCodes	Pointer to suback array, if command is a SUBSCRIBE.

◆ resendPublishes()

static MQTTStatus_t resendPublishes ( MQTTAgentContext_t * pMqttAgentContext )

static

Resend QoS 1 and 2 publishes after resuming a session.

Parameters

[in] pMqttAgentContext Agent context for the MQTT connection.

Returns: MQTTSuccess if all publishes resent successfully, else error code from MQTT_Publish.

◆ clearPendingAcknowledgments()

static void clearPendingAcknowledgments	(	MQTTAgentContext_t *	pMqttAgentContext,
		bool	clearOnlySubUnsubEntries
	)

static

Clears the list of pending acknowledgments by invoking each callback with MQTTRecvFailed either for ALL operations in the list OR only for Subscribe/Unsubscribe operations.

Parameters

[in]	pMqttAgentContext	Agent context of the MQTT connection.
[in]	clearOnlySubUnsubEntries	Flag indicating whether all entries OR entries pertaining to only Subscribe/Unsubscribe operations should be cleaned from the list.

◆ validateStruct()

static bool validateStruct	(	const MQTTAgentContext_t *	pMqttAgentContext,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

static

Validate an MQTTAgentContext_t and a MQTTAgentCommandInfo_t from API functions.

Parameters

[in]	pMqttAgentContext	MQTTAgentContext_t to validate.
[in]	pCommandInfo	MQTTAgentCommandInfo_t to validate.

Returns: true if parameters are valid, else false.

◆ validateParams()

static bool validateParams	(	MQTTAgentCommandType_t	commandType,
		const void *	pParams
	)

static

Validate the parameters for a CONNECT, SUBSCRIBE, UNSUBSCRIBE or PUBLISH.

Parameters

[in]	commandType	CONNECT, SUBSCRIBE, UNSUBSCRIBE, or PUBLISH.
[in]	pParams	Parameter structure to validate.

Returns: true if parameter structure is valid, else false.

◆ isSpaceInPendingAckList()

static bool isSpaceInPendingAckList ( const MQTTAgentContext_t * pAgentContext )

static

Called before accepting any PUBLISH or SUBSCRIBE messages to check there is space in the pending ACK list for the outgoing PUBLISH or SUBSCRIBE.

Note: Because the MQTT agent is inherently multi threaded, and this function is called from the context of the application task and not the MQTT agent task, this function can only return a best effort result. It can definitely say if there is space for a new pending ACK when the function is called, but the case of space being exhausted when the agent executes a command that results in an ACK must still be handled.

Parameters

[in] pAgentContext Pointer to the context for the MQTT connection to which the PUBLISH or SUBSCRIBE message is to be sent.

Returns: true if there is space in that MQTT connection's ACK list, otherwise false;

◆ MQTTAgent_Init()

MQTTStatus_t MQTTAgent_Init	(	MQTTAgentContext_t *	pMqttAgentContext,
		const MQTTAgentMessageInterface_t *	pMsgInterface,
		const MQTTFixedBuffer_t *	pNetworkBuffer,
		const TransportInterface_t *	pTransportInterface,
		MQTTGetCurrentTimeFunc_t	getCurrentTimeMs,
		MQTTAgentIncomingPublishCallback_t	incomingCallback,
		void *	pIncomingPacketContext
	)

Perform any initialization the MQTT agent requires before it can be used. Must be called before any other function.

Parameters

[in]	pMqttAgentContext	Pointer to struct to initialize.
[in]	pMsgInterface	Command interface to use for allocating and sending commands.
[in]	pNetworkBuffer	Pointer to network buffer to use.
[in]	pTransportInterface	Transport interface to use with the MQTT library. See https://www.freertos.org/network-interface.html
[in]	getCurrentTimeMs	Pointer to a function that returns a count value that increments every millisecond.
[in]	incomingCallback	The callback to execute when receiving publishes.
[in]	pIncomingPacketContext	A pointer to a context structure defined by the application writer.

Note: The pIncomingPacketContext context provided for the incoming publish callback MUST remain in scope throughout the period that the agent task is running.

Returns: Appropriate status code from MQTT_Init().

Example

// Function for obtaining a timestamp.
uint32_t getTimeStampMs();
// Callback function for receiving packets.
void incomingPublishCallback( MQTTAgentContext_t * pMqttAgentContext,
                              uint16_t packetId,
                              MQTTPublishInfo_t * pPublishInfo );
// Platform function for network send interface.
int32_t networkSend( NetworkContext_t * pContext, const void * pBuffer, size_t bytes );
// Platform for network receive interface.
int32_t networkRecv( NetworkContext_t * pContext, void * pBuffer, size_t bytes );
 
// Platform function for Agent Message Send.
bool agentSendMessage( MQTTAgentMessageContext_t * pMsgCtx,
                       MQTTAgentCommand_t * const * pCommandToSend,
                       uint32_t blockTimeMs );
// Platform function for Agent Message Receive.
bool agentReceiveMessage( MQTTAgentMessageContext_t * pMsgCtx,
                          MQTTAgentCommand_t ** pCommandToSend,
                          uint32_t blockTimeMs );
// Platform function to Get Agent Command.
MQTTAgentCommand_t * getCommand( uint32_t blockTimeMs );
// Platform function to Release Agent Command.
bool releaseCommand( MQTTAgentCommand_t * pCommandToRelease );
 
// Variables used in this example.
MQTTAgentMessageInterface_t messageInterface;
MQTTAgentMessageContext_t messageContext;
MQTTAgentContext_t agentContext;
TransportInterface_t transport;
// Buffer for storing outgoing and incoming MQTT packets.
MQTTFixedBuffer_t fixedBuffer;
uint8_t buffer[ 1024 ];
MQTTStatus_t status;
 
// Set transport interface members.
transport.pNetworkContext = &someTransportContext;
transport.send = networkSend;
transport.recv = networkRecv;
 
// Set agent message interface members.
messageInterface.pMsgCtx = &messageContext;
messageInterface.send = agentSendMessage;
messageInterface.recv = agentReceiveMessage;
messageInterface.getCommand = getCommand;
messageInterface.releaseCommand = releaseCommand;
 
// Set buffer members.
fixedBuffer.pBuffer = buffer;
fixedBuffer.size = 1024;
 
status = MQTTAgent_Init( &agentContext,
                         &messageInterface,
                         &networkBuffer,
                         &transportInterface,
                         stubGetTime,
                         stubPublishCallback,
                         incomingPacketContext );
 
if( status == MQTTSuccess )
{
   // Do something with agentContext. The transport and message interfaces, and
   // fixedBuffer structs were copied into the context, so the original structs
   // do not need to stay in scope.
}

◆ MQTTAgent_CommandLoop()

MQTTStatus_t MQTTAgent_CommandLoop ( MQTTAgentContext_t * pMqttAgentContext )

Process commands from the command queue in a loop.

Parameters

[in] pMqttAgentContext The MQTT agent to use.

Returns: appropriate error code, or MQTTSuccess from a successful disconnect or termination.

Example

// Variables used in this example.
MQTTStatus_t status;
MQTTAgentContext_t mqttAgentContext;
 
status = MQTTAgent_CommandLoop( &mqttAgentContext );
 
// The function returns on either receiving a terminate command,
// undergoing network disconnection OR encountering an error.
if( ( status == MQTTSuccess ) && ( mqttAgentContext.mqttContext.connectStatus == MQTTNotConnected ) )
{
   // A terminate command was processed and MQTT connection was closed.
   // Need to close socket connection.
   Platform_DisconnectNetwork( mqttAgentContext.mqttContext.transportInterface.pNetworkContext );
}
else if( status == MQTTSuccess )
{
   // Terminate command was processed but MQTT connection was not
   // closed. Thus, need to close both MQTT and socket connections.
   status = MQTT_Disconnect( &( mqttAgentContext.mqttContext ) );
   assert( status == MQTTSuccess );
   Platform_DisconnectNetwork( mqttAgentContext.mqttContext.transportInterface.pNetworkContext );
}
else
{
    // Handle error.
}

◆ MQTTAgent_ResumeSession()

MQTTStatus_t MQTTAgent_ResumeSession	(	MQTTAgentContext_t *	pMqttAgentContext,
		bool	sessionPresent
	)

Resume a session by resending publishes if a session is present in the broker, or clear state information if not.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	sessionPresent	The session present flag from the broker.

Note: This function is NOT thread-safe and should only be called from the context of the task responsible for MQTTAgent_CommandLoop.

Returns: MQTTSuccess if it succeeds in resending publishes, else an appropriate error code from MQTT_Publish()

Example

// Variables used in this example.
MQTTStatus_t status;
MQTTAgentContext_t mqttAgentContext;
MQTTConnectInfo_t connectInfo = { 0 };
MQTTPublishInfo_t willInfo = { 0 };
bool sessionPresent;
 
// The example assumes that all variables have been filled with
// data for the MQTT_Connect call
// Refer to the MQTT_Connect API for a more detailed example.
 
// Attempt to resume session with the broker.
status = MQTT_Connect( &( mqttAgentContext.mqttContext ), &connectInfo, &willInfo, 100, &sessionPresent )
 
if( status == MQTTSuccess )
{
    // Process the session present status sent by the broker.
    status = MQTTAgent_ResumeSession( &mqttAgentContext, sessionPresent );
}

◆ MQTTAgent_CancelAll()

MQTTStatus_t MQTTAgent_CancelAll ( MQTTAgentContext_t * pMqttAgentContext )

Cancel all enqueued commands and those awaiting acknowledgment while the command loop is not running.

Canceled commands will be terminated with return code MQTTRecvFailed.

Parameters

[in] pMqttAgentContext The MQTT agent to use.

Note: This function is NOT thread-safe and should only be called from the context of the task responsible for MQTTAgent_CommandLoop.

Returns: MQTTBadParameter if an invalid context is given, else MQTTSuccess.

Example

// Variables used in this example.
MQTTStatus_t status;
MQTTAgentContext_t mqttAgentContext;
 
status = MQTTAgent_CommandLoop( &mqttAgentContext );
 
//An error was returned, but reconnection is not desired. Cancel all commands
//that are in the queue or awaiting an acknowledgment.
if( status != MQTTSuccess )
{
   //Cancel commands so any completion callbacks will be invoked.
   status = MQTTAgent_CancelAll( &mqttAgentContext );
}
 
Platform_DisconnectNetwork( mqttAgentContext.mqttContext.transportInterface.pNetworkContext );

◆ MQTTAgent_Subscribe()

MQTTStatus_t MQTTAgent_Subscribe	(	const MQTTAgentContext_t *	pMqttAgentContext,
		MQTTAgentSubscribeArgs_t *	pSubscriptionArgs,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a command to call MQTT_Subscribe() for an MQTT connection.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pSubscriptionArgs	Struct describing topic to subscribe to.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
MQTTSubscribeInfo_t subscribeInfo = { 0 };
MQTTAgentSubscribeArgs_t subscribeArgs = { 0 };
 
// Function for command complete callback.
void subscribeCmdCompleteCb( MQTTAgentCommandContext_t * pCmdCallbackContext,
                             MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.CmdCompleteCallback = subscribeCmdCompleteCb;
commandInfo.blockTimeMs = 500;
 
// Fill the information for topic filters to subscribe to.
subscribeInfo.qos = Qos1;
subscribeInfo.pTopicFilter = "/foo/bar";
subscribeInfo.topicFilterLength = strlen("/foo/bar");
subscribeArgs.pSubscribeInfo = &subscribeInfo;
subscribeArgs.numSubscriptions = 1U;
 
status = MQTTAgent_Subscribe( &agentContext, &subscribeArgs, &commandInfo );
 
if( status == MQTTSuccess )
{
  // Command to send subscribe request to the server has been queued. Notification
  // about completion of the subscribe operation will be notified to application
  // through invocation of subscribeCmdCompleteCb().
}

◆ MQTTAgent_Unsubscribe()

MQTTStatus_t MQTTAgent_Unsubscribe	(	const MQTTAgentContext_t *	pMqttAgentContext,
		MQTTAgentSubscribeArgs_t *	pSubscriptionArgs,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a command to call MQTT_Unsubscribe() for an MQTT connection.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pSubscriptionArgs	List of topics to unsubscribe from.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
MQTTSubscribeInfo_t unsubscribeInfo = { 0 };
MQTTAgentSubscribeArgs_t unsubscribeArgs = { 0 };
 
// Function for command complete callback.
void unsubscribeCmdCompleteCb( MQTTAgentCommandContext_t * pCmdCallbackContext,
                               MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = unsubscribeCmdCompleteCb;
commandInfo.blockTimeMs = 500;
 
// Fill the information for topics to unsubscribe from.
unsubscribeInfo.pTopicFilter = "/foo/bar";
unsubscribeInfo.topicFilterLength = strlen("/foo/bar");
unsubscribeArgs.pSubscribeInfo = &unsubscribeInfo;
unsubscribeArgs.numSubscriptions = 1U;
 
status = MQTTAgent_Unsubscribe( &agentContext, &unsubscribeArgs, &commandInfo );
 
if( status == MQTTSuccess )
{
  // Command to send Unsubscribe request to the server has been queued. Notification
  // about completion of the Unsubscribe operation will be notified to application
  // through invocation of unsubscribeCompleteCb().
}

◆ MQTTAgent_Publish()

MQTTStatus_t MQTTAgent_Publish	(	const MQTTAgentContext_t *	pMqttAgentContext,
		MQTTPublishInfo_t *	pPublishInfo,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a command to call MQTT_Publish() for an MQTT connection.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pPublishInfo	MQTT PUBLISH information.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
MQTTPublishInfo_t publishInfo = { 0 };
 
// Function for command complete callback.
void publishCmdCompleteCb( MQTTAgentCommandContext_t * pCmdCallbackContext,
                           MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = publishCmdCompleteCb;
commandInfo.blockTimeMs = 500;
 
// Fill the information for publish operation.
publishInfo.qos = MQTTQoS1;
publishInfo.pTopicName = "/some/topic/name";
publishInfo.topicNameLength = strlen( publishInfo.pTopicName );
publishInfo.pPayload = "Hello World!";
publishInfo.payloadLength = strlen( "Hello World!" );
 
status = MQTTAgent_Publish( &agentContext, &publishInfo, &commandInfo );
 
if( status == MQTTSuccess )
{
   // Command to publish message to broker has been queued.
   // The event of publish operation completion will be notified with
   // the invocation of the publishCmdCompleteCb().
}

◆ MQTTAgent_ProcessLoop()

MQTTStatus_t MQTTAgent_ProcessLoop	(	const MQTTAgentContext_t *	pMqttAgentContext,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Send a message to the MQTT agent purely to trigger an iteration of its loop, which will result in a call to MQTT_ProcessLoop(). This function can be used to wake the MQTT agent task when it is known data may be available on the connected socket.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
 
// Function for command complete callback.
void cmdCompleteCb( MQTTAgentCommandContext_t * pCmdCallbackContext,
                    MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = cmdCompleteCb;
commandInfo.blockTimeMs = 500;
 
status = MQTTAgent_ProcessLoop( &agentContext, &commandInfo );
 
if( status == MQTTSuccess )
{
   // Command to call MQTT_ProcessLoop() has been queued.
   // After processing the command, if an incoming publish is received,
   // the event will be notified with invocation of the incoming publish
   // callback configured in the agent context.
}

◆ MQTTAgent_Connect()

MQTTStatus_t MQTTAgent_Connect	(	const MQTTAgentContext_t *	pMqttAgentContext,
		MQTTAgentConnectArgs_t *	pConnectArgs,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a command to call MQTT_Connect() for an MQTT connection. If a session is resumed with the broker, it will also resend the necessary QoS1/2 publishes.

Note: The MQTTAgent_Connect function is provided to give a thread safe equivalent to the MQTT_Connect API. However, it is RECOMMENDED that instead of the application tasks (i.e. tasks other than the agent task), the agent be responsible for creating the MQTT connection (by calling MQTT_Connect) before starting the command loop (with the MQTTAgent_CommandLoop() call). In that case, the agent SHOULD also be responsible for disconnecting the MQTT connection after the command loop has terminated (through an MQTTAgent_Terminate() call from an application task).

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in,out]	pConnectArgs	Struct holding args for MQTT_Connect(). On a successful connection after the command is processed, the sessionPresent member will be filled to indicate whether the broker resumed an existing session.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTConnectInfo_t connectInfo = { 0 };
MQTTPublishInfo_t willInfo = { 0 };
MQTTAgentConnectArgs_t connectionArgs;
MQTTAgentCommandInfo_t commandInfo = { 0 };
 
// True for creating a new session with broker, false if we want to resume an old one.
connectInfo.cleanSession = true;
// Client ID must be unique to broker. This field is required.
connectInfo.pClientIdentifier = "someClientID";
connectInfo.clientIdentifierLength = strlen( connectInfo.pClientIdentifier );
 
// Value for keep alive.
connectInfo.keepAliveSeconds = 60;
// The following fields are optional.
// Optional username and password.
connectInfo.pUserName = "someUserName";
connectInfo.userNameLength = strlen( connectInfo.pUserName );
connectInfo.pPassword = "somePassword";
connectInfo.passwordLength = strlen( connectInfo.pPassword );
 
// The last will and testament is optional, it will be published by the broker
// should this client disconnect without sending a DISCONNECT packet.
willInfo.qos = MQTTQoS0;
willInfo.pTopicName = "/lwt/topic/name";
willInfo.topicNameLength = strlen( willInfo.pTopicName );
willInfo.pPayload = "LWT Message";
willInfo.payloadLength = strlen( "LWT Message" );
 
// Fill the MQTTAgentConnectArgs_t structure.
connectArgs.pConnectInfo = &connectInfo;
connectArgs.pWillInfo = &willInfo;
// Time to block for CONNACK response when command is processed.
connectArgs.timeoutMs = 500;
 
// Function for command complete callback.
void connectCmdCallback( MQTTAgentCommandContext_t * pCmdCallbackContext,
                         MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = connectCmdCallback;
commandInfo.blockTimeMs = 500;
 
status = MQTTAgent_Connect( &agentContext, &connectArgs, &commandInfo );
 
if( status == MQTTSuccess )
{
  // Command for creating the MQTT connection has been queued.
  // The MQTT connection event will be notified through the
  // invocation of connectCmdCallback().
}

◆ MQTTAgent_Disconnect()

MQTTStatus_t MQTTAgent_Disconnect	(	const MQTTAgentContext_t *	pMqttAgentContext,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a command to disconnect an MQTT connection.

Note: MQTTAgent_CommandLoop will return after processing a DISCONNECT command to allow the network connection to be disconnected. However, any pending commands in the queue, as well as those waiting for an acknowledgment, will NOT be terminated.; The MQTTAgent_Disconnect function is provided to give a thread safe equivalent to the MQTT_Disconnect API. However, if the agent task is responsible for creating the MQTT connection (before calling MQTTAgent_CommandLoop()), then it is RECOMMENDED that an application task (i.e. a task other than the agent task) use MQTTAgent_Terminate to terminate the command loop in the agent, and the agent task be responsible for disconnecting the MQTT connection.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
 
// Function for command complete callback.
void disconnectCmdCallback( MQTTAgentCommandContext_t * pCmdCallbackContext,
                            MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = disconnectCmdCallback;
commandInfo.blockTimeMs = 500;
 
status = MQTTAgent_Disconnect( &agentContext, &commandInfo );
 
if( status == MQTTSuccess )
{
  // Command for closing the MQTT connection has been queued.
  // The MQTT disconnection event will be notified through the
  // invocation of disconnectCmdCallback().
}

◆ MQTTAgent_Ping()

MQTTStatus_t MQTTAgent_Ping	(	const MQTTAgentContext_t *	pMqttAgentContext,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a command to call MQTT_Ping() for an MQTT connection.

Note: This API function ONLY enqueues a command to send a ping request to the server, and DOES NOT wait for a ping response to be received from the server. To detect whether a Ping Response, has not been received from the server, the MQTTAgent_CommandLoop function SHOULD be used, which returns the MQTTKeepAliveTimeout return code on a ping response (or keep-alive) timeout.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
 
// Function for command complete callback.
void pingRequestCompleteCb( MQTTAgentCommandContext_t * pCmdCallbackContext,
                            MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = pingRequestCompleteCb;
commandInfo.blockTimeMs = 500;
 
status = MQTTAgent_Ping( &agentContext, &commandInfo );
 
if( status == MQTTSuccess )
{
  // Command for sending request has been queued. Application can
  // handle keep-alive timeout if detected through return value of
  // MQTTAgent_CommandLoop in the task running the agent.
}

◆ MQTTAgent_Terminate()

MQTTStatus_t MQTTAgent_Terminate	(	const MQTTAgentContext_t *	pMqttAgentContext,
		const MQTTAgentCommandInfo_t *	pCommandInfo
	)

Add a termination command to the command queue.

On command loop termination, all pending commands in the queue, as well as those waiting for an acknowledgment, will be terminated with error code MQTTRecvFailed.

Note: Commands may still be posted to the command queue after MQTTAgent_CommandLoop has returned. It is the responsibility of the application to cancel any commands that are posted while the command loop is not running, such as by invoking MQTTAgent_CancelAll.; We RECOMMEND that this function is used from application task(s), that is a task not running the agent, to terminate the agent loop instead of calling MQTTAgent_Disconnect, so that the logic for creating and closing MQTT connection is owned by the agent task.

Parameters

[in]	pMqttAgentContext	The MQTT agent to use.
[in]	pCommandInfo	The information pertaining to the command, including: cmdCompleteCallback Optional callback to invoke when the command completes. pCmdCompleteCallbackContext Optional completion callback context. blockTimeMs The maximum amount of time in milliseconds to wait for the command to be posted to the MQTT agent, should the agent's event queue be full. Tasks wait in the Blocked state so don't use any CPU time.

Note: The context passed to the callback through pCmdContext member of pCommandInfo parameter MUST remain in scope at least until the callback has been executed by the agent task.

Returns: MQTTSuccess if the command was posted to the MQTT agent's event queue. Otherwise an enumerated error code.

Example

// Variables used in this example.
MQTTAgentContext_t agentContext;
MQTTStatus_t status;
MQTTAgentCommandInfo_t commandInfo = { 0 };
 
// Function for command complete callback.
void terminateCallback( MQTTAgentCommandContext_t * pCmdCallbackContext,
                        MQTTAgentReturnInfo_t * pReturnInfo );
 
// Fill the command information.
commandInfo.cmdCompleteCallback = terminateCallback;
commandInfo.blockTimeMs = 500;
 
status = MQTTAgent_Terminate( &agentContext, &commandInfo );
 
if( status == MQTTSuccess )
{
  // Command to terminate the agent loop has been queued.
}

Functions

Detailed Description

Function Documentation

◆ addAwaitingOperation()

◆ getAwaitingOperation()

◆ createCommand()

◆ addCommandToQueue()

◆ processCommand()

◆ mqttEventCallback()

◆ handleAcks()

◆ getAgentFromMQTTContext()

◆ createAndAddCommand()

◆ concludeCommand()

◆ resendPublishes()

◆ clearPendingAcknowledgments()

◆ validateStruct()

◆ validateParams()

◆ isSpaceInPendingAckList()

◆ MQTTAgent_Init()

◆ MQTTAgent_CommandLoop()

◆ MQTTAgent_ResumeSession()

◆ MQTTAgent_CancelAll()

◆ MQTTAgent_Subscribe()

◆ MQTTAgent_Unsubscribe()

◆ MQTTAgent_Publish()

◆ MQTTAgent_ProcessLoop()

◆ MQTTAgent_Connect()

◆ MQTTAgent_Disconnect()

◆ MQTTAgent_Ping()

◆ MQTTAgent_Terminate()