iot_mqtt_internal.h

Go to the documentation of this file.

/*
 * 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 "private/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               ( 1000 )
#endif
#ifndef IOT_MQTT_RETRY_MS_CEILING
    #define IOT_MQTT_RETRY_MS_CEILING               ( 60000 )
#endif
 
#ifndef EMPTY_ELSE_MARKER
    #define EMPTY_ELSE_MARKER
#endif
 
/*
 * 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                       ( 128 )  
#define AWS_IOT_MQTT_SERVER_MAX_TOPIC_LENGTH                   ( 256 )  
#define AWS_IOT_MQTT_SERVER_MAX_TOPIC_FILTERS_PER_SUBSCRIBE    ( 8 )    
/*
 * 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                       ( 0x80000000 )
 
/*---------------------- 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;                     
    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. */
            union
            {
                IotSemaphore_t waitSemaphore;   
                IotMqttCallbackInfo_t callback; 
            } notify;                           
            IotMqttError_t status;              
            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; 
            const void * pReceivedData;       
        } publish;
    } u;                                      
} _mqttOperation_t;
 
typedef struct _mqttConnection
{
    bool awsIotMqttMode;                             
    bool ownNetworkConnection;                       
    void * pNetworkConnection;                       
    const IotNetworkInterface_t * pNetworkInterface; 
    IotMqttCallbackInfo_t disconnectCallback;        
    #if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1
        const IotMqttSerializer_t * pSerializer; 
    #endif
 
    bool disconnected;                 
    IotMutex_t referencesMutex;        
    int32_t references;                
    IotListDouble_t pendingProcessing; 
    IotListDouble_t pendingResponse;   
    IotListDouble_t subscriptionList;  
    IotMutex_t subscriptionMutex;      
    _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;     
    char pTopicFilter[];            
} _mqttSubscription_t;
 
typedef struct _mqttPacket
{
    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 );
 
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( void * pNetworkConnection,
                                const IotNetworkInterface_t * pNetworkInterface );
 
size_t _IotMqtt_GetRemainingLength( void * 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( void * pNetworkConnection,
                           const IotNetworkInterface_t * pNetworkInterface,
                           uint8_t * pIncomingByte );
 
void _IotMqtt_CloseNetworkConnection( IotMqttDisconnectReason_t disconnectReason,
                                      _mqttConnection_t * pMqttConnection );
 
#endif /* ifndef IOT_MQTT_INTERNAL_H_ */