core_mqtt_state.c File Reference

Implements the functions in core_mqtt_state.h. More...

#include <assert.h>
#include <string.h>
#include "core_mqtt_state.h"
#include "core_mqtt_config_defaults.h"

Macros
#define	MQTT_INVALID_STATE_COUNT   ( ~ZERO_SIZE_T )
	This macro depicts the invalid value for the state publishes.
#define	UINT16_BITMAP_BIT_SET_AT(position)   ( ( uint16_t ) 0x01U << ( ( uint16_t ) position ) )
	Create a 16-bit bitmap with bit set at specified position.
#define	UINT16_SET_BIT(x, position)   ( ( x ) = ( uint16_t ) ( ( x ) | ( UINT16_BITMAP_BIT_SET_AT( position ) ) ) )
	Set a bit in an 16-bit unsigned integer.
#define	UINT16_CHECK_BIT(x, position)   ( ( ( x ) & ( UINT16_BITMAP_BIT_SET_AT( position ) ) ) == ( UINT16_BITMAP_BIT_SET_AT( position ) ) )
	Macro for checking if a bit is set in a 16-bit unsigned integer.

Functions
static bool	validateTransitionPublish (MQTTPublishState_t currentState, MQTTPublishState_t newState, MQTTStateOperation_t opType, MQTTQoS_t qos)
	Test if a transition to new state is possible, when dealing with PUBLISHes.
static bool	validateTransitionAck (MQTTPublishState_t currentState, MQTTPublishState_t newState)
	Test if a transition to a new state is possible, when dealing with acks.
static bool	isPublishOutgoing (MQTTPubAckType_t packetType, MQTTStateOperation_t opType)
	Test if the publish corresponding to an ack is outgoing or incoming.
static size_t	findInRecord (const MQTTPubAckInfo_t *records, size_t recordCount, uint16_t packetId, MQTTQoS_t *pQos, MQTTPublishState_t *pCurrentState)
	Find a packet ID in the state record.
static void	compactRecords (MQTTPubAckInfo_t *records, size_t recordCount)
	Compact records.
static MQTTStatus_t	addRecord (MQTTPubAckInfo_t *records, size_t recordCount, uint16_t packetId, MQTTQoS_t qos, MQTTPublishState_t publishState)
	Store a new entry in the state record.
static void	updateRecord (MQTTPubAckInfo_t *records, size_t recordIndex, MQTTPublishState_t newState, bool shouldDelete)
	Update and possibly delete an entry in the state record.
static uint16_t	stateSelect (const MQTTContext_t *pMqttContext, uint16_t searchStates, MQTTStateCursor_t *pCursor)
	Get the packet ID and index of an outgoing publish in specified states.
static MQTTStatus_t	updateStateAck (MQTTPubAckInfo_t *records, size_t maxRecordCount, size_t recordIndex, uint16_t packetId, MQTTPublishState_t currentState, MQTTPublishState_t newState)
	Update the state records for an ACK after state transition validations.
static MQTTStatus_t	updateStatePublish (const MQTTContext_t *pMqttContext, size_t recordIndex, uint16_t packetId, MQTTStateOperation_t opType, MQTTQoS_t qos, MQTTPublishState_t currentState, MQTTPublishState_t newState)
	Update the state record for a PUBLISH packet after validating the state transitions.
MQTTPublishState_t	MQTT_CalculateStateAck (MQTTPubAckType_t packetType, MQTTStateOperation_t opType, MQTTQoS_t qos)
	Calculate the state from a PUBACK, PUBREC, PUBREL, or PUBCOMP.
MQTTStatus_t	MQTT_ReserveState (const MQTTContext_t *pMqttContext, uint16_t packetId, MQTTQoS_t qos)
	Reserve an entry for an outgoing QoS 1 or Qos 2 publish.
MQTTPublishState_t	MQTT_CalculateStatePublish (MQTTStateOperation_t opType, MQTTQoS_t qos)
	Calculate the new state for a publish from its qos and operation type.
MQTTStatus_t	MQTT_UpdateStatePublish (const MQTTContext_t *pMqttContext, uint16_t packetId, MQTTStateOperation_t opType, MQTTQoS_t qos, MQTTPublishState_t *pNewState)
	Update the state record for a PUBLISH packet.
MQTTStatus_t	MQTT_RemoveStateRecord (const MQTTContext_t *pMqttContext, uint16_t packetId)
	Remove the state record for a PUBLISH packet.
MQTTStatus_t	MQTT_UpdateStateAck (const MQTTContext_t *pMqttContext, uint16_t packetId, MQTTPubAckType_t packetType, MQTTStateOperation_t opType, MQTTPublishState_t *pNewState)
	Update the state record for an ACKed publish.
uint16_t	MQTT_PubrelToResend (const MQTTContext_t *pMqttContext, MQTTStateCursor_t *pCursor, MQTTPublishState_t *pState)
	Get the packet ID of next pending PUBREL ack to be resent.
uint16_t	MQTT_PublishToResend (const MQTTContext_t *pMqttContext, MQTTStateCursor_t *pCursor)
	Get the packet ID of next pending publish to be resent.
const char *	MQTT_State_strerror (MQTTPublishState_t state)
	State to string conversion for state engine.

Variables
static const size_t	ZERO_SIZE_T = 0U
	A global static variable used to generate the macro MQTT_INVALID_STATE_COUNT of size_t length.

Detailed Description

Implements the functions in core_mqtt_state.h.

Macro Definition Documentation

UINT16_BITMAP_BIT_SET_AT

#define UINT16_BITMAP_BIT_SET_AT

(

position

)

( ( uint16_t ) 0x01U << ( ( uint16_t ) position ) )

Create a 16-bit bitmap with bit set at specified position.

Parameters

[in]

position

The position at which the bit need to be set.

UINT16_SET_BIT

#define UINT16_SET_BIT	(		x,
			position
	)		( ( x ) = ( uint16_t ) ( ( x ) | ( UINT16_BITMAP_BIT_SET_AT( position ) ) ) )

Set a bit in an 16-bit unsigned integer.

Parameters

[in]	x	The 16-bit unsigned integer to set a bit.
[in]	position	The position at which the bit need to be set.

UINT16_CHECK_BIT

#define UINT16_CHECK_BIT	(		x,
			position
	)		( ( ( x ) & ( UINT16_BITMAP_BIT_SET_AT( position ) ) ) == ( UINT16_BITMAP_BIT_SET_AT( position ) ) )

Macro for checking if a bit is set in a 16-bit unsigned integer.

Parameters

[in]	x	The unsigned 16-bit integer to check.
[in]	position	Which bit to check.

Function Documentation

validateTransitionPublish()

static bool validateTransitionPublish ( MQTTPublishState_t  currentState, MQTTPublishState_t  newState, MQTTStateOperation_t  opType, MQTTQoS_t  qos  )

static

Test if a transition to new state is possible, when dealing with PUBLISHes.

Parameters

[in]	currentState	The current state.
[in]	newState	State to transition to.
[in]	opType	Reserve, Send, or Receive.
[in]	qos	0, 1, or 2.

Note

This function does not validate the current state, or the new state based on either the operation type or QoS. It assumes the new state is valid given the opType and QoS, which will be the case if calculated by MQTT_CalculateStatePublish().

Returns

true if transition is possible, else false

validateTransitionAck()

static bool validateTransitionAck ( MQTTPublishState_t  currentState, MQTTPublishState_t  newState  )

static

Test if a transition to a new state is possible, when dealing with acks.

Parameters

[in]	currentState	The current state.
[in]	newState	State to transition to.

Returns

true if transition is possible, else false.

isPublishOutgoing()

static bool isPublishOutgoing ( MQTTPubAckType_t  packetType, MQTTStateOperation_t  opType  )

static

Test if the publish corresponding to an ack is outgoing or incoming.

Parameters

[in]	packetType	PUBACK, PUBREC, PUBREL, or PUBCOMP.
[in]	opType	Send, or Receive.

Returns

true if corresponds to outgoing publish, else false.

findInRecord()

static size_t findInRecord ( const MQTTPubAckInfo_t *  records, size_t  recordCount, uint16_t  packetId, MQTTQoS_t *  pQos, MQTTPublishState_t *  pCurrentState  )

static

Find a packet ID in the state record.

Parameters

[in]	records	State record array.
[in]	recordCount	Length of record array.
[in]	packetId	packet ID to search for.
[out]	pQos	QoS retrieved from record.
[out]	pCurrentState	state retrieved from record.

Returns

index of the packet id in the record if it exists, else the record length.

compactRecords()

static void compactRecords ( MQTTPubAckInfo_t *  records, size_t  recordCount  )

static

Compact records.

Records are arranged in the relative order to maintain message ordering. This will lead to fragmentation and this function will help in defragmenting the records array.

Parameters

[in]	records	State record array.
[in]	recordCount	Length of record array.

addRecord()

static MQTTStatus_t addRecord ( MQTTPubAckInfo_t *  records, size_t  recordCount, uint16_t  packetId, MQTTQoS_t  qos, MQTTPublishState_t  publishState  )

static

Store a new entry in the state record.

Parameters

[in]	records	State record array.
[in]	recordCount	Length of record array.
[in]	packetId	Packet ID of new entry.
[in]	qos	QoS of new entry.
[in]	publishState	State of new entry.

Returns

MQTTSuccess, MQTTNoMemory, or MQTTStateCollision.

updateRecord()

static void updateRecord ( MQTTPubAckInfo_t *  records, size_t  recordIndex, MQTTPublishState_t  newState, bool  shouldDelete  )

static

Update and possibly delete an entry in the state record.

Parameters

[in]	records	State record array.
[in]	recordIndex	index of record to update.
[in]	newState	New state to update.
[in]	shouldDelete	Whether an existing entry should be deleted.

stateSelect()

static uint16_t stateSelect ( const MQTTContext_t *  pMqttContext, uint16_t  searchStates, MQTTStateCursor_t *  pCursor  )

static

Get the packet ID and index of an outgoing publish in specified states.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in]	searchStates	The states to search for in 2-byte bit map.
[in,out]	pCursor	Index at which to start searching.

Returns

Packet ID of the outgoing publish.

updateStateAck()

static MQTTStatus_t updateStateAck ( MQTTPubAckInfo_t *  records, size_t  maxRecordCount, size_t  recordIndex, uint16_t  packetId, MQTTPublishState_t  currentState, MQTTPublishState_t  newState  )

static

Update the state records for an ACK after state transition validations.

Parameters

[in]	records	State records pointer.
[in]	maxRecordCount	The maximum number of records.
[in]	recordIndex	Index at which the record is stored.
[in]	packetId	Packet id of the packet.
[in]	currentState	Current state of the publish record.
[in]	newState	New state of the publish.

Returns

MQTTIllegalState, or MQTTSuccess.

updateStatePublish()

static MQTTStatus_t updateStatePublish ( const MQTTContext_t *  pMqttContext, size_t  recordIndex, uint16_t  packetId, MQTTStateOperation_t  opType, MQTTQoS_t  qos, MQTTPublishState_t  currentState, MQTTPublishState_t  newState  )

static

Update the state record for a PUBLISH packet after validating the state transitions.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in]	recordIndex	Index in state records at which publish record exists.
[in]	packetId	ID of the PUBLISH packet.
[in]	opType	Send or Receive.
[in]	qos	0, 1, or 2.
[in]	currentState	Current state of the publish record.
[in]	newState	New state of the publish record.

Returns

MQTTIllegalState, MQTTStateCollision or MQTTSuccess.

MQTT_CalculateStateAck()

MQTTPublishState_t MQTT_CalculateStateAck	(	MQTTPubAckType_t	packetType,
		MQTTStateOperation_t	opType,
		MQTTQoS_t	qos
	)

Calculate the state from a PUBACK, PUBREC, PUBREL, or PUBCOMP.

Parameters

[in]	packetType	PUBACK, PUBREC, PUBREL, or PUBCOMP.
[in]	opType	Send or Receive.
[in]	qos	1 or 2.

Returns

The calculated state.

MQTT_ReserveState()

MQTTStatus_t MQTT_ReserveState	(	const MQTTContext_t *	pMqttContext,
		uint16_t	packetId,
		MQTTQoS_t	qos
	)

Reserve an entry for an outgoing QoS 1 or Qos 2 publish.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in]	packetId	The ID of the publish packet.
[in]	qos	1 or 2.

Returns

MQTTSuccess, MQTTNoMemory, or MQTTStateCollision.

MQTT_CalculateStatePublish()

MQTTPublishState_t MQTT_CalculateStatePublish	(	MQTTStateOperation_t	opType,
		MQTTQoS_t	qos
	)

Calculate the new state for a publish from its qos and operation type.

Parameters

[in]	opType	Send or Receive.
[in]	qos	0, 1, or 2.

Returns

The calculated state.

MQTT_UpdateStatePublish()

MQTTStatus_t MQTT_UpdateStatePublish	(	const MQTTContext_t *	pMqttContext,
		uint16_t	packetId,
		MQTTStateOperation_t	opType,
		MQTTQoS_t	qos,
		MQTTPublishState_t *	pNewState
	)

Update the state record for a PUBLISH packet.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in]	packetId	ID of the PUBLISH packet.
[in]	opType	Send or Receive.
[in]	qos	0, 1, or 2.
[out]	pNewState	Updated state of the publish.

Returns

MQTTBadParameter, MQTTIllegalState, MQTTStateCollision or MQTTSuccess.

MQTT_RemoveStateRecord()

MQTTStatus_t MQTT_RemoveStateRecord	(	const MQTTContext_t *	pMqttContext,
		uint16_t	packetId
	)

Remove the state record for a PUBLISH packet.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in]	packetId	ID of the PUBLISH packet.

Returns

MQTTBadParameter or MQTTSuccess.

MQTT_UpdateStateAck()

MQTTStatus_t MQTT_UpdateStateAck	(	const MQTTContext_t *	pMqttContext,
		uint16_t	packetId,
		MQTTPubAckType_t	packetType,
		MQTTStateOperation_t	opType,
		MQTTPublishState_t *	pNewState
	)

Update the state record for an ACKed publish.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in]	packetId	ID of the ack packet.
[in]	packetType	PUBACK, PUBREC, PUBREL, or PUBCOMP.
[in]	opType	Send or Receive.
[out]	pNewState	Updated state of the publish.

Returns

MQTTBadParameter if an invalid parameter is passed; MQTTBadResponse if the packet from the network is not found in the records; MQTTIllegalState if the requested update would result in an illegal transition; MQTTSuccess otherwise.

MQTT_PubrelToResend()

uint16_t MQTT_PubrelToResend	(	const MQTTContext_t *	pMqttContext,
		MQTTStateCursor_t *	pCursor,
		MQTTPublishState_t *	pState
	)

Get the packet ID of next pending PUBREL ack to be resent.

This function will need to be called to get the packet for which a PUBREL need to be sent when a session is reestablished. Calling this function repeatedly until packet id is 0 will give all the packets for which a PUBREL need to be resent in the correct order.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in,out]	pCursor	Index at which to start searching.
[out]	pState	State indicating that PUBREL packet need to be sent.

MQTT_PublishToResend()

uint16_t MQTT_PublishToResend	(	const MQTTContext_t *	pMqttContext,
		MQTTStateCursor_t *	pCursor
	)

Get the packet ID of next pending publish to be resent.

This function will need to be called to get the packet for which a publish need to be sent when a session is reestablished. Calling this function repeatedly until packet id is 0 will give all the packets for which a publish need to be resent in the correct order.

Parameters

[in]	pMqttContext	Initialized MQTT context.
[in,out]	pCursor	Index at which to start searching.

Example

// For this example assume this function returns an outgoing unacknowledged
// QoS 1 or 2 publish from its packet identifier.
MQTTPublishInfo_t * getPublish( uint16_t packetID );
 
// Variables used in this example.
MQTTStatus_t status;
MQTTStateCursor_t cursor = MQTT_STATE_CURSOR_INITIALIZER;
bool sessionPresent;
uint16_t packetID;
MQTTPublishInfo_t * pResendPublish = NULL;
MQTTConnectInfo_t connectInfo = { 0 };
 
// This is assumed to have been initialized before the call to MQTT_Connect().
MQTTContext_t * pContext;
 
// Set clean session to false to attempt session resumption.
connectInfo.cleanSession = false;
connectInfo.pClientIdentifier = "someClientID";
connectInfo.clientIdentifierLength = strlen( connectInfo.pClientIdentifier );
connectInfo.keepAliveSeconds = 60;
// Optional connect parameters are not relevant to this example.
 
// Create an MQTT connection. Use 100 milliseconds as a timeout.
status = MQTT_Connect( pContext, &connectInfo, NULL, 100, &sessionPresent );
 
if( status == MQTTSuccess )
{
     if( sessionPresent )
     {
         // Loop while packet ID is nonzero.
         while( ( packetID = MQTT_PublishToResend( pContext, &cursor ) ) != 0 )
         {
             // Assume this function will succeed.
             pResendPublish = getPublish( packetID );
             // Set DUP flag.
             pResendPublish->dup = true;
             status = MQTT_Publish( pContext, pResendPublish, packetID );
 
             if( status != MQTTSuccess )
             {
                 // Application can decide how to handle a failure.
             }
         }
     }
     else
     {
         // The broker did not resume a session, so we can clean up the
         // list of outgoing publishes.
     }
}

MQTT_State_strerror()

const char * MQTT_State_strerror

(

MQTTPublishState_t

state

)

State to string conversion for state engine.

Parameters

[in]

state

The state to convert to a string.

Returns

The string representation of the state.