iot_mqtt_internal.h

Go to the documentation of this file.

/*
 * IoT MQTT V2.1.0
 * Copyright (C) 2018 Amazon.com, Inc. or its affiliates.  All Rights Reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
 * the Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#ifndef IOT_MQTT_INTERNAL_H_
#define IOT_MQTT_INTERNAL_H_

/* The config header is always included first. */
#include "iot_config.h"

/* Linear containers (lists and queues) include. */
#include "iot_linear_containers.h"

/* MQTT include. */
#include "iot_mqtt.h"

/* Task pool include. */
#include "iot_taskpool.h"

#if IOT_MQTT_ENABLE_ASSERTS == 1
    #ifndef IotMqtt_Assert
        #ifdef Iot_DefaultAssert
            #define IotMqtt_Assert( expression )    Iot_DefaultAssert( expression )
        #else
            #error "Asserts are enabled for MQTT, but IotMqtt_Assert is not defined"
        #endif
    #endif
#else
    #define IotMqtt_Assert( expression )
#endif

/* Configure logs for MQTT functions. */
#ifdef IOT_LOG_LEVEL_MQTT
    #define LIBRARY_LOG_LEVEL        IOT_LOG_LEVEL_MQTT
#else
    #ifdef IOT_LOG_LEVEL_GLOBAL
        #define LIBRARY_LOG_LEVEL    IOT_LOG_LEVEL_GLOBAL
    #else
        #define LIBRARY_LOG_LEVEL    IOT_LOG_NONE
    #endif
#endif

#define LIBRARY_LOG_NAME    ( "MQTT" )
#include "iot_logging_setup.h"

/*
 * Provide default values for undefined memory allocation functions based on
 * the usage of dynamic memory allocation.
 */
#if IOT_STATIC_MEMORY_ONLY == 1
    #include "iot_static_memory.h"

    void * IotMqtt_MallocConnection( size_t size );

    void IotMqtt_FreeConnection( void * ptr );

    #define IotMqtt_MallocMessage    Iot_MallocMessageBuffer

    #define IotMqtt_FreeMessage      Iot_FreeMessageBuffer

    void * IotMqtt_MallocOperation( size_t size );

    void IotMqtt_FreeOperation( void * ptr );

    void * IotMqtt_MallocSubscription( size_t size );

    void IotMqtt_FreeSubscription( void * ptr );
#else /* if IOT_STATIC_MEMORY_ONLY == 1 */
    #ifndef IotMqtt_MallocConnection
        #ifdef Iot_DefaultMalloc
            #define IotMqtt_MallocConnection    Iot_DefaultMalloc
        #else
            #error "No malloc function defined for IotMqtt_MallocConnection"
        #endif
    #endif

    #ifndef IotMqtt_FreeConnection
        #ifdef Iot_DefaultFree
            #define IotMqtt_FreeConnection    Iot_DefaultFree
        #else
            #error "No free function defined for IotMqtt_FreeConnection"
        #endif
    #endif

    #ifndef IotMqtt_MallocMessage
        #ifdef Iot_DefaultMalloc
            #define IotMqtt_MallocMessage    Iot_DefaultMalloc
        #else
            #error "No malloc function defined for IotMqtt_MallocMessage"
        #endif
    #endif

    #ifndef IotMqtt_FreeMessage
        #ifdef Iot_DefaultFree
            #define IotMqtt_FreeMessage    Iot_DefaultFree
        #else
            #error "No free function defined for IotMqtt_FreeMessage"
        #endif
    #endif

    #ifndef IotMqtt_MallocOperation
        #ifdef Iot_DefaultMalloc
            #define IotMqtt_MallocOperation    Iot_DefaultMalloc
        #else
            #error "No malloc function defined for IotMqtt_MallocOperation"
        #endif
    #endif

    #ifndef IotMqtt_FreeOperation
        #ifdef Iot_DefaultFree
            #define IotMqtt_FreeOperation    Iot_DefaultFree
        #else
            #error "No free function defined for IotMqtt_FreeOperation"
        #endif
    #endif

    #ifndef IotMqtt_MallocSubscription
        #ifdef Iot_DefaultMalloc
            #define IotMqtt_MallocSubscription    Iot_DefaultMalloc
        #else
            #error "No malloc function defined for IotMqtt_MallocSubscription"
        #endif
    #endif

    #ifndef IotMqtt_FreeSubscription
        #ifdef Iot_DefaultFree
            #define IotMqtt_FreeSubscription    Iot_DefaultFree
        #else
            #error "No free function defined for IotMqtt_FreeSubscription"
        #endif
    #endif
#endif /* if IOT_STATIC_MEMORY_ONLY == 1 */

#ifndef AWS_IOT_MQTT_ENABLE_METRICS
    #define AWS_IOT_MQTT_ENABLE_METRICS             ( 1 )
#endif
#ifndef IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES
    #define IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES    ( 0 )
#endif
#ifndef IOT_MQTT_RESPONSE_WAIT_MS
    #define IOT_MQTT_RESPONSE_WAIT_MS               ( 1000U )
#endif
#ifndef IOT_MQTT_RETRY_MS_CEILING
    #define IOT_MQTT_RETRY_MS_CEILING               ( 60000U )
#endif

#define MQTT_SERVER_MAX_CLIENTID_LENGTH                        ( ( uint16_t ) 23 )          
#define MQTT_SERVER_MAX_PUBLISH_PAYLOAD_LENGTH                 ( ( size_t ) ( 268435456 ) ) 
#define MQTT_SERVER_MAX_LWT_PAYLOAD_LENGTH                     ( ( size_t ) UINT16_MAX )    
/*
 * Constants related to limits defined in AWS Service Limits.
 *
 * For details, see
 * https://docs.aws.amazon.com/general/latest/gr/aws_service_limits.html
 *
 * Used to validate parameters if when connecting to an AWS IoT MQTT server.
 */
#define AWS_IOT_MQTT_SERVER_MAX_CLIENTID_LENGTH                ( ( uint16_t ) 128 )      
#define AWS_IOT_MQTT_SERVER_MAX_TOPIC_LENGTH                   ( ( uint16_t ) 256 )      
#define AWS_IOT_MQTT_SERVER_MAX_TOPIC_FILTERS_PER_SUBSCRIBE    ( ( size_t ) 8 )          
#define AWS_IOT_MQTT_SERVER_MAX_PUBLISH_PAYLOAD_LENGTH         ( ( size_t ) ( 131072 ) ) 
/*
 * MQTT control packet type and flags. Always the first byte of an MQTT
 * packet.
 *
 * For details, see
 * http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/csprd02/mqtt-v3.1.1-csprd02.html#_Toc385349757
 */
#define MQTT_PACKET_TYPE_CONNECT                               ( ( uint8_t ) 0x10U ) 
#define MQTT_PACKET_TYPE_CONNACK                               ( ( uint8_t ) 0x20U ) 
#define MQTT_PACKET_TYPE_PUBLISH                               ( ( uint8_t ) 0x30U ) 
#define MQTT_PACKET_TYPE_PUBACK                                ( ( uint8_t ) 0x40U ) 
#define MQTT_PACKET_TYPE_SUBSCRIBE                             ( ( uint8_t ) 0x82U ) 
#define MQTT_PACKET_TYPE_SUBACK                                ( ( uint8_t ) 0x90U ) 
#define MQTT_PACKET_TYPE_UNSUBSCRIBE                           ( ( uint8_t ) 0xa2U ) 
#define MQTT_PACKET_TYPE_UNSUBACK                              ( ( uint8_t ) 0xb0U ) 
#define MQTT_PACKET_TYPE_PINGREQ                               ( ( uint8_t ) 0xc0U ) 
#define MQTT_PACKET_TYPE_PINGRESP                              ( ( uint8_t ) 0xd0U ) 
#define MQTT_PACKET_TYPE_DISCONNECT                            ( ( uint8_t ) 0xe0U ) 
#define MQTT_REMAINING_LENGTH_INVALID                          ( ( size_t ) 268435456 )

#define MQTT_INTERNAL_FLAG_BLOCK_ON_SEND                       ( 0x80000000U )

#define MQTT_REMOVE_ALL_SUBSCRIPTIONS                          ( -1 )

/*---------------------- MQTT internal data structures ----------------------*/

struct _mqttConnection;
typedef struct _mqttOperation
{
    /* Pointers to neighboring queue elements. */
    IotLink_t link;                           
    bool incomingPublish;                     
    struct _mqttConnection * pMqttConnection; 
    IotTaskPoolJobStorage_t jobStorage;       
    IotTaskPoolJob_t job;                     
    /* MISRA rule 19.2 doesn't allow usage of union
     * but it is intentionally used here to reduce the size of struct. */
    /* coverity[misra_c_2012_rule_19_2_violation] */
    union
    {
        /* If incomingPublish is false, this struct is valid. */
        struct
        {
            /* Basic operation information. */
            int32_t jobReference;        
            IotMqttOperationType_t type; 
            uint32_t flags;              
            uint16_t packetIdentifier;   
            /* Serialized packet and size. */
            uint8_t * pMqttPacket;           
            uint8_t * pPacketIdentifierHigh; 
            size_t packetSize;               
            /* How to notify of an operation's completion. */

            /* MISRA rule 19.2 doesn't allow usage of union
             * but it is intentionally used here to reduce the size of struct. */
            /* coverity[misra_c_2012_rule_19_2_violation] */
            union
            {
                IotSemaphore_t waitSemaphore;   
                IotMqttCallbackInfo_t callback; 
            } notify;                           
            IotMqttError_t status;              
            /* MISRA rule 19.2 doesn't allow usage of union
             * but it is intentionally used here to reduce the size of struct. */
            /* coverity[misra_c_2012_rule_19_2_violation] */
            union
            {
                struct
                {
                    uint32_t count;        
                    uint32_t limit;        
                    uint32_t nextPeriodMs; 
                } retry;                   
                struct
                {
                    uint32_t failure;      
                    uint32_t keepAliveMs;  
                    uint32_t nextPeriodMs; 
                } ping;                    
            } periodic;                    
        } operation;

        /* If incomingPublish is true, this struct is valid. */
        struct
        {
            IotMqttPublishInfo_t publishInfo; 
            void * pReceivedData;             
        } publish;
    } u;                                      
} _mqttOperation_t;

typedef struct _mqttConnection
{
    bool awsIotMqttMode;                             
    bool ownNetworkConnection;                       
    IotNetworkConnection_t pNetworkConnection;       
    const IotNetworkInterface_t * pNetworkInterface; 
    IotMqttCallbackInfo_t disconnectCallback;        
    const IotMqttSerializer_t * pSerializer;         
    bool disconnected;                               
    IotMutex_t referencesMutex;                      
    int32_t references;                              
    IotListDouble_t pendingProcessing;               
    IotListDouble_t pendingResponse;                 
    IotListDouble_t subscriptionList;                
    IotMutex_t subscriptionMutex;                    
    uint64_t lastMessageTime;                        
    _mqttOperation_t pingreq;                        
} _mqttConnection_t;

typedef struct _mqttSubscription
{
    IotLink_t link;     
    int32_t references; 
    bool unsubscribed;

    struct
    {
        uint16_t identifier;        
        size_t order;               
    } packetInfo;                   
    IotMqttCallbackInfo_t callback; 
    uint16_t topicFilterLength;     
    /* A flexible length array is used here so that the topic filter may
     * be of an arbitrary length. */
    /* coverity[misra_c_2012_rule_18_7_violation] */
    char pTopicFilter[]; 
} _mqttSubscription_t;

typedef struct _mqttPacket
{
    /* MISRA rule 19.2 doesn't allow usage of union
     * but it is intentionally used here to reduce the size of struct. */
    /* coverity[misra_c_2012_rule_19_2_violation] */
    union
    {
        _mqttConnection_t * pMqttConnection;

        _mqttOperation_t * pIncomingPublish;
    } u;                       
    uint8_t * pRemainingData;  
    size_t remainingLength;    
    uint16_t packetIdentifier; 
    uint8_t type;              
} _mqttPacket_t;

/*-------------------- MQTT struct validation functions ---------------------*/

bool _IotMqtt_ValidateConnect( const IotMqttConnectInfo_t * pConnectInfo );

bool _IotMqtt_ValidatePublish( bool awsIotMqttMode,
                               const IotMqttPublishInfo_t * pPublishInfo,
                               uint32_t flags,
                               const IotMqttCallbackInfo_t * pCallbackInfo,
                               const IotMqttOperation_t * const pPublishOperation );

bool _IotMqtt_ValidateLwtPublish( bool awsIotMqttMode,
                                  const IotMqttPublishInfo_t * pLwtPublishInfo );

bool _IotMqtt_ValidateOperation( IotMqttOperation_t operation );

bool _IotMqtt_ValidateSubscriptionList( IotMqttOperationType_t operation,
                                        bool awsIotMqttMode,
                                        const IotMqttSubscription_t * pListStart,
                                        size_t listSize );

/*-------------------- MQTT packet serializer functions ---------------------*/

uint8_t _IotMqtt_GetPacketType( IotNetworkConnection_t pNetworkConnection,
                                const IotNetworkInterface_t * pNetworkInterface );

size_t _IotMqtt_GetRemainingLength( IotNetworkConnection_t pNetworkConnection,
                                    const IotNetworkInterface_t * pNetworkInterface );
IotMqttError_t _IotMqtt_SerializeConnect( const IotMqttConnectInfo_t * pConnectInfo,
                                          uint8_t ** pConnectPacket,
                                          size_t * pPacketSize );

IotMqttError_t _IotMqtt_DeserializeConnack( _mqttPacket_t * pConnack );

IotMqttError_t _IotMqtt_SerializePublish( const IotMqttPublishInfo_t * pPublishInfo,
                                          uint8_t ** pPublishPacket,
                                          size_t * pPacketSize,
                                          uint16_t * pPacketIdentifier,
                                          uint8_t ** pPacketIdentifierHigh );

void _IotMqtt_PublishSetDup( uint8_t * pPublishPacket,
                             uint8_t * pPacketIdentifierHigh,
                             uint16_t * pNewPacketIdentifier );

IotMqttError_t _IotMqtt_DeserializePublish( _mqttPacket_t * pPublish );

IotMqttError_t _IotMqtt_SerializePuback( uint16_t packetIdentifier,
                                         uint8_t ** pPubackPacket,
                                         size_t * pPacketSize );

IotMqttError_t _IotMqtt_DeserializePuback( _mqttPacket_t * pPuback );

IotMqttError_t _IotMqtt_SerializeSubscribe( const IotMqttSubscription_t * pSubscriptionList,
                                            size_t subscriptionCount,
                                            uint8_t ** pSubscribePacket,
                                            size_t * pPacketSize,
                                            uint16_t * pPacketIdentifier );

IotMqttError_t _IotMqtt_DeserializeSuback( _mqttPacket_t * pSuback );

IotMqttError_t _IotMqtt_SerializeUnsubscribe( const IotMqttSubscription_t * pSubscriptionList,
                                              size_t subscriptionCount,
                                              uint8_t ** pUnsubscribePacket,
                                              size_t * pPacketSize,
                                              uint16_t * pPacketIdentifier );

IotMqttError_t _IotMqtt_DeserializeUnsuback( _mqttPacket_t * pUnsuback );

IotMqttError_t _IotMqtt_SerializePingreq( uint8_t ** pPingreqPacket,
                                          size_t * pPacketSize );

IotMqttError_t _IotMqtt_DeserializePingresp( _mqttPacket_t * pPingresp );

IotMqttError_t _IotMqtt_SerializeDisconnect( uint8_t ** pDisconnectPacket,
                                             size_t * pPacketSize );

void _IotMqtt_FreePacket( uint8_t * pPacket );

/*-------------------- MQTT operation record functions ----------------------*/

IotMqttError_t _IotMqtt_CreateOperation( _mqttConnection_t * pMqttConnection,
                                         uint32_t flags,
                                         const IotMqttCallbackInfo_t * pCallbackInfo,
                                         _mqttOperation_t ** pNewOperation );

bool _IotMqtt_DecrementOperationReferences( _mqttOperation_t * pOperation,
                                            bool cancelJob );

void _IotMqtt_DestroyOperation( _mqttOperation_t * pOperation );

void _IotMqtt_ProcessKeepAlive( IotTaskPool_t pTaskPool,
                                IotTaskPoolJob_t pKeepAliveJob,
                                void * pContext );

void _IotMqtt_ProcessIncomingPublish( IotTaskPool_t pTaskPool,
                                      IotTaskPoolJob_t pPublishJob,
                                      void * pContext );

void _IotMqtt_ProcessSend( IotTaskPool_t pTaskPool,
                           IotTaskPoolJob_t pSendJob,
                           void * pContext );

void _IotMqtt_ProcessCompletedOperation( IotTaskPool_t pTaskPool,
                                         IotTaskPoolJob_t pOperationJob,
                                         void * pContext );

IotMqttError_t _IotMqtt_ScheduleOperation( _mqttOperation_t * pOperation,
                                           IotTaskPoolRoutine_t jobRoutine,
                                           uint32_t delay );

_mqttOperation_t * _IotMqtt_FindOperation( _mqttConnection_t * pMqttConnection,
                                           IotMqttOperationType_t type,
                                           const uint16_t * pPacketIdentifier );

void _IotMqtt_Notify( _mqttOperation_t * pOperation );

/*----------------- MQTT subscription management functions ------------------*/

IotMqttError_t _IotMqtt_AddSubscriptions( _mqttConnection_t * pMqttConnection,
                                          uint16_t subscribePacketIdentifier,
                                          const IotMqttSubscription_t * pSubscriptionList,
                                          size_t subscriptionCount );

void _IotMqtt_InvokeSubscriptionCallback( _mqttConnection_t * pMqttConnection,
                                          IotMqttCallbackParam_t * pCallbackParam );

void _IotMqtt_RemoveSubscriptionByPacket( _mqttConnection_t * pMqttConnection,
                                          uint16_t packetIdentifier,
                                          int32_t order );

void _IotMqtt_RemoveSubscriptionByTopicFilter( _mqttConnection_t * pMqttConnection,
                                               const IotMqttSubscription_t * pSubscriptionList,
                                               size_t subscriptionCount );

/*------------------ MQTT connection management functions -------------------*/

bool _IotMqtt_IncrementConnectionReferences( _mqttConnection_t * pMqttConnection );

void _IotMqtt_DecrementConnectionReferences( _mqttConnection_t * pMqttConnection );

bool _IotMqtt_GetNextByte( IotNetworkConnection_t pNetworkConnection,
                           const IotNetworkInterface_t * pNetworkInterface,
                           uint8_t * pIncomingByte );

void _IotMqtt_CloseNetworkConnection( IotMqttDisconnectReason_t disconnectReason,
                                      _mqttConnection_t * pMqttConnection );

#if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1

    #define SERIALIZER_OVERRIDE_SELECTOR( funcType_t, funcName, defaultFunc, serializerMember ) \
    static funcType_t funcName( const IotMqttSerializer_t * pSerializer );                      \
    static funcType_t funcName( const IotMqttSerializer_t * pSerializer )                       \
    {                                                                                           \
        funcType_t returnValue = defaultFunc;                                                   \
        if( pSerializer != NULL )                                                               \
        {                                                                                       \
            if( pSerializer->serializerMember != NULL )                                         \
            {                                                                                   \
                returnValue = pSerializer->serializerMember;                                    \
            }                                                                                   \
        }                                                                                       \
                                                                                                \
        return returnValue;                                                                     \
    }
#endif /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */

#endif /* ifndef IOT_MQTT_INTERNAL_H_ */