/*
 * IoT MQTT V2.1.1
 * Copyright (C) 2020 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_operation.c
 * @brief Implements functions that process MQTT operations.
 */

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

/* Standard includes. */
#include <string.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"

/**
 * @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( IotMqttPublishSetDup_t,
                                  _getMqttPublishSetDupFunc,
                                  _IotMqtt_PublishSetDup,
                                  serialize.publishSetDup )
    SERIALIZER_OVERRIDE_SELECTOR( IotMqttFreePacket_t,
                                  _getMqttFreePacketFunc,
                                  _IotMqtt_FreePacket,
                                  freePacket )
#else /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */
    #define _getMqttFreePacketFunc( pSerializer )       _IotMqtt_FreePacket
    #define _getMqttPublishSetDupFunc( pSerializer )    _IotMqtt_PublishSetDup
#endif /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */
/** @endcond */

/*-----------------------------------------------------------*/

/**
 * @brief First parameter to #_mqttOperation_match.
 */
typedef struct _operationMatchParam
{
    IotMqttOperationType_t type;        /**< @brief The type of operation to look for. */
    const uint16_t * pPacketIdentifier; /**< @brief The packet identifier associated with the operation.
                                         * Set to `NULL` to ignore packet identifier. */
} _operationMatchParam_t;

/*-----------------------------------------------------------*/

/**
 * @brief Match an MQTT operation by type and packet identifier.
 *
 * @param[in] pOperationLink Pointer to the link member of an #_mqttOperation_t.
 * @param[in] pMatch Pointer to an #_operationMatchParam_t.
 *
 * @return `true` if the operation matches the parameters in `pArgument`; `false`
 * otherwise.
 */
static bool _mqttOperation_match( const IotLink_t * const pOperationLink,
                                  void * pMatch );

/**
 * @brief Check if an operation with retry has exceeded its retry limit.
 *
 * If a PUBLISH operation is available for retry, this function also sets any
 * necessary DUP flags.
 *
 * @param[in] pOperation The operation to check.
 *
 * @return `true` if the operation may be retried; `false` otherwise.
 */
static bool _checkRetryLimit( _mqttOperation_t * pOperation );

/**
 * @brief Schedule the next send of an operation with retry.
 *
 * @param[in] pOperation The operation to schedule.
 *
 * @return `true` if the reschedule succeeded; `false` otherwise.
 */
static bool _scheduleNextRetry( _mqttOperation_t * pOperation );

/**
 * @brief Schedule a callback for a completed MQTT operation.
 *
 * @param[in] pOperation The completed MQTT operation.
 *
 * @return `IOT_MQTT_SUCCESS` if the schedule was successful;
 * `IOT_MQTT_SCHEDULING_ERROR` otherwise.
 */
static IotMqttError_t _scheduleCallback( _mqttOperation_t * pOperation );

/**
 * @brief Complete a pending send operation.
 *
 * @param[in] pOperation The pending MQTT send operation.
 * @param[out] pDestroyOperation Whether the operation should be destroyed afterwards.
 *
 * @return `true` if the operation is awaiting a response from the network;
 * `false` if not.
 */
static bool _completePendingSend( _mqttOperation_t * pOperation,
                                  bool * pDestroyOperation );

/**
 * @brief Initialize newly created  MQTT operation.
 *
 * @param[in] pMqttConnection The MQTT connection associated with the operation.
 * @param[in] pOperation pointer to the new operation.
 * @param[in] flags Flags variable passed to a user-facing MQTT function.
 * @param[in] pCallbackInfo User-provided callback function and parameter.
 *
 *
 * @return #IOT_MQTT_SUCCESS or #IOT_MQTT_NO_MEMORY.
 */
static IotMqttError_t _initializeOperation( _mqttConnection_t * pMqttConnection,
                                            _mqttOperation_t * pOperation,
                                            uint32_t flags,
                                            const IotMqttCallbackInfo_t * pCallbackInfo );

/**
 * @brief Send MQTT Ping Request to the broker.
 *
 * @param[in] pMqttConnection The MQTT connection associated with the request.
 *
 * @return `true` if send is successful; `false` otherwise.
 */
static bool _sendPingRequest( _mqttConnection_t * pMqttConnection );

/*-----------------------------------------------------------*/

static bool _mqttOperation_match( const IotLink_t * const pOperationLink,
                                  void * pMatch )
{
    bool match = false;

    /* Because this function is called from a container function, the given link
     * must never be NULL. */
    IotMqtt_Assert( pOperationLink != NULL );

    /* Casting `pOperationLink` to uint8_t * is done only to calculate the
     * starting address of the struct and does not modify the link it points to.
     * Adding parentheses to parameters of IotLink_Container is not applicable
     * because it uses type-casting and offsetof, and would cause compiling errors. */
    /* coverity[misra_c_2012_rule_11_8_violation] */
    /* coverity[misra_c_2012_rule_20_7_violation] */
    /* coverity[caretline] */
    const _mqttOperation_t * pOperation = IotLink_Container( _mqttOperation_t,
                                                             pOperationLink,
                                                             link );
    const _operationMatchParam_t * pParam = ( _operationMatchParam_t * ) pMatch;

    /* Check for matching operations. */
    if( pParam->type == pOperation->u.operation.type )
    {
        /* Check for matching packet identifiers. */
        if( pParam->pPacketIdentifier == NULL )
        {
            match = true;
        }
        else
        {
            match = ( *( pParam->pPacketIdentifier ) == pOperation->u.operation.packetIdentifier );
        }
    }

    return match;
}

/*-----------------------------------------------------------*/

static bool _checkRetryLimit( _mqttOperation_t * pOperation )
{
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;
    bool status = true, setDup = false;

    /* Only PUBLISH may be retried. */
    IotMqtt_Assert( pOperation->u.operation.type == IOT_MQTT_PUBLISH_TO_SERVER );

    /* Check if the retry limit is exhausted. */
    if( pOperation->u.operation.periodic.retry.count > pOperation->u.operation.periodic.retry.limit )
    {
        /* The retry count may be at most one more than the retry limit, which
         * accounts for the final check for a PUBACK. */
        IotMqtt_Assert( pOperation->u.operation.periodic.retry.count ==
                        ( pOperation->u.operation.periodic.retry.limit + 1U ) );

        IotLogDebug( "(MQTT connection %p, PUBLISH operation %p) No response received after %lu retries.",
                     pMqttConnection,
                     pOperation,
                     pOperation->u.operation.periodic.retry.limit );

        status = false;
    }
    else
    {
        if( pOperation->u.operation.periodic.retry.count == 1U )
        {
            /* The DUP flag should always be set on the first retry. */
            setDup = true;
        }
        else if( pMqttConnection->awsIotMqttMode == true )
        {
            /* In AWS IoT MQTT mode, the DUP flag (really a change to the packet
             * identifier) must be reset on every retry. */
            setDup = true;
        }
        else
        {
            setDup = false;
        }

        if( setDup == true )
        {
            /* In AWS IoT MQTT mode, the references mutex must be locked to
             * prevent the packet identifier from being read while it is being
             * changed. */
            if( pMqttConnection->awsIotMqttMode == true )
            {
                IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
            }

            /* Set the DUP flag */
            _getMqttPublishSetDupFunc( pMqttConnection->pSerializer )( pOperation->u.operation.pMqttPacket,
                                                                       pOperation->u.operation.pPacketIdentifierHigh,
                                                                       &( pOperation->u.operation.packetIdentifier ) );

            if( pMqttConnection->awsIotMqttMode == true )
            {
                IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );
            }
        }
    }

    return status;
}

/*-----------------------------------------------------------*/

static bool _scheduleNextRetry( _mqttOperation_t * pOperation )
{
    bool firstRetry = false;
    uint32_t scheduleDelay = 0;
    IotMqttError_t status = IOT_MQTT_STATUS_PENDING;
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;

    /* This function should never be called with retry count greater than
     * retry limit. */
    IotMqtt_Assert( pOperation->u.operation.periodic.retry.count <=
                    pOperation->u.operation.periodic.retry.limit );

    /* Increment the retry count. */
    ( pOperation->u.operation.periodic.retry.count )++;

    /* Check for a response shortly for the final retry. Otherwise, calculate the
     * next retry period. */
    if( pOperation->u.operation.periodic.retry.count >
        pOperation->u.operation.periodic.retry.limit )
    {
        scheduleDelay = IOT_MQTT_RESPONSE_WAIT_MS;

        IotLogDebug( "(MQTT connection %p, PUBLISH operation %p) Final retry was sent. Will check "
                     "for response in %d ms.",
                     pMqttConnection,
                     pOperation,
                     IOT_MQTT_RESPONSE_WAIT_MS );
    }
    else
    {
        scheduleDelay = pOperation->u.operation.periodic.retry.nextPeriodMs;

        /* Double the retry period, subject to a ceiling value. */
        pOperation->u.operation.periodic.retry.nextPeriodMs *= 2U;

        if( pOperation->u.operation.periodic.retry.nextPeriodMs > IOT_MQTT_RETRY_MS_CEILING )
        {
            pOperation->u.operation.periodic.retry.nextPeriodMs = IOT_MQTT_RETRY_MS_CEILING;
        }

        /* In some implementations IotLogDebug() maps to C standard printing API
         * that need specific primitive types for format specifiers. Also
         * inttypes.h may not be available on some C99 compilers, despite
         * stdint.h being available. */
        /* coverity[misra_c_2012_directive_4_6_violation] */
        IotLogDebug( "(MQTT connection %p, PUBLISH operation %p) Scheduling retry %lu of %lu in %lu ms.",
                     pMqttConnection,
                     pOperation,
                     ( unsigned long ) pOperation->u.operation.periodic.retry.count,
                     ( unsigned long ) pOperation->u.operation.periodic.retry.limit,
                     ( unsigned long ) scheduleDelay );

        /* Check if this is the first retry. */
        firstRetry = ( pOperation->u.operation.periodic.retry.count == 1U );

        /* On the first retry, the PUBLISH will be moved from the pending processing
         * list to the pending responses list. Lock the connection references mutex
         * to manipulate the lists. */
        if( firstRetry == true )
        {
            IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
        }
    }

    /* Reschedule the PUBLISH for another send. */
    status = _IotMqtt_ScheduleOperation( pOperation,
                                         _IotMqtt_ProcessSend,
                                         scheduleDelay );

    /* Check for successful reschedule. */
    if( status == IOT_MQTT_SUCCESS )
    {
        /* Move a successfully rescheduled PUBLISH from the pending processing
         * list to the pending responses list on the first retry. */
        if( firstRetry == true )
        {
            if( IotLink_IsLinked( &( pOperation->link ) ) == true )
            {
                IotListDouble_Remove( &( pOperation->link ) );
            }

            IotListDouble_InsertHead( &( pMqttConnection->pendingResponse ),
                                      &( pOperation->link ) );
        }
    }

    /* The references mutex only needs to be unlocked on the first retry, since
     * only the first retry manipulates the connection lists. */
    if( firstRetry == true )
    {
        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );
    }

    return( status == IOT_MQTT_SUCCESS );
}

/*-----------------------------------------------------------*/

static IotMqttError_t _scheduleCallback( _mqttOperation_t * pOperation )
{
    IotMqttError_t status = IOT_MQTT_SCHEDULING_ERROR;
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;

    /* Non-waitable operation should have job reference of 1. */
    IotMqtt_Assert( pOperation->u.operation.jobReference == 1 );

    /* Schedule an invocation of the callback. */
    if( pOperation->u.operation.notify.callback.function != NULL )
    {
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

        status = _IotMqtt_ScheduleOperation( pOperation,
                                             _IotMqtt_ProcessCompletedOperation,
                                             0 );

        if( status == IOT_MQTT_SUCCESS )
        {
            IotLogDebug( "(MQTT connection %p, %s operation %p) Callback scheduled.",
                         pOperation->pMqttConnection,
                         IotMqtt_OperationType( pOperation->u.operation.type ),
                         pOperation );

            /* Place the scheduled operation back in the list of operations pending
             * processing. */
            if( IotLink_IsLinked( &( pOperation->link ) ) == true )
            {
                IotListDouble_Remove( &( pOperation->link ) );
            }

            IotListDouble_InsertHead( &( pMqttConnection->pendingProcessing ),
                                      &( pOperation->link ) );
        }
        else
        {
            IotLogWarn( "(MQTT connection %p, %s operation %p) Failed to schedule callback.",
                        pOperation->pMqttConnection,
                        IotMqtt_OperationType( pOperation->u.operation.type ),
                        pOperation );
        }

        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );
    }

    return status;
}

/*-----------------------------------------------------------*/

static bool _completePendingSend( _mqttOperation_t * pOperation,
                                  bool * pDestroyOperation )
{
    bool networkPending = false, waitable = false;
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;

    /* Check if this operation is waitable. */
    waitable = ( pOperation->u.operation.flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE;

    /* Check if this operation should be scheduled for retransmission. */
    if( pOperation->u.operation.periodic.retry.limit > 0U )
    {
        if( _scheduleNextRetry( pOperation ) == false )
        {
            pOperation->u.operation.status = IOT_MQTT_SCHEDULING_ERROR;
        }
        else
        {
            /* A successfully scheduled PUBLISH retry is awaiting a response
             * from the network. */
            networkPending = true;
        }
    }
    else
    {
        /* Decrement reference count to signal completion of send job. Check
         * if the operation should be destroyed. */
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

        if( waitable == true )
        {
            *pDestroyOperation = _IotMqtt_DecrementOperationReferences( pOperation, false );
        }

        /* If the operation should not be destroyed, transfer it from the
         * pending processing to the pending response list. */
        if( *pDestroyOperation == false )
        {
            if( IotLink_IsLinked( &( pOperation->link ) ) == true )
            {
                IotListDouble_Remove( &( pOperation->link ) );
            }

            IotListDouble_InsertHead( &( pMqttConnection->pendingResponse ),
                                      &( pOperation->link ) );

            /* This operation is now awaiting a response from the network. */
            networkPending = true;
        }

        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );
    }

    return networkPending;
}

/*-----------------------------------------------------------*/

static IotMqttError_t _initializeOperation( _mqttConnection_t * pMqttConnection,
                                            _mqttOperation_t * pOperation,
                                            uint32_t flags,
                                            const IotMqttCallbackInfo_t * pCallbackInfo )
{
    IotMqttError_t status = IOT_MQTT_SUCCESS;
    bool waitable = ( ( flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE );

    IotMqtt_Assert( pMqttConnection != NULL );
    IotMqtt_Assert( pOperation != NULL );

    /* Clear the operation data. */
    ( void ) memset( pOperation, 0x00, sizeof( _mqttOperation_t ) );

    /* Initialize some members of the new operation. */
    pOperation->pMqttConnection = pMqttConnection;
    pOperation->u.operation.jobReference = 1;
    pOperation->u.operation.flags = flags;
    pOperation->u.operation.status = IOT_MQTT_STATUS_PENDING;

    /* Check if the waitable flag is set. If it is, create a semaphore to
     * wait on. */
    if( waitable == true )
    {
        /* Create a semaphore to wait on for a waitable operation. */
        if( IotSemaphore_Create( &( pOperation->u.operation.notify.waitSemaphore ), 0, 1 ) == false )
        {
            IotLogError( "(MQTT connection %p) Failed to create semaphore for "
                         "waitable operation.",
                         pMqttConnection );

            status = IOT_MQTT_NO_MEMORY;
        }
        else
        {
            /* A waitable operation is created with an additional reference for the
             * Wait function. */
            ( pOperation->u.operation.jobReference )++;
        }
    }
    else
    {
        /* If the waitable flag isn't set but a callback is, copy the callback
         * information. */
        if( pCallbackInfo != NULL )
        {
            pOperation->u.operation.notify.callback = *pCallbackInfo;
        }
    }

    return status;
}

/*-----------------------------------------------------------*/

static bool _sendPingRequest( _mqttConnection_t * pMqttConnection )
{
    size_t bytesSent = 0;
    bool status = true;
    uint32_t swapStatus = 0;
    _mqttOperation_t * pPingreqOperation = NULL;

    IotMqtt_Assert( pMqttConnection != NULL );

    IotLogDebug( "(MQTT connection %p) Sending PINGREQ.", pMqttConnection );

    pPingreqOperation = &( pMqttConnection->pingreq );

    /* Because PINGREQ may be used to keep the MQTT connection alive, it is
     * more important than other operations. Bypass the queue of jobs for
     * operations by directly sending the PINGREQ in this job. */
    bytesSent = pMqttConnection->pNetworkInterface->send( pMqttConnection->pNetworkConnection,
                                                          pPingreqOperation->u.operation.pMqttPacket,
                                                          pPingreqOperation->u.operation.packetSize );

    if( bytesSent != pPingreqOperation->u.operation.packetSize )
    {
        IotLogError( "(MQTT connection %p) Failed to send PINGREQ.", pMqttConnection );
        status = false;
    }
    else
    {
        /* Update the timestamp of the last message on successful transmission. */
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
        pMqttConnection->lastMessageTime = IotClock_GetTimeMs();
        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

        /* Assume the keep-alive will fail. The network receive callback will
         * clear the failure flag upon receiving a PINGRESP. */
        swapStatus = Atomic_CompareAndSwap_u32( &( pPingreqOperation->u.operation.periodic.ping.failure ),
                                                1,
                                                0 );
        IotMqtt_Assert( swapStatus == 1U );

        /* Set the period for scheduling a PINGRESP check. */
        pPingreqOperation->u.operation.periodic.ping.nextPeriodMs = IOT_MQTT_RESPONSE_WAIT_MS;

        IotLogDebug( "(MQTT connection %p) PINGREQ sent.", pMqttConnection );
    }

    return status;
}

/*-----------------------------------------------------------*/

IotMqttError_t _IotMqtt_CreateOperation( _mqttConnection_t * pMqttConnection,
                                         uint32_t flags,
                                         const IotMqttCallbackInfo_t * pCallbackInfo,
                                         _mqttOperation_t ** pNewOperation )
{
    IotMqttError_t status = IOT_MQTT_SUCCESS;
    bool decrementOnError = false;
    _mqttOperation_t * pOperation = NULL;
    bool waitable = ( ( flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE );

    /* If the waitable flag is set, make sure that there's no callback. */
    if( waitable == true )
    {
        if( pCallbackInfo != NULL )
        {
            IotLogError( "Callback should not be set for a waitable operation." );

            status = IOT_MQTT_BAD_PARAMETER;
        }
    }

    if( status == IOT_MQTT_SUCCESS )
    {
        IotLogDebug( "(MQTT connection %p) Creating new operation record.",
                     pMqttConnection );

        /* Increment the reference count for the MQTT connection when creating a new
         * operation. */
        if( _IotMqtt_IncrementConnectionReferences( pMqttConnection ) == false )
        {
            IotLogError( "(MQTT connection %p) New operation record cannot be created"
                         " for a closed connection",
                         pMqttConnection );

            status = IOT_MQTT_NETWORK_ERROR;
        }
        else
        {
            /* Reference count will need to be decremented on error. */
            decrementOnError = true;
        }
    }

    if( status == IOT_MQTT_SUCCESS )
    {
        /* Allocate memory for a new operation. */
        pOperation = IotMqtt_MallocOperation( sizeof( _mqttOperation_t ) );

        if( pOperation == NULL )
        {
            IotLogError( "(MQTT connection %p) Failed to allocate memory for new "
                         "operation record.",
                         pMqttConnection );

            status = IOT_MQTT_NO_MEMORY;
        }
    }

    if( status == IOT_MQTT_SUCCESS )
    {
        status = _initializeOperation( pMqttConnection, pOperation, flags, pCallbackInfo );
    }

    if( status == IOT_MQTT_SUCCESS )
    {
        /* Add this operation to the MQTT connection's operation list. */
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
        IotListDouble_InsertHead( &( pMqttConnection->pendingProcessing ),
                                  &( pOperation->link ) );
        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

        /* Set the output parameter. */
        *pNewOperation = pOperation;
    }

    /* Clean up operation and decrement reference count if this function failed. */

    if( status != IOT_MQTT_SUCCESS )
    {
        if( decrementOnError == true )
        {
            _IotMqtt_DecrementConnectionReferences( pMqttConnection );
        }

        if( pOperation != NULL )
        {
            IotMqtt_FreeOperation( pOperation );
        }
    }

    return status;
}

/*-----------------------------------------------------------*/

bool _IotMqtt_DecrementOperationReferences( _mqttOperation_t * pOperation,
                                            bool cancelJob )
{
    bool destroyOperation = false;
    taskPoolError_t taskPoolStatus = TASKPOOL_SUCCESS;
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;

    /* Attempt to cancel the operation's job. */
    if( cancelJob == true )
    {
        /* Try to cancel the job only if it is initialised. */
        if( taskPoolIsJobInitialised( &( pOperation->job ) ) )
        {
            taskPoolStatus = taskPoolTryCancel( &( pOperation->job ) );
        }
        else
        {
            taskPoolStatus = TASKPOOL_GENERAL_FAILURE;
        }

        if( taskPoolStatus == TASKPOOL_SUCCESS )
        {
            IotLogDebug( "(MQTT connection %p, %s operation %p) Job canceled.",
                         pMqttConnection,
                         IotMqtt_OperationType( pOperation->u.operation.type ),
                         pOperation );
        }
    }

    /* Decrement job reference count. */
    if( taskPoolStatus == TASKPOOL_SUCCESS )
    {
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
        pOperation->u.operation.jobReference--;

        /* In some implementations IotLogDebug() maps to C standard printing API
         * that need specific primitive types for format specifiers. Also
         * inttypes.h may not be available on some C99 compilers, despite
         * stdint.h being available. */
        /* coverity[misra_c_2012_directive_4_6_violation] */
        IotLogDebug( "(MQTT connection %p, %s operation %p) Job reference changed"
                     " from %d to %d.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation,
                     ( int ) ( pOperation->u.operation.jobReference + 1 ),
                     ( int ) ( pOperation->u.operation.jobReference ) );

        /* The job reference count must be 0 or 1 after the decrement. */
        IotMqtt_Assert( ( pOperation->u.operation.jobReference == 0 ) ||
                        ( pOperation->u.operation.jobReference == 1 ) );

        /* This operation may be destroyed if its reference count is 0. */
        if( pOperation->u.operation.jobReference == 0 )
        {
            destroyOperation = true;
        }

        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );
    }

    return destroyOperation;
}

/*-----------------------------------------------------------*/

void _IotMqtt_DestroyOperation( _mqttOperation_t * pOperation )
{
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;

    IotLogDebug( "(MQTT connection %p, %s operation %p) Destroying operation.",
                 pMqttConnection,
                 IotMqtt_OperationType( pOperation->u.operation.type ),
                 pOperation );

    /* The job reference count must be between 0 and 2. */
    IotMqtt_Assert( ( pOperation->u.operation.jobReference >= 0 ) &&
                    ( pOperation->u.operation.jobReference <= 2 ) );

    /* Jobs to be destroyed should be removed from the MQTT connection's
     * lists. */
    IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

    if( IotLink_IsLinked( &( pOperation->link ) ) == true )
    {
        IotLogDebug( "(MQTT connection %p, %s operation %p) Removed operation from connection lists.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation,
                     pMqttConnection );

        IotListDouble_Remove( &( pOperation->link ) );
    }
    else
    {
        IotLogDebug( "(MQTT connection %p, %s operation %p) Operation was not present in connection lists.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation );
    }

    IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

    /* Free any allocated MQTT packet. */
    if( pOperation->u.operation.pMqttPacket != NULL )
    {
        _getMqttFreePacketFunc( pMqttConnection->pSerializer )( pOperation->u.operation.pMqttPacket );

        IotLogDebug( "(MQTT connection %p, %s operation %p) MQTT packet freed.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation );
    }
    else
    {
        IotLogDebug( "(MQTT connection %p, %s operation %p) Operation has no allocated MQTT packet.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation );
    }

    /* Check if a wait semaphore was created for this operation. */
    if( ( pOperation->u.operation.flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE )
    {
        IotSemaphore_Destroy( &( pOperation->u.operation.notify.waitSemaphore ) );

        IotLogDebug( "(MQTT connection %p, %s operation %p) Wait semaphore destroyed.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation );
    }

    IotLogDebug( "(MQTT connection %p, %s operation %p) Operation record destroyed.",
                 pMqttConnection,
                 IotMqtt_OperationType( pOperation->u.operation.type ),
                 pOperation );

    /* Free the memory used to hold operation data. */
    IotMqtt_FreeOperation( pOperation );

    /* Decrement the MQTT connection's reference count after destroying an
     * operation. */
    _IotMqtt_DecrementConnectionReferences( pMqttConnection );
}

/*-----------------------------------------------------------*/

void _IotMqtt_ProcessKeepAlive( taskPoolJob_t * pKeepAliveJob,
                                void * pContext )
{
    bool status = true;
    taskPoolError_t taskPoolStatus = TASKPOOL_SUCCESS;
    uint32_t scheduleDelay = 0;
    uint64_t elapsedTime = 0;

    /* Retrieve the MQTT connection from the context. */
    _mqttConnection_t * pMqttConnection = ( _mqttConnection_t * ) pContext;
    _mqttOperation_t * pPingreqOperation = &( pMqttConnection->pingreq );

    /* Check parameters. */
    IotMqtt_Assert( pKeepAliveJob == &( pPingreqOperation->job ) );

    /* Check that keep-alive interval is valid. The MQTT spec states its maximum
     * value is 65,535 seconds. */
    IotMqtt_Assert( pPingreqOperation->u.operation.periodic.ping.keepAliveMs <= 65535000U );

    /* Only two values are valid for the next keep alive job delay. */
    IotMqtt_Assert( ( pPingreqOperation->u.operation.periodic.ping.nextPeriodMs ==
                      pPingreqOperation->u.operation.periodic.ping.keepAliveMs ) ||
                    ( pPingreqOperation->u.operation.periodic.ping.nextPeriodMs
                      == IOT_MQTT_RESPONSE_WAIT_MS ) );

    IotLogDebug( "(MQTT connection %p) Keep-alive job started.", pMqttConnection );

    /* Determine whether to send a PINGREQ or check for PINGRESP. */
    if( pPingreqOperation->u.operation.periodic.ping.nextPeriodMs ==
        pPingreqOperation->u.operation.periodic.ping.keepAliveMs )
    {
        /* Only send the PINGREQ if the keep-alive period has elapsed since the connection
         * was last used. */
        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
        elapsedTime = IotClock_GetTimeMs() - pMqttConnection->lastMessageTime;
        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

        if( elapsedTime < ( uint64_t ) pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs )
        {
            /* In some implementations IotLogDebug() maps to C standard printing API
             * that need specific primitive types for format specifiers. Also
             * inttypes.h may not be available on some C99 compilers, despite
             * stdint.h being available. */
            /* coverity[misra_c_2012_directive_4_6_violation] */
            IotLogDebug( "(MQTT connection %p) Connection was last used %llu ms ago, which "
                         "is less than keep-alive period %lu ms. PINGREQ will not be sent.",
                         pMqttConnection,
                         ( unsigned long long ) elapsedTime,
                         ( unsigned long ) pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs );

            /* Schedule the next keep-alive job one keep-alive period after the last packet was sent. */
            scheduleDelay = pPingreqOperation->u.operation.periodic.ping.keepAliveMs - ( ( uint32_t ) elapsedTime );
        }
        else
        {
            status = _sendPingRequest( pMqttConnection );
        }
    }
    else
    {
        IotLogDebug( "(MQTT connection %p) Checking for PINGRESP.", pMqttConnection );

        if( Atomic_Add_u32( &( pPingreqOperation->u.operation.periodic.ping.failure ), 0U ) == 0U )
        {
            IotLogDebug( "(MQTT connection %p) PINGRESP was received.", pMqttConnection );

            /* PINGRESP was received. Schedule the next PINGREQ transmission. */

            /* This function is called for two purposes:
             *
             * 1. To send a PINGREQ.
             * 2. To check that the corresponding PINGRESP is received within
             * IOT_MQTT_RESPONSE_WAIT_MS.
             *
             * The way it differentiates between the two is by checking
             * ping.nextPeriodMs in pPingreqOperation->u.operation.periodic:
             *
             * If ping.nextPeriodMs is set to ping.keepAliveMs,
             * the invocation is for sending PINGREQ.
             * Otherwise, the invocation is for checking that PINGRESP is received
             * within IOT_MQTT_RESPONSE_WAIT_MS.
             *
             * Therefore, it is necessary to set ping.nextPeriodMs
             * to ping.keepAliveMs to ensure that PINGREQ is sent in
             * the next invocation. But we must ensure that the next time to send
             * PINGREQ is calculated from the moment last PINGREQ was sent and NOT
             * when we checked for PINGRESP. As a result we need to schedule the next
             * invocation at ping.keepAliveMs - IOT_MQTT_RESPONSE_WAIT_MS.
             * The following diagram also explains it:
             *
             *      WaitMS    KeepAliveMS - WaitMS
             *    <-------->|<-------------------->
             *    ---------------------------------
             *    ^         ^                     ^
             *    |         |                     |
             * PINGREQ   PINGRESP              PINGREQ
             * (Call 1)  (Call 2)             (Call 3)
             *    <------------------------------->
             *                 KeepAliveMS
             * WaitMS = IOT_MQTT_RESPONSE_WAIT_MS.
             * KeepAliveMS = pPingreqOperation->u.operation.periodic.ping.keepAliveMs.
             * Call 1 - First PINGREQ is sent.
             * Call 2 - PINGRESP is checked after IOT_MQTT_RESPONSE_WAIT_MS.
             * Call 3 - Next PINGREQ is sent. Time difference between Call 2 and
             * Call 3 is KeepAliveMS - WaitMS, while time difference between Call 1
             * and Call 3 is KeepAliveMS. */
            pPingreqOperation->u.operation.periodic.ping.nextPeriodMs =
                pPingreqOperation->u.operation.periodic.ping.keepAliveMs;

            IotMqtt_Assert( pPingreqOperation->u.operation.periodic.ping.keepAliveMs
                            > IOT_MQTT_RESPONSE_WAIT_MS );

            /* Subtract time taken for PINGRESP check. */
            scheduleDelay = pPingreqOperation->u.operation.periodic.ping.keepAliveMs
                            - IOT_MQTT_RESPONSE_WAIT_MS;
        }
        else
        {
            IotLogError( "(MQTT connection %p) Failed to receive PINGRESP within %d ms.",
                         pMqttConnection,
                         IOT_MQTT_RESPONSE_WAIT_MS );

            /* The network receive callback did not clear the failure flag. */
            status = false;
        }
    }

    /* Reschedule this job. When a PINGREQ is sent, schedule a check for PINGRESP.
     * When PINGREQ is not sent (because the connection was recently used) schedule
     * another PINGREQ after the keep-alive period. */
    if( status == true )
    {
        if( scheduleDelay == 0 )
        {
            scheduleDelay = pPingreqOperation->u.operation.periodic.ping.nextPeriodMs;
        }

        IotMutex_Lock( &( pMqttConnection->referencesMutex ) );

        /* Re-create the keep-alive job for rescheduling. This should never fail. */
        taskPoolInitializeJob( _IotMqtt_ProcessKeepAlive,
                               pContext,
                               pKeepAliveJob );

        IotMqtt_Assert( scheduleDelay > 0 );
        taskPoolStatus = taskPoolScheduleDeferred( pKeepAliveJob,
                                                   scheduleDelay );

        if( taskPoolStatus == TASKPOOL_SUCCESS )
        {
            /* In some implementations IotLogDebug() maps to a C standard printing API
             * that need specific primitive types for format specifiers. Also,
             * inttypes.h may not be available on some C99 compilers, despite
             * stdint.h being available. */
            /* coverity[misra_c_2012_directive_4_6_violation] */
            IotLogDebug( "(MQTT connection %p) Next keep-alive job in %lu ms.",
                         pMqttConnection,
                         ( unsigned long ) scheduleDelay );
        }
        else
        {
            IotLogError( "(MQTT connection %p) Failed to reschedule keep-alive job, error %s.",
                         pMqttConnection,
                         taskPoolStrError( taskPoolStatus ) );

            status = false;
        }

        IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );
    }

    /* Close the connection on failures. */
    if( status == false )
    {
        _IotMqtt_CloseNetworkConnection( IOT_MQTT_KEEP_ALIVE_TIMEOUT,
                                         pMqttConnection );

        /* Keep-alive has failed and will no longer use this MQTT connection. */
        _IotMqtt_DecrementConnectionReferences( pMqttConnection );
    }
}

/*-----------------------------------------------------------*/

void _IotMqtt_ProcessIncomingPublish( taskPoolJob_t * pPublishJob,
                                      void * pContext )
{
    _mqttOperation_t * pOperation = pContext;
    IotMqttCallbackParam_t callbackParam = { .mqttConnection = NULL };

    /* Check parameters. The task pool and job parameter is not used when asserts
     * are disabled. */
    ( void ) pPublishJob;
    IotMqtt_Assert( pOperation->incomingPublish == true );
    IotMqtt_Assert( pPublishJob == &( pOperation->job ) );

    /* Remove the operation from the pending processing list. */
    IotMutex_Lock( &( pOperation->pMqttConnection->referencesMutex ) );

    if( IotLink_IsLinked( &( pOperation->link ) ) == true )
    {
        IotListDouble_Remove( &( pOperation->link ) );
    }

    IotMutex_Unlock( &( pOperation->pMqttConnection->referencesMutex ) );

    /* Process the current PUBLISH. */
    callbackParam.u.message.info = pOperation->u.publish.publishInfo;

    _IotMqtt_InvokeSubscriptionCallback( pOperation->pMqttConnection,
                                         &callbackParam );

    /* Free buffers associated with the current PUBLISH message. */
    IotMqtt_Assert( pOperation->u.publish.pReceivedData != NULL );
    IotMqtt_FreeMessage( pOperation->u.publish.pReceivedData );

    /* Free the incoming PUBLISH operation. */
    IotMqtt_FreeOperation( pOperation );
}

/*-----------------------------------------------------------*/

void _IotMqtt_ProcessSend( taskPoolJob_t * pSendJob,
                           void * pContext )
{
    size_t bytesSent = 0;
    bool destroyOperation = false, waitable = false, networkPending = false;
    _mqttOperation_t * pOperation = ( _mqttOperation_t * ) pContext;
    _mqttConnection_t * pMqttConnection = pOperation->pMqttConnection;

    /* Check parameters. The task pool and job parameter is not used when asserts
     * are disabled. */
    ( void ) pSendJob;
    IotMqtt_Assert( pSendJob == &( pOperation->job ) );

    /* The given operation must have an allocated packet and be waiting for a status. */
    IotMqtt_Assert( pOperation->u.operation.pMqttPacket != NULL );
    IotMqtt_Assert( pOperation->u.operation.packetSize != 0U );
    IotMqtt_Assert( pOperation->u.operation.status == IOT_MQTT_STATUS_PENDING );

    /* Check if this operation is waitable. */
    waitable = ( pOperation->u.operation.flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE;

    /* Check PUBLISH retry counts and limits. */
    if( pOperation->u.operation.periodic.retry.limit > 0U )
    {
        if( _checkRetryLimit( pOperation ) == false )
        {
            pOperation->u.operation.status = IOT_MQTT_RETRY_NO_RESPONSE;
        }
    }

    /* Send an operation that is waiting for a response. */
    if( pOperation->u.operation.status == IOT_MQTT_STATUS_PENDING )
    {
        IotLogDebug( "(MQTT connection %p, %s operation %p) Sending MQTT packet.",
                     pMqttConnection,
                     IotMqtt_OperationType( pOperation->u.operation.type ),
                     pOperation );

        /* Transmit the MQTT packet from the operation over the network. */
        bytesSent = pMqttConnection->pNetworkInterface->send( pMqttConnection->pNetworkConnection,
                                                              pOperation->u.operation.pMqttPacket,
                                                              pOperation->u.operation.packetSize );

        /* Check transmission status. */
        if( bytesSent != pOperation->u.operation.packetSize )
        {
            pOperation->u.operation.status = IOT_MQTT_NETWORK_ERROR;
        }
        else
        {
            /* Update the timestamp of the last message on successful transmission. */
            IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
            pMqttConnection->lastMessageTime = IotClock_GetTimeMs();
            IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

            /* DISCONNECT operations are considered successful upon successful transmission. */
            if( pOperation->u.operation.type == IOT_MQTT_DISCONNECT )
            {
                /* DISCONNECT operations are always waitable. */
                IotMqtt_Assert( waitable == true );

                pOperation->u.operation.status = IOT_MQTT_SUCCESS;
            }
            /* Non-waitable operations with no callback are also considered successful. */
            else if( waitable == false )
            {
                if( pOperation->u.operation.notify.callback.function == NULL )
                {
                    pOperation->u.operation.status = IOT_MQTT_SUCCESS;
                }
            }
            else
            {
                /* Empty else MISRA 15.7 */
            }
        }
    }

    /* Check if this operation requires further processing. */
    if( pOperation->u.operation.status == IOT_MQTT_STATUS_PENDING )
    {
        networkPending = _completePendingSend( pOperation, &destroyOperation );
    }

    /* Destroy the operation or notify of completion if necessary. */
    if( destroyOperation == true )
    {
        _IotMqtt_DestroyOperation( pOperation );
    }
    else
    {
        /* Do not check the operation status if a network response is pending,
         * since a network response could modify the status. */
        if( networkPending == false )
        {
            /* Operations that are not waiting for a network response either failed or
             * completed successfully. Check that a status was set. */
            IotMqtt_Assert( pOperation->u.operation.status != IOT_MQTT_STATUS_PENDING );

            /* Notify of operation completion if this job set a status. */
            _IotMqtt_Notify( pOperation );
        }
    }
}

/*-----------------------------------------------------------*/

void _IotMqtt_ProcessCompletedOperation( taskPoolJob_t * pOperationJob,
                                         void * pContext )
{
    bool destroyOperation = false;
    _mqttOperation_t * pOperation = ( _mqttOperation_t * ) pContext;
    IotMqttCallbackParam_t callbackParam = { 0 };

    /* Check parameters. The task pool and job parameter is not used when asserts
     * are disabled. */
    ( void ) pOperationJob;
    IotMqtt_Assert( pOperationJob == &( pOperation->job ) );

    /* The operation's callback function and status must be set. */
    IotMqtt_Assert( pOperation->u.operation.notify.callback.function != NULL );
    IotMqtt_Assert( pOperation->u.operation.status != IOT_MQTT_STATUS_PENDING );

    callbackParam.mqttConnection = pOperation->pMqttConnection;
    callbackParam.u.operation.type = pOperation->u.operation.type;
    callbackParam.u.operation.reference = pOperation;
    callbackParam.u.operation.result = pOperation->u.operation.status;

    /* Invoke the user callback function. */
    pOperation->u.operation.notify.callback.function( pOperation->u.operation.notify.callback.pCallbackContext,
                                                      &callbackParam );

    /* Decrement the operation reference count. This function is at the end of the
     * operation lifecycle, so the operation must be destroyed here. */
    destroyOperation = _IotMqtt_DecrementOperationReferences( pOperation, false );
    IotMqtt_Assert( destroyOperation == true );
    _IotMqtt_DestroyOperation( pOperation );
}

/*-----------------------------------------------------------*/

IotMqttError_t _IotMqtt_ScheduleOperation( _mqttOperation_t * pOperation,
                                           taskPoolRoutine_t jobRoutine,
                                           uint32_t delay )
{
    IotMqttError_t status = IOT_MQTT_SUCCESS;
    taskPoolError_t taskPoolStatus = TASKPOOL_SUCCESS;

    /* Check that job routine is valid. */
    IotMqtt_Assert( ( jobRoutine == _IotMqtt_ProcessSend ) ||
                    ( jobRoutine == _IotMqtt_ProcessCompletedOperation ) ||
                    ( jobRoutine == _IotMqtt_ProcessIncomingPublish ) );

    /* Creating a new job should never fail when parameters are valid. */
    taskPoolInitializeJob( jobRoutine,
                           pOperation,
                           &( pOperation->job ) );

    /* Schedule the new job with a delay. */
    taskPoolStatus = taskPoolScheduleDeferred( &( pOperation->job ),
                                               delay );

    if( taskPoolStatus != TASKPOOL_SUCCESS )
    {
        /* Scheduling a newly-created job should never be invalid or illegal. */
        IotMqtt_Assert( taskPoolStatus != TASKPOOL_ILLEGAL_OPERATION );

        /* Coverity finds a MISRA 13.2 violation in this log statement as the order
         * of evaluation for IotMqtt_OperationType and taskPoolStrError is not
         * defined. This is not an issue as these functions do not change data and
         * only convert codes into constant strings. */
        /* coverity[misra_c_2012_rule_13_2_violation] */
        IotLogWarn( "(MQTT connection %p, %s operation %p) Failed to schedule operation job, error %s.",
                    pOperation->pMqttConnection,
                    IotMqtt_OperationType( pOperation->u.operation.type ),
                    pOperation,
                    taskPoolStrError( taskPoolStatus ) );

        status = IOT_MQTT_SCHEDULING_ERROR;
    }

    return status;
}

/*-----------------------------------------------------------*/

_mqttOperation_t * _IotMqtt_FindOperation( _mqttConnection_t * pMqttConnection,
                                           IotMqttOperationType_t type,
                                           const uint16_t * pPacketIdentifier )
{
    bool waitable = false;
    taskPoolError_t taskPoolStatus = TASKPOOL_SUCCESS;
    _mqttOperation_t * pResult = NULL;
    IotLink_t * pResultLink = NULL;
    _operationMatchParam_t operationMatchParams;

    ( void ) memset( &operationMatchParams, 0, sizeof( _operationMatchParam_t ) );

    /* Set the members of the search parameter. */
    operationMatchParams.type = type;
    operationMatchParams.pPacketIdentifier = pPacketIdentifier;

    IotLogDebug( "(MQTT connection %p) Searching for operation %s "
                 "with packet identifier %hu.",
                 pMqttConnection,
                 IotMqtt_OperationType( type ),
                 ( pPacketIdentifier == NULL ) ? 0U : *pPacketIdentifier );

    /* Find and remove the first matching element in the list. */
    IotMutex_Lock( &( pMqttConnection->referencesMutex ) );
    pResultLink = IotListDouble_FindFirstMatch( &( pMqttConnection->pendingResponse ),
                                                NULL,
                                                _mqttOperation_match,
                                                &operationMatchParams );

    /* Check if a match was found. */
    if( pResultLink != NULL )
    {
        /* Get operation pointer and check if it is waitable. */

        /* Adding parentheses to parameters of IotLink_Container is not applicable
         * because it uses type-casting and offsetof, and would cause compiling errors. */
        /* coverity[misra_c_2012_rule_20_7_violation] */
        /* coverity[caretline] */
        pResult = IotLink_Container( _mqttOperation_t, pResultLink, link );
        waitable = ( pResult->u.operation.flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE;

        /* Check if the matched operation is a PUBLISH with retry. If it is, cancel
         * the retry job. */
        if( pResult->u.operation.periodic.retry.limit > 0U )
        {
            /* Try to cancel the job only if it is initialised. */
            if( taskPoolIsJobInitialised( &( pResult->job ) ) )
            {
                taskPoolStatus = taskPoolTryCancel( &( pResult->job ) );
            }
            else
            {
                taskPoolStatus = TASKPOOL_GENERAL_FAILURE;
            }

            /* If the retry job could not be canceled, then it is currently
             * executing. Ignore the operation. */
            if( taskPoolStatus != TASKPOOL_SUCCESS )
            {
                pResult = NULL;
            }
        }
        else
        {
            /* An operation with no retry in the pending responses list should
             * always have a job reference of 1. */
            IotMqtt_Assert( pResult->u.operation.jobReference == 1 );

            /* Increment job references of a waitable operation to prevent Wait from
             * destroying this operation if it times out. */
            if( waitable == true )
            {
                ( pResult->u.operation.jobReference )++;

                /* In some implementations IotLogDebug() maps to C standard printing API
                 * that need specific primitive types for format specifiers. Also
                 * inttypes.h may not be available on some C99 compilers, despite
                 * stdint.h being available. */
                /* coverity[misra_c_2012_directive_4_6_violation] */
                IotLogDebug( "(MQTT connection %p, %s operation %p) Job reference changed from %d to %d.",
                             pMqttConnection,
                             IotMqtt_OperationType( type ),
                             pResult,
                             ( int ) ( pResult->u.operation.jobReference - 1 ),
                             ( int ) ( pResult->u.operation.jobReference ) );
            }
        }
    }

    if( pResult != NULL )
    {
        IotLogDebug( "(MQTT connection %p) Found operation %s.",
                     pMqttConnection,
                     IotMqtt_OperationType( type ) );

        /* Remove the matched operation from the list. */
        IotListDouble_Remove( &( pResult->link ) );
    }
    else
    {
        IotLogDebug( "(MQTT connection %p) Operation %s not found.",
                     pMqttConnection,
                     IotMqtt_OperationType( type ) );
    }

    IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );

    return pResult;
}

/*-----------------------------------------------------------*/

void _IotMqtt_Notify( _mqttOperation_t * pOperation )
{
    IotMqttError_t status = IOT_MQTT_SCHEDULING_ERROR;

    /* Check if operation is waitable. */
    bool waitable = ( pOperation->u.operation.flags & IOT_MQTT_FLAG_WAITABLE ) == IOT_MQTT_FLAG_WAITABLE;

    /* Remove any lingering subscriptions if a SUBSCRIBE failed. Rejected
     * subscriptions are removed by the deserializer, so not removed here. */
    if( pOperation->u.operation.type == IOT_MQTT_SUBSCRIBE )
    {
        switch( pOperation->u.operation.status )
        {
            case IOT_MQTT_SUCCESS:
                break;

            case IOT_MQTT_SERVER_REFUSED:
                break;

            default:
                _IotMqtt_RemoveSubscriptionByPacket( pOperation->pMqttConnection,
                                                     pOperation->u.operation.packetIdentifier,
                                                     -1 );
                break;
        }
    }

    /* Schedule callback invocation for non-waitable operation. */
    if( waitable == false )
    {
        status = _scheduleCallback( pOperation );
    }

    /* Operations that weren't scheduled may be destroyed. */
    if( status == IOT_MQTT_SCHEDULING_ERROR )
    {
        /* Decrement reference count of operations not scheduled. */
        if( _IotMqtt_DecrementOperationReferences( pOperation, false ) == true )
        {
            _IotMqtt_DestroyOperation( pOperation );
        }
        else
        {
            /* Only waitable operations will have a reference count greater than 1.
             * Non-waitable operations will not reach this block. */
            IotMqtt_Assert( waitable == true );

            /* Post to a waitable operation's semaphore. */
            IotLogDebug( "(MQTT connection %p, %s operation %p) Waitable operation "
                         "notified of completion.",
                         pOperation->pMqttConnection,
                         IotMqtt_OperationType( pOperation->u.operation.type ),
                         pOperation );

            IotSemaphore_Post( &( pOperation->u.operation.notify.waitSemaphore ) );
        }
    }
    else
    {
        IotMqtt_Assert( status == IOT_MQTT_SUCCESS );
    }
}

/*-----------------------------------------------------------*/
