/*
 * 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.
 */

/**
 * @file iot_mqtt_api.c
 * @brief Implements most user-facing functions of the MQTT library.
 */

/* The config header is always included first. */
#include "iot_config.h"

/* Standard includes. */
#include <string.h>

/* Error handling include. */
#include "iot_error.h"

/* MQTT internal include. */
#include "private/iot_mqtt_internal.h"

/* Platform layer includes. */
#include "platform/iot_clock.h"
#include "platform/iot_threads.h"

/* Atomics include. */
#include "iot_atomic.h"

/* Validate MQTT configuration settings. */
#if IOT_MQTT_ENABLE_ASSERTS != 0 && IOT_MQTT_ENABLE_ASSERTS != 1
    #error "IOT_MQTT_ENABLE_ASSERTS must be 0 or 1."
#endif
#if IOT_MQTT_ENABLE_METRICS != 0 && IOT_MQTT_ENABLE_METRICS != 1
    #error "IOT_MQTT_ENABLE_METRICS must be 0 or 1."
#endif
#if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES != 0 && IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES != 1
    #error "IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES must be 0 or 1."
#endif
#if IOT_MQTT_RESPONSE_WAIT_MS <= 0
    #error "IOT_MQTT_RESPONSE_WAIT_MS cannot be 0 or negative."
#endif
#if IOT_MQTT_RETRY_MS_CEILING <= 0
    #error "IOT_MQTT_RETRY_MS_CEILING cannot be 0 or negative."
#endif

/*-----------------------------------------------------------*/

/**
 * @brief Uninitialized value for @ref _initCalled.
 */
#define MQTT_LIBRARY_UNINITIALIZED    ( ( uint32_t ) 0 )

/**
 * @brief Initialized value for @ref _initCalled.
 */
#define MQTT_LIBRARY_INITIALIZED      ( ( uint32_t ) 1 )

/*-----------------------------------------------------------*/

/**
 * @brief Check if the library is initialized.
 *
 * @return `true` if IotMqtt_Init was called; `false` otherwise.
 */
static bool _checkInit( void );

/**
 * @brief Set the unsubscribed flag of an MQTT subscription.
 *
 * @param[in] pSubscriptionLink Pointer to the link member of an #_mqttSubscription_t.
 * @param[in] pMatch Not used.
 *
 * @return Always returns `true`.
 */
static bool _mqttSubscription_setUnsubscribe( const IotLink_t * pSubscriptionLink,
                                              void * pMatch );

/**
 * @brief Destroy an MQTT subscription if its reference count is 0.
 *
 * @param[in] pData The subscription to destroy. This parameter is of type
 * `void*` for compatibility with [free]
 * (http://pubs.opengroup.org/onlinepubs/9699919799/functions/free.html).
 */
static void _mqttSubscription_tryDestroy( void * pData );

/**
 * @brief Decrement the reference count of an MQTT operation and attempt to
 * destroy it.
 *
 * @param[in] pData The operation data to destroy. This parameter is of type
 * `void*` for compatibility with [free]
 * (http://pubs.opengroup.org/onlinepubs/9699919799/functions/free.html).
 */
static void _mqttOperation_tryDestroy( void * pData );

/**
 * @brief Initialize the keep-alive operation for an MQTT connection.
 *
 * @param[in] pNetworkInfo User-provided network information for the new
 * connection.
 * @param[in] keepAliveSeconds User-provided keep-alive interval.
 * @param[out] pMqttConnection The MQTT connection associated with the keep-alive.
 *
 * @return `true` if the keep-alive job was successfully created; `false` otherwise.
 */
static bool _createKeepAliveOperation( const IotMqttNetworkInfo_t * pNetworkInfo,
                                       uint16_t keepAliveSeconds,
                                       _mqttConnection_t * pMqttConnection );

/**
 * @brief Initialize a network connection, creating it if necessary.
 *
 * @param[in] pNetworkInfo User-provided network information for the connection
 * connection.
 * @param[out] pNetworkConnection On success, the created and/or initialized network connection.
 * @param[out] pCreatedNewNetworkConnection On success, `true` if a new network connection was created; `false` if an existing one will be used.
 *
 * @return Any #IotNetworkError_t, as defined by the network stack.
 */
static IotNetworkError_t _createNetworkConnection( const IotMqttNetworkInfo_t * pNetworkInfo,
                                                   IotNetworkConnection_t * pNetworkConnection,
                                                   bool * pCreatedNewNetworkConnection );

/**
 * @brief Creates a new MQTT connection and initializes its members.
 *
 * @param[in] awsIotMqttMode Specifies if this connection is to an AWS IoT MQTT server.
 * @param[in] pNetworkInfo User-provided network information for the new
 * connection.
 * @param[in] keepAliveSeconds User-provided keep-alive interval for the new connection.
 *
 * @return Pointer to a newly-created MQTT connection; `NULL` on failure.
 */
static _mqttConnection_t * _createMqttConnection( bool awsIotMqttMode,
                                                  const IotMqttNetworkInfo_t * pNetworkInfo,
                                                  uint16_t keepAliveSeconds );

/**
 * @brief Destroys the members of an MQTT connection.
 *
 * @param[in] pMqttConnection Which connection to destroy.
 */
static void _destroyMqttConnection( _mqttConnection_t * pMqttConnection );

/**
 * @brief Common setup function for subscribe and unsubscribe operations.
 *
 * See @ref mqtt_function_subscribeasync or @ref mqtt_function_unsubscribeasync for a
 * description of the parameters and return values.
 */
static IotMqttError_t _subscriptionCommonSetup( IotMqttOperationType_t operation,
                                                IotMqttConnection_t mqttConnection,
                                                const IotMqttSubscription_t * pSubscriptionList,
                                                size_t subscriptionCount,
                                                uint32_t flags,
                                                IotMqttOperation_t * const pOperationReference );

/**
 * @brief Utility function for creating and serializing subscription requests
 *
 * See @ref mqtt_function_subscribeasync or @ref mqtt_function_unsubscribeasync for a
 * description of the parameters and return values.
 */
static IotMqttError_t _subscriptionCreateAndSerialize( IotMqttOperationType_t operation,
                                                       IotMqttConnection_t mqttConnection,
                                                       IotMqttSerializeSubscribe_t serializeSubscription,
                                                       const IotMqttSubscription_t * pSubscriptionList,
                                                       size_t subscriptionCount,
                                                       uint32_t flags,
                                                       const IotMqttCallbackInfo_t * pCallbackInfo,
                                                       _mqttOperation_t ** ppSubscriptionOperation );

/**
 * @brief The common component of both @ref mqtt_function_subscribeasync and @ref
 * mqtt_function_unsubscribeasync.
 *
 * See @ref mqtt_function_subscribeasync or @ref mqtt_function_unsubscribeasync for a
 * description of the parameters and return values.
 */
static IotMqttError_t _subscriptionCommon( IotMqttOperationType_t operation,
                                           IotMqttConnection_t mqttConnection,
                                           IotMqttSerializeSubscribe_t serializeSubscription,
                                           const IotMqttSubscription_t * pSubscriptionList,
                                           size_t subscriptionCount,
                                           uint32_t flags,
                                           const IotMqttCallbackInfo_t * pCallbackInfo,
                                           IotMqttOperation_t * const pOperationReference );

/**
 * @cond DOXYGEN_IGNORE
 * Doxygen should ignore this section.
 *
 * Declaration of local MQTT serializer override selectors
 */
#if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqttSerializePingreq_t,
                                   _getMqttPingreqSerializer,
                                   _IotMqtt_SerializePingreq,
                                   serialize.pingreq )
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqtt_SerializePublish_t,
                                   _getMqttPublishSerializer,
                                   _IotMqtt_SerializePublish,
                                   serialize.publish )
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqttFreePacket_t,
                                   _getMqttFreePacketFunc,
                                   _IotMqtt_FreePacket,
                                   freePacket )
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqttSerializeSubscribe_t,
                                   _getMqttSubscribeSerializer,
                                   _IotMqtt_SerializeSubscribe,
                                   serialize.subscribe )
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqttSerializeSubscribe_t,
                                   _getMqttUnsubscribeSerializer,
                                   _IotMqtt_SerializeUnsubscribe,
                                   serialize.unsubscribe )
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqttSerializeConnect_t,
                                   _getMqttConnectSerializer,
                                   _IotMqtt_SerializeConnect,
                                   serialize.connect )
    _SERIALIZER_OVERRIDE_SELECTOR( IotMqttSerializeDisconnect_t,
                                   _getMqttDisconnectSerializer,
                                   _IotMqtt_SerializeDisconnect,
                                   serialize.disconnect )
#else /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */
    #define _getMqttPingreqSerializer( pSerializer )        _IotMqtt_SerializePingreq
    #define _getMqttPublishSerializer( pSerializer )        _IotMqtt_SerializePublish
    #define _getMqttFreePacketFunc( pSerializer )           _IotMqtt_FreePacket
    #define _getMqttSubscribeSerializer( pSerializer )      _IotMqtt_SerializeSubscribe
    #define _getMqttUnsubscribeSerializer( pSerializer )    _IotMqtt_SerializeUnsubscribe
    #define _getMqttConnectSerializer( pSerializer )        _IotMqtt_SerializeConnect
    #define _getMqttDisconnectSerializer( pSerializer )     _IotMqtt_SerializeDisconnect
#endif /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */
/** @endcond */

/*-----------------------------------------------------------*/

/**
 * @brief Tracks whether @ref mqtt_function_init has been called.
 *
 * API functions will fail if @ref mqtt_function_init was not called.
 */
static volatile uint32_t _initCalled = MQTT_LIBRARY_UNINITIALIZED;

/*-----------------------------------------------------------*/

static bool _checkInit( void )
{
    bool status = true;

    if( _initCalled == MQTT_LIBRARY_UNINITIALIZED )
    {
        IotLogError( "IotMqtt_Init was not called." );

        status = false;
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    return status;
}

/*-----------------------------------------------------------*/

static bool _mqttSubscription_setUnsubscribe( const IotLink_t * pSubscriptionLink,
                                              void * pMatch )
{
    /* Because this function is called from a container function, the given link
     * must never be NULL. */
    IotMqtt_Assert( pSubscriptionLink != NULL );

    _mqttSubscription_t * pSubscription = IotLink_Container( _mqttSubscription_t,
                                                             pSubscriptionLink,
                                                             link );

    /* Silence warnings about unused parameters. */
    ( void ) pMatch;

    /* Set the unsubscribed flag. */
    pSubscription->unsubscribed = true;

    return true;
}

/*-----------------------------------------------------------*/

static void _mqttSubscription_tryDestroy( void * pData )
{
    _mqttSubscription_t * pSubscription = ( _mqttSubscription_t * ) pData;

    /* Reference count must not be negative. */
    IotMqtt_Assert( pSubscription->references >= 0 );

    /* Unsubscribed flag should be set. */
    IotMqtt_Assert( pSubscription->unsubscribed == true );

    /* Free the subscription if it has no references. */
    if( pSubscription->references == 0 )
    {
        IotMqtt_FreeSubscription( pSubscription );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }
}

/*-----------------------------------------------------------*/

static void _mqttOperation_tryDestroy( void * pData )
{
    _mqttOperation_t * pOperation = ( _mqttOperation_t * ) pData;
    IotTaskPoolError_t taskPoolStatus = IOT_TASKPOOL_SUCCESS;

    /* Incoming PUBLISH operations may always be freed. */
    if( pOperation->incomingPublish == true )
    {
        /* Cancel the incoming PUBLISH operation's job. */
        taskPoolStatus = IotTaskPool_TryCancel( IOT_SYSTEM_TASKPOOL,
                                                pOperation->job,
                                                NULL );

        /* If the operation's job was not canceled, it must be already executing.
         * Any other return value is invalid. */
        IotMqtt_Assert( ( taskPoolStatus == IOT_TASKPOOL_SUCCESS ) ||
                        ( taskPoolStatus == IOT_TASKPOOL_CANCEL_FAILED ) );

        /* Check if the incoming PUBLISH job was canceled. */
        if( taskPoolStatus == IOT_TASKPOOL_SUCCESS )
        {
            /* Job was canceled. Process incoming PUBLISH now to clean up. */
            _IotMqtt_ProcessIncomingPublish( IOT_SYSTEM_TASKPOOL,
                                             pOperation->job,
                                             pOperation );
        }
        else
        {
            /* The executing job will process the PUBLISH, so nothing is done here. */
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        /* Decrement reference count and destroy operation if possible. */
        if( _IotMqtt_DecrementOperationReferences( pOperation, true ) == true )
        {
            _IotMqtt_DestroyOperation( pOperation );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
}

/*-----------------------------------------------------------*/

static bool _createKeepAliveOperation( const IotMqttNetworkInfo_t * pNetworkInfo,
                                       uint16_t keepAliveSeconds,
                                       _mqttConnection_t * pMqttConnection )
{
    bool status = true;
    IotMqttError_t serializeStatus = IOT_MQTT_SUCCESS;
    IotTaskPoolError_t jobStatus = IOT_TASKPOOL_SUCCESS;

    /* Network information is not used when MQTT packet serializers are disabled. */
    ( void ) pNetworkInfo;

    /* Set PINGREQ operation members. */
    pMqttConnection->pingreq.u.operation.type = IOT_MQTT_PINGREQ;

    /* Convert the keep-alive interval to milliseconds. */
    pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs = keepAliveSeconds * 1000;
    pMqttConnection->pingreq.u.operation.periodic.ping.nextPeriodMs = keepAliveSeconds * 1000;

    /* Generate a PINGREQ packet. */
    serializeStatus = _getMqttPingreqSerializer( pMqttConnection->pSerializer )( &( pMqttConnection->pingreq.u.operation.pMqttPacket ),
                                                                                 &( pMqttConnection->pingreq.u.operation.packetSize ) );

    if( serializeStatus != IOT_MQTT_SUCCESS )
    {
        IotLogError( "Failed to allocate PINGREQ packet for new connection." );

        status = false;
    }
    else
    {
        /* Create the task pool job that processes keep-alive. */
        jobStatus = IotTaskPool_CreateJob( _IotMqtt_ProcessKeepAlive,
                                           pMqttConnection,
                                           &( pMqttConnection->pingreq.jobStorage ),
                                           &( pMqttConnection->pingreq.job ) );

        /* Task pool job creation for a pre-allocated job should never fail.
         * Abort the program if it does. */
        if( jobStatus != IOT_TASKPOOL_SUCCESS )
        {
            IotLogError( "Failed to create keep-alive job for new connection." );

            IotMqtt_Assert( false );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        /* Keep-alive references its MQTT connection, so increment reference. */
        ( pMqttConnection->references )++;
    }

    return status;
}

/*-----------------------------------------------------------*/

static IotNetworkError_t _createNetworkConnection( const IotMqttNetworkInfo_t * pNetworkInfo,
                                                   IotNetworkConnection_t * pNetworkConnection,
                                                   bool * pCreatedNewNetworkConnection )
{
    IOT_FUNCTION_ENTRY( IotNetworkError_t, IOT_NETWORK_SUCCESS );

    /* Network info must not be NULL. */
    if( pNetworkInfo == NULL )
    {
        IotLogError( "Network information cannot be NULL." );

        IOT_SET_AND_GOTO_CLEANUP( IOT_NETWORK_BAD_PARAMETER );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Create a new network connection if requested. Otherwise, copy the existing
     * network connection. */
    if( pNetworkInfo->createNetworkConnection == true )
    {
        status = pNetworkInfo->pNetworkInterface->create( pNetworkInfo->u.setup.pNetworkServerInfo,
                                                          pNetworkInfo->u.setup.pNetworkCredentialInfo,
                                                          pNetworkConnection );

        if( status == IOT_NETWORK_SUCCESS )
        {
            /* This MQTT connection owns the network connection it created and
             * should destroy it on cleanup. */
            *pCreatedNewNetworkConnection = true;
        }
        else
        {
            IotLogError( "Failed to create network connection: %d", status );

            IOT_GOTO_CLEANUP();
        }
    }
    else
    {
        /* A connection already exists; the caller should not destroy
         * it on cleanup. */
        *pNetworkConnection = pNetworkInfo->u.pNetworkConnection;
        *pCreatedNewNetworkConnection = false;
    }

    IOT_FUNCTION_EXIT_NO_CLEANUP();
}

/*-----------------------------------------------------------*/

static _mqttConnection_t * _createMqttConnection( bool awsIotMqttMode,
                                                  const IotMqttNetworkInfo_t * pNetworkInfo,
                                                  uint16_t keepAliveSeconds )
{
    IOT_FUNCTION_ENTRY( bool, true );
    _mqttConnection_t * pMqttConnection = NULL;
    bool referencesMutexCreated = false, subscriptionMutexCreated = false;

    /* Allocate memory for the new MQTT connection. */
    pMqttConnection = IotMqtt_MallocConnection( sizeof( _mqttConnection_t ) );

    if( pMqttConnection == NULL )
    {
        IotLogError( "Failed to allocate memory for new connection." );

        IOT_SET_AND_GOTO_CLEANUP( false );
    }
    else
    {
        /* Clear the MQTT connection, then copy the MQTT server mode, network
         * interface, and disconnect callback. */
        ( void ) memset( pMqttConnection, 0x00, sizeof( _mqttConnection_t ) );
        pMqttConnection->awsIotMqttMode = awsIotMqttMode;
        pMqttConnection->pNetworkInterface = pNetworkInfo->pNetworkInterface;
        pMqttConnection->disconnectCallback = pNetworkInfo->disconnectCallback;

        /* Start a new MQTT connection with a reference count of 1. */
        pMqttConnection->references = 1;
    }

    /* Create the references mutex for a new connection. It is a recursive mutex. */
    referencesMutexCreated = IotMutex_Create( &( pMqttConnection->referencesMutex ), true );

    if( referencesMutexCreated == false )
    {
        IotLogError( "Failed to create references mutex for new connection." );

        IOT_SET_AND_GOTO_CLEANUP( false );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Create the subscription mutex for a new connection. */
    subscriptionMutexCreated = IotMutex_Create( &( pMqttConnection->subscriptionMutex ), false );

    if( subscriptionMutexCreated == false )
    {
        IotLogError( "Failed to create subscription mutex for new connection." );

        IOT_SET_AND_GOTO_CLEANUP( false );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Create the new connection's subscription and operation lists. */
    IotListDouble_Create( &( pMqttConnection->subscriptionList ) );
    IotListDouble_Create( &( pMqttConnection->pendingProcessing ) );
    IotListDouble_Create( &( pMqttConnection->pendingResponse ) );

    /* Check if keep-alive is active for this connection. */
    if( keepAliveSeconds != 0 )
    {
        if( _createKeepAliveOperation( pNetworkInfo,
                                       keepAliveSeconds,
                                       pMqttConnection ) == false )
        {
            IOT_SET_AND_GOTO_CLEANUP( false );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Clean up mutexes and connection if this function failed. */
    IOT_FUNCTION_CLEANUP_BEGIN();

    if( status == false )
    {
        if( subscriptionMutexCreated == true )
        {
            IotMutex_Destroy( &( pMqttConnection->subscriptionMutex ) );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        if( referencesMutexCreated == true )
        {
            IotMutex_Destroy( &( pMqttConnection->referencesMutex ) );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        if( pMqttConnection != NULL )
        {
            IotMqtt_FreeConnection( pMqttConnection );
            pMqttConnection = NULL;
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    return pMqttConnection;
}

/*-----------------------------------------------------------*/

static void _destroyMqttConnection( _mqttConnection_t * pMqttConnection )
{
    IotNetworkError_t networkStatus = IOT_NETWORK_SUCCESS;

    /* Clean up keep-alive if still allocated. */
    if( pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs != 0 )
    {
        IotLogDebug( "(MQTT connection %p) Cleaning up keep-alive.", pMqttConnection );

        _getMqttFreePacketFunc( pMqttConnection->pSerializer )( pMqttConnection->pingreq.u.operation.pMqttPacket );

        /* Clear data about the keep-alive. */
        pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs = 0;
        pMqttConnection->pingreq.u.operation.pMqttPacket = NULL;
        pMqttConnection->pingreq.u.operation.packetSize = 0;

        /* Decrement reference count. */
        pMqttConnection->references--;
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* A connection to be destroyed should have no keep-alive and at most 1
     * reference. */
    IotMqtt_Assert( pMqttConnection->references <= 1 );
    IotMqtt_Assert( pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs == 0 );
    IotMqtt_Assert( pMqttConnection->pingreq.u.operation.pMqttPacket == NULL );
    IotMqtt_Assert( pMqttConnection->pingreq.u.operation.packetSize == 0 );

    /* Remove all subscriptions. */
    IotMutex_Lock( &( pMqttConnection->subscriptionMutex ) );
    IotListDouble_RemoveAllMatches( &( pMqttConnection->subscriptionList ),
                                    _mqttSubscription_setUnsubscribe,
                                    NULL,
                                    _mqttSubscription_tryDestroy,
                                    offsetof( _mqttSubscription_t, link ) );
    IotMutex_Unlock( &( pMqttConnection->subscriptionMutex ) );

    /* Destroy an owned network connection. */
    if( pMqttConnection->ownNetworkConnection == true )
    {
        networkStatus = pMqttConnection->pNetworkInterface->destroy( pMqttConnection->pNetworkConnection );

        if( networkStatus != IOT_NETWORK_SUCCESS )
        {
            IotLogWarn( "(MQTT connection %p) Failed to destroy network connection.",
                        pMqttConnection );
        }
        else
        {
            IotLogInfo( "(MQTT connection %p) Network connection destroyed.",
                        pMqttConnection );
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Destroy mutexes. */
    IotMutex_Destroy( &( pMqttConnection->referencesMutex ) );
    IotMutex_Destroy( &( pMqttConnection->subscriptionMutex ) );

    IotLogDebug( "(MQTT connection %p) Connection destroyed.", pMqttConnection );

    /* Free connection. */
    IotMqtt_FreeConnection( pMqttConnection );
}

/*-----------------------------------------------------------*/
static IotMqttError_t _subscriptionCommonSetup( IotMqttOperationType_t operation,
                                                IotMqttConnection_t mqttConnection,
                                                const IotMqttSubscription_t * pSubscriptionList,
                                                size_t subscriptionCount,
                                                uint32_t flags,
                                                IotMqttOperation_t * const pOperationReference )
{
    IOT_FUNCTION_ENTRY( IotMqttError_t, IOT_MQTT_SUCCESS );

    /* This function should only be called for subscribe or unsubscribe. */
    IotMqtt_Assert( ( operation == IOT_MQTT_SUBSCRIBE ) ||
                    ( operation == IOT_MQTT_UNSUBSCRIBE ) );

    /* Check that IotMqtt_Init was called. */
    if( _checkInit() == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NOT_INITIALIZED );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check that all elements in the subscription list are valid. */
    if( _IotMqtt_ValidateSubscriptionList( operation,
                                           mqttConnection->awsIotMqttMode,
                                           pSubscriptionList,
                                           subscriptionCount ) == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check that a reference pointer is provided for a waitable operation. */
    if( ( flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE )
    {
        if( pOperationReference == NULL )
        {
            IotLogError( "Reference must be provided for a waitable %s.",
                         IotMqtt_OperationType( operation ) );

            IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    IOT_FUNCTION_EXIT_NO_CLEANUP();
}

/*-----------------------------------------------------------*/

static IotMqttError_t _subscriptionCreateAndSerialize( IotMqttOperationType_t operation,
                                                       IotMqttConnection_t mqttConnection,
                                                       IotMqttSerializeSubscribe_t serializeSubscription,
                                                       const IotMqttSubscription_t * pSubscriptionList,
                                                       size_t subscriptionCount,
                                                       uint32_t flags,
                                                       const IotMqttCallbackInfo_t * pCallbackInfo,
                                                       _mqttOperation_t ** ppSubscriptionOperation )
{
    IOT_FUNCTION_ENTRY( IotMqttError_t, IOT_MQTT_SUCCESS );
    _mqttOperation_t * pSubscriptionOperation = NULL;

    /* Create a subscription operation. */
    status = _IotMqtt_CreateOperation( mqttConnection,
                                       flags,
                                       pCallbackInfo,
                                       ppSubscriptionOperation );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        pSubscriptionOperation = ( *ppSubscriptionOperation );
    }

    /* Check the subscription operation data and set the operation type. */
    IotMqtt_Assert( pSubscriptionOperation->u.operation.status == IOT_MQTT_STATUS_PENDING );
    IotMqtt_Assert( pSubscriptionOperation->u.operation.periodic.retry.limit == 0 );
    pSubscriptionOperation->u.operation.type = operation;

    /* Generate a subscription packet from the subscription list. */
    status = serializeSubscription( pSubscriptionList,
                                    subscriptionCount,
                                    &( pSubscriptionOperation->u.operation.pMqttPacket ),
                                    &( pSubscriptionOperation->u.operation.packetSize ),
                                    &( pSubscriptionOperation->u.operation.packetIdentifier ) );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check the serialized MQTT packet. */
    IotMqtt_Assert( pSubscriptionOperation->u.operation.pMqttPacket != NULL );
    IotMqtt_Assert( pSubscriptionOperation->u.operation.packetSize > 0 );

    IOT_FUNCTION_EXIT_NO_CLEANUP();
}

/*-----------------------------------------------------------*/

static IotMqttError_t _subscriptionCommon( IotMqttOperationType_t operation,
                                           IotMqttConnection_t mqttConnection,
                                           IotMqttSerializeSubscribe_t serializeSubscription,
                                           const IotMqttSubscription_t * pSubscriptionList,
                                           size_t subscriptionCount,
                                           uint32_t flags,
                                           const IotMqttCallbackInfo_t * pCallbackInfo,
                                           IotMqttOperation_t * const pOperationReference )
{
    IOT_FUNCTION_ENTRY( IotMqttError_t, IOT_MQTT_SUCCESS );
    _mqttOperation_t * pSubscriptionOperation = NULL;

    /* Create and serialize the subscription operation. */
    status = _subscriptionCreateAndSerialize( operation,
                                              mqttConnection,
                                              serializeSubscription,
                                              pSubscriptionList,
                                              subscriptionCount,
                                              flags,
                                              pCallbackInfo,
                                              &pSubscriptionOperation );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Add the subscription list for a SUBSCRIBE. */
    if( operation == IOT_MQTT_SUBSCRIBE )
    {
        status = _IotMqtt_AddSubscriptions( mqttConnection,
                                            pSubscriptionOperation->u.operation.packetIdentifier,
                                            pSubscriptionList,
                                            subscriptionCount );

        if( status != IOT_MQTT_SUCCESS )
        {
            IOT_GOTO_CLEANUP();
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }

    /* Set the reference, if provided. */
    if( pOperationReference != NULL )
    {
        *pOperationReference = pSubscriptionOperation;
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Send the SUBSCRIBE packet. */
    if( ( flags & MQTT_INTERNAL_FLAG_BLOCK_ON_SEND ) == MQTT_INTERNAL_FLAG_BLOCK_ON_SEND )
    {
        _IotMqtt_ProcessSend( IOT_SYSTEM_TASKPOOL, pSubscriptionOperation->job, pSubscriptionOperation );
    }
    else
    {
        status = _IotMqtt_ScheduleOperation( pSubscriptionOperation,
                                             _IotMqtt_ProcessSend,
                                             0 );

        if( status != IOT_MQTT_SUCCESS )
        {
            IotLogError( "(MQTT connection %p) Failed to schedule %s for sending.",
                         mqttConnection,
                         IotMqtt_OperationType( operation ) );

            if( operation == IOT_MQTT_SUBSCRIBE )
            {
                _IotMqtt_RemoveSubscriptionByPacket( mqttConnection,
                                                     pSubscriptionOperation->u.operation.packetIdentifier,
                                                     MQTT_REMOVE_ALL_SUBSCRIPTIONS );
            }
            else
            {
                EMPTY_ELSE_MARKER;
            }

            /* Clear the previously set (and now invalid) reference. */
            if( pOperationReference != NULL )
            {
                *pOperationReference = IOT_MQTT_OPERATION_INITIALIZER;
            }
            else
            {
                EMPTY_ELSE_MARKER;
            }

            IOT_GOTO_CLEANUP();
        }
    }

    /* Clean up if this function failed. */
    IOT_FUNCTION_CLEANUP_BEGIN();

    if( status != IOT_MQTT_SUCCESS )
    {
        if( pSubscriptionOperation != NULL )
        {
            _IotMqtt_DestroyOperation( pSubscriptionOperation );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        status = IOT_MQTT_STATUS_PENDING;

        IotLogInfo( "(MQTT connection %p) %s operation scheduled.",
                    mqttConnection,
                    IotMqtt_OperationType( operation ) );
    }

    IOT_FUNCTION_CLEANUP_END();
}

/*-----------------------------------------------------------*/

bool _IotMqtt_IncrementConnectionReferences( _mqttConnection_t * pMqttConnection )
{
    bool disconnected = false;

    /* Lock the mutex protecting the reference count. */
    IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

    /* Reference count must not be negative. */
    IotMqtt_Assert( pMqttConnection->references >= 0 );

    /* Read connection status. */
    disconnected = pMqttConnection->disconnected;

    /* Increment the connection's reference count if it is not disconnected. */
    if( disconnected == false )
    {
        ( pMqttConnection->references )++;
        IotLogDebug( "(MQTT connection %p) Reference count changed from %ld to %ld.",
                     pMqttConnection,
                     ( long int ) pMqttConnection->references - 1,
                     ( long int ) pMqttConnection->references );
    }
    else
    {
        IotLogWarn( "(MQTT connection %p) Attempt to use closed connection.", pMqttConnection );
    }

    IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

    return( disconnected == false );
}

/*-----------------------------------------------------------*/

void _IotMqtt_DecrementConnectionReferences( _mqttConnection_t * pMqttConnection )
{
    bool destroyConnection = false;

    /* Lock the mutex protecting the reference count. */
    IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

    /* Decrement reference count. It must not be negative. */
    ( pMqttConnection->references )--;
    IotMqtt_Assert( pMqttConnection->references >= 0 );

    IotLogDebug( "(MQTT connection %p) Reference count changed from %ld to %ld.",
                 pMqttConnection,
                 ( long int ) pMqttConnection->references + 1,
                 ( long int ) pMqttConnection->references );

    /* Check if this connection may be destroyed. */
    if( pMqttConnection->references == 0 )
    {
        destroyConnection = true;
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

    /* Destroy an unreferenced MQTT connection. */
    if( destroyConnection == true )
    {
        IotLogDebug( "(MQTT connection %p) Connection will be destroyed now.",
                     pMqttConnection );
        _destroyMqttConnection( pMqttConnection );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_Init( void )
{
    IotMqttError_t status = IOT_MQTT_SUCCESS;
    uint32_t allowInitialization = Atomic_CompareAndSwap_u32( &_initCalled,
                                                              MQTT_LIBRARY_INITIALIZED,
                                                              MQTT_LIBRARY_UNINITIALIZED );

    if( allowInitialization == 1 )
    {
        /* Call any additional serializer initialization function if serializer
         * overrides are enabled. */
        #if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1
            #ifdef _IotMqtt_InitSerializeAdditional
                if( _IotMqtt_InitSerializeAdditional() == false )
                {
                    /* Log initialization status. */
                    IotLogError( "Failed to initialize MQTT library serializer. " );

                    status = IOT_MQTT_INIT_FAILED;
                }
                else
                {
                    EMPTY_ELSE_MARKER;
                }
            #endif /* ifdef _IotMqtt_InitSerializeAdditional */
        #endif /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */

        if( status == IOT_MQTT_SUCCESS )
        {
            IotLogInfo( "MQTT library successfully initialized." );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        IotLogWarn( "IotMqtt_Init called with library already initialized." );
    }

    return status;
}

/*-----------------------------------------------------------*/

void IotMqtt_Cleanup( void )
{
    uint32_t allowCleanup = Atomic_CompareAndSwap_u32( &_initCalled,
                                                       MQTT_LIBRARY_UNINITIALIZED,
                                                       MQTT_LIBRARY_INITIALIZED );

    if( allowCleanup == 1 )
    {
        /* Call any additional serializer cleanup initialization function if serializer
         * overrides are enabled. */
        #if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1
            #ifdef _IotMqtt_CleanupSerializeAdditional
                _IotMqtt_CleanupSerializeAdditional();
            #endif
        #endif

        IotLogInfo( "MQTT library cleanup done." );
    }
    else
    {
        IotLogWarn( "IotMqtt_Init was not called before IotMqtt_Cleanup." );
    }
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_Connect( const IotMqttNetworkInfo_t * pNetworkInfo,
                                const IotMqttConnectInfo_t * pConnectInfo,
                                uint32_t timeoutMs,
                                IotMqttConnection_t * const pMqttConnection )
{
    IOT_FUNCTION_ENTRY( IotMqttError_t, IOT_MQTT_SUCCESS );
    bool ownNetworkConnection = false;
    IotNetworkError_t networkStatus = IOT_NETWORK_SUCCESS;
    IotTaskPoolError_t taskPoolStatus = IOT_TASKPOOL_SUCCESS;
    IotNetworkConnection_t pNetworkConnection = { 0 };
    _mqttOperation_t * pOperation = NULL;
    _mqttConnection_t * pNewMqttConnection = NULL;

    /* Check that IotMqtt_Init was called. */
    if( _checkInit() == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NOT_INITIALIZED );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Validate network interface and connect info. */
    if( _IotMqtt_ValidateConnect( pConnectInfo ) == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    networkStatus = _createNetworkConnection( pNetworkInfo,
                                              &pNetworkConnection,
                                              &ownNetworkConnection );

    if( networkStatus != IOT_NETWORK_SUCCESS )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NETWORK_ERROR );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    IotLogInfo( "Establishing new MQTT connection." );

    /* Initialize a new MQTT connection object. */
    pNewMqttConnection = _createMqttConnection( pConnectInfo->awsIotMqttMode,
                                                pNetworkInfo,
                                                pConnectInfo->keepAliveSeconds );

    if( pNewMqttConnection == NULL )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NO_MEMORY );
    }
    else
    {
        /* Set the network connection associated with the MQTT connection. */
        pNewMqttConnection->pNetworkConnection = pNetworkConnection;
        pNewMqttConnection->ownNetworkConnection = ownNetworkConnection;

        /* Set the MQTT packet serializer overrides. */
        #if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1
            pNewMqttConnection->pSerializer = pNetworkInfo->pMqttSerializer;
        #else
            pNewMqttConnection->pSerializer = NULL;
        #endif
    }

    /* Set the MQTT receive callback. */
    networkStatus = pNewMqttConnection->pNetworkInterface->setReceiveCallback( pNetworkConnection,
                                                                               IotMqtt_ReceiveCallback,
                                                                               pNewMqttConnection );

    if( networkStatus != IOT_NETWORK_SUCCESS )
    {
        IotLogError( "Failed to set MQTT network receive callback." );

        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NETWORK_ERROR );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Create a CONNECT operation. */
    status = _IotMqtt_CreateOperation( pNewMqttConnection,
                                       IOT_MQTT_FLAG_WAITABLE,
                                       NULL,
                                       &pOperation );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Ensure the members set by operation creation and serialization
     * are appropriate for a blocking CONNECT. */
    IotMqtt_Assert( pOperation->u.operation.status == IOT_MQTT_STATUS_PENDING );
    IotMqtt_Assert( ( pOperation->u.operation.flags & IOT_MQTT_FLAG_WAITABLE )
                    == IOT_MQTT_FLAG_WAITABLE );
    IotMqtt_Assert( pOperation->u.operation.periodic.retry.limit == 0 );

    /* Set the operation type. */
    pOperation->u.operation.type = IOT_MQTT_CONNECT;

    /* Add previous session subscriptions. */
    if( pConnectInfo->pPreviousSubscriptions != NULL )
    {
        /* Previous subscription count should have been validated as nonzero. */
        IotMqtt_Assert( pConnectInfo->previousSubscriptionCount > 0 );

        status = _IotMqtt_AddSubscriptions( pNewMqttConnection,
                                            2,
                                            pConnectInfo->pPreviousSubscriptions,
                                            pConnectInfo->previousSubscriptionCount );

        if( status != IOT_MQTT_SUCCESS )
        {
            IOT_GOTO_CLEANUP();
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Convert the connect info and will info objects to an MQTT CONNECT packet. */
    status = _getMqttConnectSerializer( pNetworkInfo->pMqttSerializer )( pConnectInfo,
                                                                         &( pOperation->u.operation.pMqttPacket ),
                                                                         &( pOperation->u.operation.packetSize ) );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check the serialized MQTT packet. */
    IotMqtt_Assert( pOperation->u.operation.pMqttPacket != NULL );
    IotMqtt_Assert( pOperation->u.operation.packetSize > 0 );

    /* Send the CONNECT packet. */
    _IotMqtt_ProcessSend( IOT_SYSTEM_TASKPOOL, pOperation->job, pOperation );

    /* Wait for the CONNECT operation to complete, i.e. wait for CONNACK. */
    status = IotMqtt_Wait( pOperation, timeoutMs );

    /* The call to wait cleans up the CONNECT operation, so set the pointer
     * to NULL. */
    pOperation = NULL;

    /* When a connection is successfully established, schedule keep-alive job. */
    if( status == IOT_MQTT_SUCCESS )
    {
        /* Check if a keep-alive job should be scheduled. */
        if( pNewMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs != 0 )
        {
            IotLogDebug( "Scheduling first MQTT keep-alive job." );

            taskPoolStatus = IotTaskPool_ScheduleDeferred( IOT_SYSTEM_TASKPOOL,
                                                           pNewMqttConnection->pingreq.job,
                                                           pNewMqttConnection->pingreq.u.operation.periodic.ping.nextPeriodMs );

            if( taskPoolStatus != IOT_TASKPOOL_SUCCESS )
            {
                IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_SCHEDULING_ERROR );
            }
            else
            {
                EMPTY_ELSE_MARKER;
            }
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    IOT_FUNCTION_CLEANUP_BEGIN();

    if( status != IOT_MQTT_SUCCESS )
    {
        IotLogError( "Failed to establish new MQTT connection, error %s.",
                     IotMqtt_strerror( status ) );

        /* The network connection must be closed if it was created. */
        if( ownNetworkConnection == true )
        {
            networkStatus = pNetworkInfo->pNetworkInterface->close( pNetworkConnection );

            if( networkStatus != IOT_NETWORK_SUCCESS )
            {
                IotLogWarn( "Failed to close network connection." );
            }
            else
            {
                IotLogInfo( "Network connection closed on error." );
            }
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        if( pOperation != NULL )
        {
            _IotMqtt_DestroyOperation( pOperation );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        if( pNewMqttConnection != NULL )
        {
            _destroyMqttConnection( pNewMqttConnection );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        IotLogInfo( "New MQTT connection %p established.", pMqttConnection );

        /* Set the output parameter. */
        *pMqttConnection = pNewMqttConnection;
    }

    IOT_FUNCTION_CLEANUP_END();
}

/*-----------------------------------------------------------*/

void IotMqtt_Disconnect( IotMqttConnection_t mqttConnection,
                         uint32_t flags )
{
    bool disconnected = false, initCalled = false;
    IotMqttError_t status = IOT_MQTT_STATUS_PENDING;
    _mqttOperation_t * pOperation = NULL;

    /* Check that IotMqtt_Init was called. */
    initCalled = _checkInit();

    if( initCalled == false )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Only send a DISCONNECT packet if the connection is active and the "cleanup only"
     * flag is not set. */
    if( ( flags & IOT_MQTT_FLAG_CLEANUP_ONLY ) == IOT_MQTT_FLAG_CLEANUP_ONLY )
    {
        IOT_GOTO_CLEANUP();
    }

    /* Read the connection status. */
    IotMutex_Lock( &( mqttConnection->referencesMutex ) );
    disconnected = mqttConnection->disconnected;
    IotMutex_Unlock( &( mqttConnection->referencesMutex ) );

    if( disconnected == true )
    {
        IOT_GOTO_CLEANUP();
    }

    IotLogInfo( "(MQTT connection %p) Disconnecting connection.", mqttConnection );

    /* Create a DISCONNECT operation. This function blocks until the DISCONNECT
     * packet is sent, so it sets IOT_MQTT_FLAG_WAITABLE. */
    status = _IotMqtt_CreateOperation( mqttConnection,
                                       IOT_MQTT_FLAG_WAITABLE,
                                       NULL,
                                       &pOperation );

    if( status == IOT_MQTT_SUCCESS )
    {
        /* Ensure that the members set by operation creation and serialization
         * are appropriate for a blocking DISCONNECT. */
        IotMqtt_Assert( pOperation->u.operation.status == IOT_MQTT_STATUS_PENDING );
        IotMqtt_Assert( ( pOperation->u.operation.flags & IOT_MQTT_FLAG_WAITABLE )
                        == IOT_MQTT_FLAG_WAITABLE );
        IotMqtt_Assert( pOperation->u.operation.periodic.retry.limit == 0 );

        /* Set the operation type. */
        pOperation->u.operation.type = IOT_MQTT_DISCONNECT;

        /* Generate a DISCONNECT packet. */
        status = _getMqttDisconnectSerializer( mqttConnection->pSerializer )( &( pOperation->u.operation.pMqttPacket ),
                                                                              &( pOperation->u.operation.packetSize ) );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    if( status == IOT_MQTT_SUCCESS )
    {
        /* Check the serialized MQTT packet. */
        IotMqtt_Assert( pOperation->u.operation.pMqttPacket != NULL );
        IotMqtt_Assert( pOperation->u.operation.packetSize > 0 );

        /* Send the DISCONNECT packet. */
        _IotMqtt_ProcessSend( IOT_SYSTEM_TASKPOOL, pOperation->job, pOperation );

        /* Wait a short time for the DISCONNECT packet to be transmitted. */
        status = IotMqtt_Wait( pOperation,
                               IOT_MQTT_RESPONSE_WAIT_MS );

        /* A wait on DISCONNECT should only ever return SUCCESS, TIMEOUT,
         * or NETWORK ERROR. */
        if( status == IOT_MQTT_SUCCESS )
        {
            IotLogInfo( "(MQTT connection %p) Connection disconnected.", mqttConnection );
        }
        else
        {
            IotMqtt_Assert( ( status == IOT_MQTT_TIMEOUT ) ||
                            ( status == IOT_MQTT_NETWORK_ERROR ) );

            IotLogWarn( "(MQTT connection %p) DISCONNECT not sent, error %s.",
                        mqttConnection,
                        IotMqtt_strerror( status ) );
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* This function has no return value and no cleanup, but uses the cleanup
     * label to exit on error. */
    IOT_FUNCTION_CLEANUP_BEGIN();

    if( initCalled == true )
    {
        /* Close the underlying network connection. This also cleans up keep-alive. */
        _IotMqtt_CloseNetworkConnection( IOT_MQTT_DISCONNECT_CALLED,
                                         mqttConnection );

        /* Check if the connection may be destroyed. */
        IotMutex_Lock( &( mqttConnection->referencesMutex ) );

        /* At this point, the connection should be marked disconnected. */
        IotMqtt_Assert( mqttConnection->disconnected == true );

        /* Attempt cancel and destroy each operation in the connection's lists. */
        IotListDouble_RemoveAll( &( mqttConnection->pendingProcessing ),
                                 _mqttOperation_tryDestroy,
                                 offsetof( _mqttOperation_t, link ) );

        IotListDouble_RemoveAll( &( mqttConnection->pendingResponse ),
                                 _mqttOperation_tryDestroy,
                                 offsetof( _mqttOperation_t, link ) );

        IotMutex_Unlock( &( mqttConnection->referencesMutex ) );

        /* Decrement the connection reference count and destroy it if possible. */
        _IotMqtt_DecrementConnectionReferences( mqttConnection );
    }
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_SubscribeAsync( IotMqttConnection_t mqttConnection,
                                       const IotMqttSubscription_t * pSubscriptionList,
                                       size_t subscriptionCount,
                                       uint32_t flags,
                                       const IotMqttCallbackInfo_t * pCallbackInfo,
                                       IotMqttOperation_t * const pSubscribeOperation )
{
    IotMqttError_t status = _subscriptionCommonSetup( IOT_MQTT_SUBSCRIBE,
                                                      mqttConnection,
                                                      pSubscriptionList,
                                                      subscriptionCount,
                                                      flags,
                                                      pSubscribeOperation );

    if( IOT_MQTT_SUCCESS == status )
    {
        status = _subscriptionCommon( IOT_MQTT_SUBSCRIBE,
                                      mqttConnection,
                                      _getMqttSubscribeSerializer( mqttConnection->pSerializer ),
                                      pSubscriptionList,
                                      subscriptionCount,
                                      flags,
                                      pCallbackInfo,
                                      pSubscribeOperation );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    return status;
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_SubscribeSync( IotMqttConnection_t mqttConnection,
                                      const IotMqttSubscription_t * pSubscriptionList,
                                      size_t subscriptionCount,
                                      uint32_t flags,
                                      uint32_t timeoutMs )
{
    IotMqttError_t status = IOT_MQTT_STATUS_PENDING;
    IotMqttOperation_t subscribeOperation = IOT_MQTT_OPERATION_INITIALIZER;

    /* Flags are not used, but the parameter is present for future compatibility. */
    ( void ) flags;

    /* Call the asynchronous SUBSCRIBE function. */
    status = IotMqtt_SubscribeAsync( mqttConnection,
                                     pSubscriptionList,
                                     subscriptionCount,
                                     IOT_MQTT_FLAG_WAITABLE | MQTT_INTERNAL_FLAG_BLOCK_ON_SEND,
                                     NULL,
                                     &subscribeOperation );

    /* Wait for the SUBSCRIBE operation to complete. */
    if( status == IOT_MQTT_STATUS_PENDING )
    {
        status = IotMqtt_Wait( subscribeOperation, timeoutMs );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Ensure that a status was set. */
    IotMqtt_Assert( status != IOT_MQTT_STATUS_PENDING );

    return status;
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_UnsubscribeAsync( IotMqttConnection_t mqttConnection,
                                         const IotMqttSubscription_t * pSubscriptionList,
                                         size_t subscriptionCount,
                                         uint32_t flags,
                                         const IotMqttCallbackInfo_t * pCallbackInfo,
                                         IotMqttOperation_t * const pUnsubscribeOperation )
{
    IotMqttError_t status = _subscriptionCommonSetup( IOT_MQTT_UNSUBSCRIBE,
                                                      mqttConnection,
                                                      pSubscriptionList,
                                                      subscriptionCount,
                                                      flags,
                                                      pUnsubscribeOperation );

    if( IOT_MQTT_SUCCESS == status )
    {
        /* Remove the MQTT subscription list for an UNSUBSCRIBE. */
        _IotMqtt_RemoveSubscriptionByTopicFilter( mqttConnection,
                                                  pSubscriptionList,
                                                  subscriptionCount );

        status = _subscriptionCommon( IOT_MQTT_UNSUBSCRIBE,
                                      mqttConnection,
                                      _getMqttUnsubscribeSerializer( mqttConnection->pSerializer ),
                                      pSubscriptionList,
                                      subscriptionCount,
                                      flags,
                                      pCallbackInfo,
                                      pUnsubscribeOperation );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    return status;
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_UnsubscribeSync( IotMqttConnection_t mqttConnection,
                                        const IotMqttSubscription_t * pSubscriptionList,
                                        size_t subscriptionCount,
                                        uint32_t flags,
                                        uint32_t timeoutMs )
{
    IotMqttError_t status = IOT_MQTT_STATUS_PENDING;
    IotMqttOperation_t unsubscribeOperation = IOT_MQTT_OPERATION_INITIALIZER;

    /* Flags are not used, but the parameter is present for future compatibility. */
    ( void ) flags;

    /* Call the asynchronous UNSUBSCRIBE function. */
    status = IotMqtt_UnsubscribeAsync( mqttConnection,
                                       pSubscriptionList,
                                       subscriptionCount,
                                       IOT_MQTT_FLAG_WAITABLE | MQTT_INTERNAL_FLAG_BLOCK_ON_SEND,
                                       NULL,
                                       &unsubscribeOperation );

    /* Wait for the UNSUBSCRIBE operation to complete. */
    if( status == IOT_MQTT_STATUS_PENDING )
    {
        status = IotMqtt_Wait( unsubscribeOperation, timeoutMs );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Ensure that a status was set. */
    IotMqtt_Assert( status != IOT_MQTT_STATUS_PENDING );

    return status;
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_PublishAsync( IotMqttConnection_t mqttConnection,
                                     const IotMqttPublishInfo_t * pPublishInfo,
                                     uint32_t flags,
                                     const IotMqttCallbackInfo_t * pCallbackInfo,
                                     IotMqttOperation_t * const pPublishOperation )
{
    IOT_FUNCTION_ENTRY( IotMqttError_t, IOT_MQTT_SUCCESS );
    _mqttOperation_t * pOperation = NULL;
    uint8_t ** pPacketIdentifierHigh = NULL;

    /* Check that IotMqtt_Init was called. */
    if( _checkInit() == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NOT_INITIALIZED );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check that the PUBLISH information is valid. */
    if( _IotMqtt_ValidatePublish( mqttConnection->awsIotMqttMode,
                                  pPublishInfo ) == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check that no notification is requested for a QoS 0 publish. */
    if( pPublishInfo->qos == IOT_MQTT_QOS_0 )
    {
        if( pCallbackInfo != NULL )
        {
            IotLogError( "QoS 0 PUBLISH should not have notification parameters set." );

            IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
        }
        else if( ( flags & IOT_MQTT_FLAG_WAITABLE ) != 0 )
        {
            IotLogError( "QoS 0 PUBLISH should not have notification parameters set." );

            IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        if( pPublishOperation != NULL )
        {
            IotLogWarn( "Ignoring reference parameter for QoS 0 publish." );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check that a reference pointer is provided for a waitable operation. */
    if( ( flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE )
    {
        if( pPublishOperation == NULL )
        {
            IotLogError( "Reference must be provided for a waitable PUBLISH." );

            IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_PARAMETER );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Create a PUBLISH operation. */
    status = _IotMqtt_CreateOperation( mqttConnection,
                                       flags,
                                       pCallbackInfo,
                                       &pOperation );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check the PUBLISH operation data and set the operation type. */
    IotMqtt_Assert( pOperation->u.operation.status == IOT_MQTT_STATUS_PENDING );
    pOperation->u.operation.type = IOT_MQTT_PUBLISH_TO_SERVER;

    /* In AWS IoT MQTT mode, a pointer to the packet identifier must be saved. */
    if( mqttConnection->awsIotMqttMode == true )
    {
        pPacketIdentifierHigh = &( pOperation->u.operation.pPacketIdentifierHigh );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Generate a PUBLISH packet from pPublishInfo. */
    status = _getMqttPublishSerializer( mqttConnection->pSerializer )( pPublishInfo,
                                                                       &( pOperation->u.operation.pMqttPacket ),
                                                                       &( pOperation->u.operation.packetSize ),
                                                                       &( pOperation->u.operation.packetIdentifier ),
                                                                       pPacketIdentifierHigh );

    if( status != IOT_MQTT_SUCCESS )
    {
        IOT_GOTO_CLEANUP();
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check the serialized MQTT packet. */
    IotMqtt_Assert( pOperation->u.operation.pMqttPacket != NULL );
    IotMqtt_Assert( pOperation->u.operation.packetSize > 0 );

    /* Initialize PUBLISH retry if retryLimit is set. */
    if( pPublishInfo->retryLimit > 0 )
    {
        /* A QoS 0 PUBLISH may not be retried. */
        if( pPublishInfo->qos != IOT_MQTT_QOS_0 )
        {
            pOperation->u.operation.periodic.retry.limit = pPublishInfo->retryLimit;
            pOperation->u.operation.periodic.retry.nextPeriodMs = pPublishInfo->retryMs;
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Set the reference, if provided. */
    if( pPublishInfo->qos != IOT_MQTT_QOS_0 )
    {
        if( pPublishOperation != NULL )
        {
            *pPublishOperation = pOperation;
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Send the PUBLISH packet. */
    if( ( flags & MQTT_INTERNAL_FLAG_BLOCK_ON_SEND ) == MQTT_INTERNAL_FLAG_BLOCK_ON_SEND )
    {
        _IotMqtt_ProcessSend( IOT_SYSTEM_TASKPOOL, pOperation->job, pOperation );
    }
    else
    {
        status = _IotMqtt_ScheduleOperation( pOperation,
                                             _IotMqtt_ProcessSend,
                                             0 );

        if( status != IOT_MQTT_SUCCESS )
        {
            IotLogError( "(MQTT connection %p) Failed to enqueue PUBLISH for sending.",
                         mqttConnection );

            /* Clear the previously set (and now invalid) reference. */
            if( pPublishInfo->qos != IOT_MQTT_QOS_0 )
            {
                if( pPublishOperation != NULL )
                {
                    *pPublishOperation = IOT_MQTT_OPERATION_INITIALIZER;
                }
                else
                {
                    EMPTY_ELSE_MARKER;
                }
            }
            else
            {
                EMPTY_ELSE_MARKER;
            }

            IOT_GOTO_CLEANUP();
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }

    /* Clean up the PUBLISH operation if this function fails. Otherwise, set the
     * appropriate return code based on QoS. */
    IOT_FUNCTION_CLEANUP_BEGIN();

    if( status != IOT_MQTT_SUCCESS )
    {
        if( pOperation != NULL )
        {
            _IotMqtt_DestroyOperation( pOperation );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        if( pPublishInfo->qos > IOT_MQTT_QOS_0 )
        {
            status = IOT_MQTT_STATUS_PENDING;
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        IotLogInfo( "(MQTT connection %p) MQTT PUBLISH operation queued.",
                    mqttConnection );
    }

    IOT_FUNCTION_CLEANUP_END();
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_PublishSync( IotMqttConnection_t mqttConnection,
                                    const IotMqttPublishInfo_t * pPublishInfo,
                                    uint32_t flags,
                                    uint32_t timeoutMs )
{
    IotMqttError_t status = IOT_MQTT_STATUS_PENDING;
    IotMqttOperation_t publishOperation = IOT_MQTT_OPERATION_INITIALIZER,
                       * pPublishOperation = NULL;

    /* Clear the flags, setting only the "serial" flag. */
    flags = MQTT_INTERNAL_FLAG_BLOCK_ON_SEND;

    /* Set the waitable flag and reference for QoS 1 PUBLISH. */
    if( pPublishInfo->qos == IOT_MQTT_QOS_1 )
    {
        flags |= IOT_MQTT_FLAG_WAITABLE;
        pPublishOperation = &publishOperation;
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Call the asynchronous PUBLISH function. */
    status = IotMqtt_PublishAsync( mqttConnection,
                                   pPublishInfo,
                                   flags,
                                   NULL,
                                   pPublishOperation );

    /* Wait for a queued QoS 1 PUBLISH to complete. */
    if( pPublishInfo->qos == IOT_MQTT_QOS_1 )
    {
        if( status == IOT_MQTT_STATUS_PENDING )
        {
            status = IotMqtt_Wait( publishOperation, timeoutMs );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    return status;
}

/*-----------------------------------------------------------*/

IotMqttError_t IotMqtt_Wait( IotMqttOperation_t operation,
                             uint32_t timeoutMs )
{
    IotMqttError_t status = IOT_MQTT_SUCCESS;
    _mqttConnection_t * pMqttConnection = NULL;

    /* Check that IotMqtt_Init was called. */
    if( _checkInit() == false )
    {
        IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NOT_INITIALIZED );
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Validate the given operation reference. */
    if( _IotMqtt_ValidateOperation( operation ) == false )
    {
        status = IOT_MQTT_BAD_PARAMETER;
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    /* Check the MQTT connection status. */
    pMqttConnection = operation->pMqttConnection;

    if( status == IOT_MQTT_SUCCESS )
    {
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

        if( pMqttConnection->disconnected == true )
        {
            IotLogError( "(MQTT connection %p, %s operation %p) MQTT connection is closed. "
                         "Operation cannot be waited on.",
                         pMqttConnection,
                         IotMqtt_OperationType( operation->u.operation.type ),
                         operation );

            status = IOT_MQTT_NETWORK_ERROR;
        }
        else
        {
            IotLogInfo( "(MQTT connection %p, %s operation %p) Waiting for operation completion.",
                        pMqttConnection,
                        IotMqtt_OperationType( operation->u.operation.type ),
                        operation );
        }

        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

        /* Only wait on an operation if the MQTT connection is active. */
        if( status == IOT_MQTT_SUCCESS )
        {
            if( IotSemaphore_TimedWait( &( operation->u.operation.notify.waitSemaphore ),
                                        timeoutMs ) == false )
            {
                status = IOT_MQTT_TIMEOUT;

                /* Attempt to cancel the job of the timed out operation. */
                ( void ) _IotMqtt_DecrementOperationReferences( operation, true );

                /* Clean up lingering subscriptions from a timed-out SUBSCRIBE. */
                if( operation->u.operation.type == IOT_MQTT_SUBSCRIBE )
                {
                    IotLogDebug( "(MQTT connection %p, SUBSCRIBE operation %p) Cleaning up"
                                 " subscriptions of timed-out SUBSCRIBE.",
                                 pMqttConnection,
                                 operation );

                    _IotMqtt_RemoveSubscriptionByPacket( pMqttConnection,
                                                         operation->u.operation.packetIdentifier,
                                                         MQTT_REMOVE_ALL_SUBSCRIPTIONS );
                }
                else
                {
                    EMPTY_ELSE_MARKER;
                }
            }
            else
            {
                /* Retrieve the status of the completed operation. */
                status = operation->u.operation.status;
            }

            IotLogInfo( "(MQTT connection %p, %s operation %p) Wait complete with result %s.",
                        pMqttConnection,
                        IotMqtt_OperationType( operation->u.operation.type ),
                        operation,
                        IotMqtt_strerror( status ) );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }

        /* Wait is finished; decrement operation reference count. */
        if( _IotMqtt_DecrementOperationReferences( operation, false ) == true )
        {
            _IotMqtt_DestroyOperation( operation );
        }
        else
        {
            EMPTY_ELSE_MARKER;
        }
    }
    else
    {
        EMPTY_ELSE_MARKER;
    }

    IOT_FUNCTION_EXIT_NO_CLEANUP();
}

/*-----------------------------------------------------------*/

const char * IotMqtt_strerror( IotMqttError_t status )
{
    const char * pMessage = NULL;

    switch( status )
    {
        case IOT_MQTT_SUCCESS:
            pMessage = "SUCCESS";
            break;

        case IOT_MQTT_STATUS_PENDING:
            pMessage = "PENDING";
            break;

        case IOT_MQTT_INIT_FAILED:
            pMessage = "INITIALIZATION FAILED";
            break;

        case IOT_MQTT_BAD_PARAMETER:
            pMessage = "BAD PARAMETER";
            break;

        case IOT_MQTT_NO_MEMORY:
            pMessage = "NO MEMORY";
            break;

        case IOT_MQTT_NETWORK_ERROR:
            pMessage = "NETWORK ERROR";
            break;

        case IOT_MQTT_SCHEDULING_ERROR:
            pMessage = "SCHEDULING ERROR";
            break;

        case IOT_MQTT_BAD_RESPONSE:
            pMessage = "BAD RESPONSE RECEIVED";
            break;

        case IOT_MQTT_TIMEOUT:
            pMessage = "TIMEOUT";
            break;

        case IOT_MQTT_SERVER_REFUSED:
            pMessage = "SERVER REFUSED";
            break;

        case IOT_MQTT_RETRY_NO_RESPONSE:
            pMessage = "NO RESPONSE";
            break;

        case IOT_MQTT_NOT_INITIALIZED:
            pMessage = "NOT INITIALIZED";
            break;

        default:
            pMessage = "INVALID STATUS";
            break;
    }

    return pMessage;
}

/*-----------------------------------------------------------*/

const char * IotMqtt_OperationType( IotMqttOperationType_t operation )
{
    const char * pMessage = NULL;

    switch( operation )
    {
        case IOT_MQTT_CONNECT:
            pMessage = "CONNECT";
            break;

        case IOT_MQTT_PUBLISH_TO_SERVER:
            pMessage = "PUBLISH";
            break;

        case IOT_MQTT_PUBACK:
            pMessage = "PUBACK";
            break;

        case IOT_MQTT_SUBSCRIBE:
            pMessage = "SUBSCRIBE";
            break;

        case IOT_MQTT_UNSUBSCRIBE:
            pMessage = "UNSUBSCRIBE";
            break;

        case IOT_MQTT_PINGREQ:
            pMessage = "PINGREQ";
            break;

        case IOT_MQTT_DISCONNECT:
            pMessage = "DISCONNECT";
            break;

        default:
            pMessage = "INVALID OPERATION";
            break;
    }

    return pMessage;
}

/*-----------------------------------------------------------*/

/* Provide access to internal functions and variables if testing. */
#if IOT_BUILD_TESTS == 1
    #include "iot_test_access_mqtt_api.c"
#endif