path: root/FreeRTOS-Labs/Source/FreeRTOS-IoT-Libraries/c_sdk/aws/jobs/src/aws_iot_jobs_subscription.c

/*
 * AWS IoT Jobs V1.0.0
 * Copyright (C) 2019 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 aws_iot_jobs_subscription.c
 * @brief Implements functions for interacting with the Jobs library's
 * subscription list.
 */

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

/* Standard includes. */
#include <string.h>

/* Jobs internal include. */
#include "private/aws_iot_jobs_internal.h"

/* Error handling include. */
#include "iot_error.h"

/* Platform layer includes. */
#include "platform/iot_threads.h"

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

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

/**
 * @brief Match two #_jobsSubscription_t by Thing Name.
 *
 * @param[in] pSubscriptionLink Pointer to the link member of a #_jobsSubscription_t
 * containing the Thing Name to check.
 * @param[in] pMatch Pointer to a `AwsIotThingName_t`.
 *
 * @return `true` if the Thing Names match; `false` otherwise.
 */
static bool _jobsSubscription_match( const IotLink_t * pSubscriptionLink,
                                     void * pMatch );

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

/**
 * @brief List of active Jobs subscriptions objects.
 */
IotListDouble_t _AwsIotJobsSubscriptions = { 0 };

/**
 * @brief Protects #_AwsIotJobsSubscriptions from concurrent access.
 */
IotMutex_t _AwsIotJobsSubscriptionsMutex;

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

static bool _jobsSubscription_match( const IotLink_t * pSubscriptionLink,
                                     void * pMatch )
{
    bool match = false;

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

    const _jobsSubscription_t * pSubscription = IotLink_Container( _jobsSubscription_t,
                                                                   pSubscriptionLink,
                                                                   link );
    const AwsIotThingName_t * pThingName = ( AwsIotThingName_t * ) pMatch;

    if( pThingName->thingNameLength == pSubscription->thingNameLength )
    {
        /* Check for matching Thing Names. */
        match = ( strncmp( pThingName->pThingName,
                           pSubscription->pThingName,
                           pThingName->thingNameLength ) == 0 );
    }

    return match;
}

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

_jobsSubscription_t * _AwsIotJobs_FindSubscription( const char * pThingName,
                                                    size_t thingNameLength,
                                                    bool createIfNotFound )
{
    _jobsSubscription_t * pSubscription = NULL;
    IotLink_t * pSubscriptionLink = NULL;
    AwsIotThingName_t thingName = { 0 };

    thingName.pThingName = pThingName;
    thingName.thingNameLength = thingNameLength;

    /* Search the list for an existing subscription for Thing Name. */
    pSubscriptionLink = IotListDouble_FindFirstMatch( &( _AwsIotJobsSubscriptions ),
                                                      NULL,
                                                      _jobsSubscription_match,
                                                      &thingName );

    /* Check if a subscription was found. */
    if( pSubscriptionLink == NULL )
    {
        if( createIfNotFound == true )
        {
            /* No subscription found. Allocate a new subscription. */
            pSubscription = AwsIotJobs_MallocSubscription( sizeof( _jobsSubscription_t ) + thingNameLength );

            if( pSubscription != NULL )
            {
                /* Clear the new subscription. */
                ( void ) memset( pSubscription, 0x00, sizeof( _jobsSubscription_t ) + thingNameLength );

                /* Set the Thing Name length and copy the Thing Name into the new subscription. */
                pSubscription->thingNameLength = thingNameLength;
                ( void ) memcpy( pSubscription->pThingName, pThingName, thingNameLength );

                /* Add the new subscription to the subscription list. */
                IotListDouble_InsertHead( &( _AwsIotJobsSubscriptions ),
                                          &( pSubscription->link ) );

                IotLogDebug( "Created new Jobs subscriptions object for %.*s.",
                             thingNameLength,
                             pThingName );
            }
            else
            {
                IotLogError( "Failed to allocate memory for %.*s Jobs subscriptions.",
                             thingNameLength,
                             pThingName );
            }
        }
    }
    else
    {
        IotLogDebug( "Found existing Jobs subscriptions object for %.*s.",
                     thingNameLength,
                     pThingName );

        pSubscription = IotLink_Container( _jobsSubscription_t, pSubscriptionLink, link );
    }

    return pSubscription;
}

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

void _AwsIotJobs_RemoveSubscription( _jobsSubscription_t * pSubscription,
                                     _jobsSubscription_t ** pRemovedSubscription )
{
    IOT_FUNCTION_ENTRY( bool, true );
    int32_t i = 0, callbackIndex = 0;

    IotLogDebug( "Checking if subscription object for %.*s can be removed.",
                 pSubscription->thingNameLength,
                 pSubscription->pThingName );

    /* Check for active operations. If any Jobs operation's subscription
     * reference count is not 0, then the subscription cannot be removed. */
    for( i = 0; i < JOBS_OPERATION_COUNT; i++ )
    {
        if( pSubscription->operationReferences[ i ] > 0 )
        {
            IotLogDebug( "Reference count %ld for %.*s subscription object. "
                         "Subscription cannot be removed yet.",
                         ( long int ) pSubscription->operationReferences[ i ],
                         pSubscription->thingNameLength,
                         pSubscription->pThingName );

            IOT_SET_AND_GOTO_CLEANUP( false );
        }
        else if( pSubscription->operationReferences[ i ] == AWS_IOT_PERSISTENT_SUBSCRIPTION )
        {
            IotLogDebug( "Subscription object for %.*s has persistent subscriptions. "
                         "Subscription will not be removed.",
                         pSubscription->thingNameLength,
                         pSubscription->pThingName );

            IOT_SET_AND_GOTO_CLEANUP( false );
        }
    }

    /* Check for active subscriptions. If any Jobs callbacks are active, then the
     * subscription cannot be removed. */
    if( pSubscription->callbackReferences > 0 )
    {
        IotLogDebug( "Notify callbacks are using %.*s subscription object. "
                     "Subscription cannot be removed yet.",
                     pSubscription->thingNameLength,
                     pSubscription->pThingName );

        IOT_SET_AND_GOTO_CLEANUP( false );
    }

    for( i = 0; i < JOBS_CALLBACK_COUNT; i++ )
    {
        for( callbackIndex = 0; callbackIndex < AWS_IOT_JOBS_NOTIFY_CALLBACKS; callbackIndex++ )
        {
            if( pSubscription->callbacks[ i ][ callbackIndex ].function != NULL )
            {
                IotLogDebug( "Found active Jobs %s callback for %.*s subscription object. "
                             "Subscription cannot be removed yet.",
                             _pAwsIotJobsCallbackNames[ i ],
                             pSubscription->thingNameLength,
                             pSubscription->pThingName );

                IOT_SET_AND_GOTO_CLEANUP( false );
            }
        }
    }

    /* Remove the subscription if unused. */
    IOT_FUNCTION_CLEANUP_BEGIN();

    if( status == true )
    {
        /* No Jobs operation subscription references or active Jobs callbacks.
         * Remove the subscription object. */
        IotListDouble_Remove( &( pSubscription->link ) );

        IotLogDebug( "Removed subscription object for %.*s.",
                     pSubscription->thingNameLength,
                     pSubscription->pThingName );

        /* If the caller requested the removed subscription, set the output parameter.
         * Otherwise, free the memory used by the subscription. */
        if( pRemovedSubscription != NULL )
        {
            *pRemovedSubscription = pSubscription;
        }
        else
        {
            _AwsIotJobs_DestroySubscription( pSubscription );
        }
    }
}

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

void _AwsIotJobs_DestroySubscription( void * pData )
{
    _jobsSubscription_t * pSubscription = ( _jobsSubscription_t * ) pData;

    /* Free the topic buffer if allocated. */
    if( pSubscription->pTopicBuffer != NULL )
    {
        AwsIotJobs_FreeString( pSubscription->pTopicBuffer );
    }

    /* Free memory used by subscription. */
    AwsIotJobs_FreeSubscription( pSubscription );
}

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

AwsIotJobsError_t _AwsIotJobs_IncrementReferences( _jobsOperation_t * pOperation,
                                                   char * pTopicBuffer,
                                                   uint16_t operationTopicLength,
                                                   AwsIotMqttCallbackFunction_t callback )
{
    IOT_FUNCTION_ENTRY( AwsIotJobsError_t, AWS_IOT_JOBS_SUCCESS );
    const _jobsOperationType_t type = pOperation->type;
    _jobsSubscription_t * pSubscription = pOperation->pSubscription;
    IotMqttError_t subscriptionStatus = IOT_MQTT_STATUS_PENDING;
    AwsIotSubscriptionInfo_t subscriptionInfo = { 0 };

    /* Do nothing if this operation has persistent subscriptions. */
    if( pSubscription->operationReferences[ type ] == AWS_IOT_PERSISTENT_SUBSCRIPTION )
    {
        IotLogDebug( "Jobs %s for %.*s has persistent subscriptions. Reference "
                     "count will not be incremented.",
                     _pAwsIotJobsOperationNames[ type ],
                     pSubscription->thingNameLength,
                     pSubscription->pThingName );

        IOT_GOTO_CLEANUP();
    }

    /* When persistent subscriptions are not present, the reference count must
     * not be negative. */
    AwsIotJobs_Assert( pSubscription->operationReferences[ type ] >= 0 );

    /* Check if there are any existing references for this operation. */
    if( pSubscription->operationReferences[ type ] == 0 )
    {
        /* Set the parameters needed to add subscriptions. */
        subscriptionInfo.mqttConnection = pOperation->mqttConnection;
        subscriptionInfo.callbackFunction = callback;
        subscriptionInfo.timeout = _AwsIotJobsMqttTimeoutMs;
        subscriptionInfo.pTopicFilterBase = pTopicBuffer;
        subscriptionInfo.topicFilterBaseLength = operationTopicLength;

        subscriptionStatus = AwsIot_ModifySubscriptions( IotMqtt_SubscribeSync,
                                                         &subscriptionInfo );

        /* Convert MQTT return code to Jobs return code. */
        switch( subscriptionStatus )
        {
            case IOT_MQTT_SUCCESS:
                status = AWS_IOT_JOBS_SUCCESS;
                break;

            case IOT_MQTT_NO_MEMORY:
                status = AWS_IOT_JOBS_NO_MEMORY;
                break;

            default:
                status = AWS_IOT_JOBS_MQTT_ERROR;
                break;
        }

        if( status != AWS_IOT_JOBS_SUCCESS )
        {
            IOT_GOTO_CLEANUP();
        }
    }

    /* Increment the number of subscription references for this operation when
     * the keep subscriptions flag is not set. */
    if( ( pOperation->flags & AWS_IOT_JOBS_FLAG_KEEP_SUBSCRIPTIONS ) == 0 )
    {
        ( pSubscription->operationReferences[ type ] )++;

        IotLogDebug( "Jobs %s subscriptions for %.*s now has count %d.",
                     _pAwsIotJobsOperationNames[ type ],
                     pSubscription->thingNameLength,
                     pSubscription->pThingName,
                     pSubscription->operationReferences[ type ] );
    }
    /* Otherwise, set the persistent subscriptions flag. */
    else
    {
        pSubscription->operationReferences[ type ] = AWS_IOT_PERSISTENT_SUBSCRIPTION;

        IotLogDebug( "Set persistent subscriptions flag for Jobs %s of %.*s.",
                     _pAwsIotJobsOperationNames[ type ],
                     pSubscription->thingNameLength,
                     pSubscription->pThingName );
    }

    IOT_FUNCTION_EXIT_NO_CLEANUP();
}

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

void _AwsIotJobs_DecrementReferences( _jobsOperation_t * pOperation,
                                      char * pTopicBuffer,
                                      _jobsSubscription_t ** pRemovedSubscription )
{
    const _jobsOperationType_t type = pOperation->type;
    _jobsSubscription_t * pSubscription = pOperation->pSubscription;
    uint16_t operationTopicLength = 0;
    AwsIotSubscriptionInfo_t subscriptionInfo = { 0 };
    AwsIotJobsRequestInfo_t requestInfo = AWS_IOT_JOBS_REQUEST_INFO_INITIALIZER;

    /* Do nothing if this Jobs operation has persistent subscriptions. */
    if( pSubscription->operationReferences[ type ] != AWS_IOT_PERSISTENT_SUBSCRIPTION )
    {
        /* Decrement the number of subscription references for this operation.
         * Ensure that it's positive. */
        ( pSubscription->operationReferences[ type ] )--;
        AwsIotJobs_Assert( pSubscription->operationReferences[ type ] >= 0 );

        /* Check if the number of references has reached 0. */
        if( pSubscription->operationReferences[ type ] == 0 )
        {
            IotLogDebug( "Reference count for %.*s %s is 0. Unsubscribing.",
                         pSubscription->thingNameLength,
                         pSubscription->pThingName,
                         _pAwsIotJobsOperationNames[ type ] );

            /* Subscription must have a topic buffer. */
            AwsIotJobs_Assert( pSubscription->pTopicBuffer != NULL );

            /* Set the parameters needed to generate a Jobs topic. */
            requestInfo.pThingName = pSubscription->pThingName;
            requestInfo.thingNameLength = pSubscription->thingNameLength;
            requestInfo.pJobId = pOperation->pJobId;
            requestInfo.jobIdLength = pOperation->jobIdLength;

            /* Generate the prefix of the Jobs topic. This function will not
             * fail when given a buffer. */
            ( void ) _AwsIotJobs_GenerateJobsTopic( ( _jobsOperationType_t ) type,
                                                    &requestInfo,
                                                    &( pSubscription->pTopicBuffer ),
                                                    &operationTopicLength );

            /* Set the parameters needed to remove subscriptions. */
            subscriptionInfo.mqttConnection = pOperation->mqttConnection;
            subscriptionInfo.timeout = _AwsIotJobsMqttTimeoutMs;
            subscriptionInfo.pTopicFilterBase = pTopicBuffer;
            subscriptionInfo.topicFilterBaseLength = operationTopicLength;

            ( void ) AwsIot_ModifySubscriptions( IotMqtt_UnsubscribeSync,
                                                 &subscriptionInfo );
        }

        /* Check if this subscription should be deleted. */
        _AwsIotJobs_RemoveSubscription( pSubscription,
                                        pRemovedSubscription );
    }
    else
    {
        IotLogDebug( "Jobs %s for %.*s has persistent subscriptions. Reference "
                     "count will not be decremented.",
                     _pAwsIotJobsOperationNames[ type ],
                     pSubscription->thingNameLength,
                     pSubscription->pThingName );
    }
}

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

AwsIotJobsError_t AwsIotJobs_RemovePersistentSubscriptions( const AwsIotJobsRequestInfo_t * pRequestInfo,
                                                            uint32_t flags )
{
    IOT_FUNCTION_ENTRY( AwsIotJobsError_t, AWS_IOT_JOBS_SUCCESS );
    int32_t i = 0;
    uint16_t operationTopicLength = 0;
    IotMqttError_t unsubscribeStatus = IOT_MQTT_STATUS_PENDING;
    AwsIotSubscriptionInfo_t subscriptionInfo = { 0 };
    _jobsSubscription_t * pSubscription = NULL;
    IotLink_t * pSubscriptionLink = NULL;
    AwsIotThingName_t thingName = { 0 };

    thingName.pThingName = pRequestInfo->pThingName;
    thingName.thingNameLength = pRequestInfo->thingNameLength;

    IotLogInfo( "Removing persistent subscriptions for %.*s.",
                pRequestInfo->thingNameLength,
                pRequestInfo->pThingName );

    /* Check parameters. */
    if( pRequestInfo->mqttConnection == IOT_MQTT_CONNECTION_INITIALIZER )
    {
        IotLogError( "MQTT connection is not initialized." );

        IOT_SET_AND_GOTO_CLEANUP( AWS_IOT_JOBS_BAD_PARAMETER );
    }

    if( AwsIot_ValidateThingName( pRequestInfo->pThingName,
                                  pRequestInfo->thingNameLength ) == false )
    {
        IotLogError( "Thing Name is not valid." );

        IOT_SET_AND_GOTO_CLEANUP( AWS_IOT_JOBS_BAD_PARAMETER );
    }

    if( ( ( flags & AWS_IOT_JOBS_FLAG_REMOVE_DESCRIBE_SUBSCRIPTIONS ) != 0 ) ||
        ( ( flags & AWS_IOT_JOBS_FLAG_REMOVE_UPDATE_SUBSCRIPTIONS ) != 0 ) )
    {
        if( ( pRequestInfo->pJobId == NULL ) || ( pRequestInfo->jobIdLength == 0 ) )
        {
            IotLogError( "Job ID must be set." );

            IOT_SET_AND_GOTO_CLEANUP( AWS_IOT_JOBS_BAD_PARAMETER );
        }

        if( pRequestInfo->jobIdLength > JOBS_MAX_ID_LENGTH )
        {
            IotLogError( "Job ID cannot be longer than %d.",
                         JOBS_MAX_ID_LENGTH );

            IOT_SET_AND_GOTO_CLEANUP( AWS_IOT_JOBS_BAD_PARAMETER );
        }
    }

    IotMutex_Lock( &( _AwsIotJobsSubscriptionsMutex ) );

    /* Search the list for an existing subscription for Thing Name. */
    pSubscriptionLink = IotListDouble_FindFirstMatch( &( _AwsIotJobsSubscriptions ),
                                                      NULL,
                                                      _jobsSubscription_match,
                                                      &thingName );

    if( pSubscriptionLink != NULL )
    {
        IotLogDebug( "Found subscription object for %.*s. Checking for persistent "
                     "subscriptions to remove.",
                     pRequestInfo->thingNameLength,
                     pRequestInfo->pThingName );

        pSubscription = IotLink_Container( _jobsSubscription_t, pSubscriptionLink, link );

        for( i = 0; i < JOBS_OPERATION_COUNT; i++ )
        {
            if( ( flags & ( 0x1UL << i ) ) != 0 )
            {
                IotLogDebug( "Removing %.*s %s persistent subscriptions.",
                             pRequestInfo->thingNameLength,
                             pRequestInfo->pThingName,
                             _pAwsIotJobsOperationNames[ i ] );

                /* Subscription must have a topic buffer. */
                AwsIotJobs_Assert( pSubscription->pTopicBuffer != NULL );

                if( pSubscription->operationReferences[ i ] == AWS_IOT_PERSISTENT_SUBSCRIPTION )
                {
                    /* Generate the prefix of the Jobs topic. This function will not
                     * fail when given a buffer. */
                    ( void ) _AwsIotJobs_GenerateJobsTopic( ( _jobsOperationType_t ) i,
                                                            pRequestInfo,
                                                            &( pSubscription->pTopicBuffer ),
                                                            &operationTopicLength );

                    /* Set the parameters needed to remove subscriptions. */
                    subscriptionInfo.mqttConnection = pRequestInfo->mqttConnection;
                    subscriptionInfo.timeout = _AwsIotJobsMqttTimeoutMs;
                    subscriptionInfo.pTopicFilterBase = pSubscription->pTopicBuffer;
                    subscriptionInfo.topicFilterBaseLength = operationTopicLength;

                    unsubscribeStatus = AwsIot_ModifySubscriptions( IotMqtt_UnsubscribeSync,
                                                                    &subscriptionInfo );

                    /* Convert MQTT return code to Shadow return code. */
                    switch( unsubscribeStatus )
                    {
                        case IOT_MQTT_SUCCESS:
                            status = AWS_IOT_JOBS_SUCCESS;
                            break;

                        case IOT_MQTT_NO_MEMORY:
                            status = AWS_IOT_JOBS_NO_MEMORY;
                            break;

                        default:
                            status = AWS_IOT_JOBS_MQTT_ERROR;
                            break;
                    }

                    if( status != AWS_IOT_JOBS_SUCCESS )
                    {
                        break;
                    }

                    /* Clear the persistent subscriptions flag and check if the
                     * subscription can be removed. */
                    pSubscription->operationReferences[ i ] = 0;
                    _AwsIotJobs_RemoveSubscription( pSubscription, NULL );
                }
                else
                {
                    IotLogDebug( "%.*s %s does not have persistent subscriptions.",
                                 pRequestInfo->thingNameLength,
                                 pRequestInfo->pThingName,
                                 _pAwsIotJobsOperationNames[ i ] );
                }
            }
        }
    }
    else
    {
        IotLogWarn( "No subscription object found for %.*s",
                    pRequestInfo->thingNameLength,
                    pRequestInfo->pThingName );
    }

    IotMutex_Unlock( &( _AwsIotJobsSubscriptionsMutex ) );

    IOT_FUNCTION_EXIT_NO_CLEANUP();
}

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