| /* | |
| * 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_operation.c | |
| * @brief Implements functions that process Jobs operations. | |
| */ | |
| /* 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" | |
| /* Platform threads include. */ | |
| #include "platform/iot_threads.h" | |
| /* Error handling include. */ | |
| #include "iot_error.h" | |
| /* MQTT include. */ | |
| #include "iot_mqtt.h" | |
| /*-----------------------------------------------------------*/ | |
| #if LIBRARY_LOG_LEVEL > IOT_LOG_NONE | |
| /** | |
| * @brief Printable names for each of the Jobs operations. | |
| */ | |
| const char * const _pAwsIotJobsOperationNames[] = | |
| { | |
| "GET PENDING", | |
| "START NEXT", | |
| "DESCRIBE", | |
| "UPDATE", | |
| "SET NOTIFY-PENDING", | |
| "SET NOTIFY-NEXT" | |
| }; | |
| #endif /* if LIBRARY_LOG_LEVEL > IOT_LOG_NONE */ | |
| /*-----------------------------------------------------------*/ | |
| /** | |
| * @brief First parameter to #_jobsOperation_match. | |
| */ | |
| typedef struct _operationMatchParams | |
| { | |
| _jobsOperationType_t type; /**< @brief GET PENDING, START NEXT, DESCRIBE, or UPDATE. */ | |
| const char * pThingName; /**< @brief Thing Name of Jobs operation. */ | |
| size_t thingNameLength; /**< @brief Length of #_operationMatchParams_t.pThingName. */ | |
| const char * pResponse; /**< @brief Jobs response document. */ | |
| size_t responseLength; /**< @brief Length of #_operationMatchParams_t.pResponse. */ | |
| } _operationMatchParams_t; | |
| /*-----------------------------------------------------------*/ | |
| /** | |
| * @brief Match a received Jobs response with a Jobs operation awaiting a | |
| * response. | |
| * | |
| * @param[in] pOperationLink Pointer to the link member of the #_jobsOperation_t | |
| * to check. | |
| * @param[in] pMatch Pointer to an #_operationMatchParams_t. | |
| * | |
| * @return `true` if `pMatch` matches the received response; `false` otherwise. | |
| */ | |
| static bool _jobsOperation_match( const IotLink_t * pOperationLink, | |
| void * pMatch ); | |
| /** | |
| * @brief Invoked when a Jobs response is received for Jobs GET PENDING. | |
| * | |
| * @param[in] pArgument Ignored. | |
| * @param[in] pMessage Received Jobs response (as an MQTT PUBLISH message). | |
| */ | |
| static void _getPendingCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ); | |
| /** | |
| * @brief Invoked when a Jobs response is received for a Jobs START NEXT. | |
| * | |
| * @param[in] pArgument Ignored. | |
| * @param[in] pMessage Received Jobs response (as an MQTT PUBLISH message). | |
| */ | |
| static void _startNextCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ); | |
| /** | |
| * @brief Invoked when a Jobs response is received for Jobs DESCRIBE. | |
| * | |
| * @param[in] pArgument Ignored. | |
| * @param[in] pMessage Received Jobs response (as an MQTT PUBLISH message). | |
| */ | |
| static void _describeCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ); | |
| /** | |
| * @brief Invoked when a Jobs response is received for a Jobs UPDATE. | |
| * | |
| * @param[in] pArgument Ignored. | |
| * @param[in] pMessage Received Jobs response (as an MQTT PUBLISH message). | |
| */ | |
| static void _updateCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ); | |
| /** | |
| * @brief Common function for processing received Jobs responses. | |
| * | |
| * @param[in] type GET PENDING, START NEXT, DESCRIBE, or UPDATE. | |
| * @param[in] pMessage Received Jobs response (as an MQTT PUBLISH message). | |
| */ | |
| static void _commonOperationCallback( _jobsOperationType_t type, | |
| IotMqttCallbackParam_t * pMessage ); | |
| /** | |
| * @brief Notify of a completed Jobs operation. | |
| * | |
| * @param[in] pOperation The operation which completed. | |
| * | |
| * Depending on the parameters passed to a user-facing Jobs function, the | |
| * notification will cause @ref jobs_function_wait to return or invoke a | |
| * user-provided callback. | |
| */ | |
| static void _notifyCompletion( _jobsOperation_t * pOperation ); | |
| /** | |
| * @brief Get a Jobs subscription to use with a Jobs operation. | |
| * | |
| * This function may use an existing Jobs subscription, or it may allocate a | |
| * new one. | |
| * | |
| * @param[in] pRequestInfo Common Jobs request parameters. | |
| * @param[in] pTopicBuffer Contains the topic to use for subscribing. | |
| * @param[in] operationTopicLength The length of the base topic in `pTopicBuffer`. | |
| * @param[in] pOperation Jobs operation that needs a subscription. | |
| * @param[out] pFreeTopicBuffer Whether the caller may free `pTopicBuffer` | |
| * (which may be assigned to a subscription). | |
| * | |
| * @return #AWS_IOT_JOBS_SUCCESS or #AWS_IOT_JOBS_NO_MEMORY | |
| */ | |
| static AwsIotJobsError_t _findSubscription( const AwsIotJobsRequestInfo_t * pRequestInfo, | |
| char * pTopicBuffer, | |
| uint16_t operationTopicLength, | |
| _jobsOperation_t * pOperation, | |
| bool * pFreeTopicBuffer ); | |
| /*-----------------------------------------------------------*/ | |
| /** | |
| * @brief List of active Jobs operations awaiting a response from the Jobs | |
| * service. | |
| */ | |
| IotListDouble_t _AwsIotJobsPendingOperations = { 0 }; | |
| /** | |
| * @brief Protects #_AwsIotJobsPendingOperations from concurrent access. | |
| */ | |
| IotMutex_t _AwsIotJobsPendingOperationsMutex; | |
| /*-----------------------------------------------------------*/ | |
| static bool _jobsOperation_match( const IotLink_t * pOperationLink, | |
| void * pMatch ) | |
| { | |
| /* Because this function is called from a container function, the given link | |
| * must never be NULL. */ | |
| AwsIotJobs_Assert( pOperationLink != NULL ); | |
| _jobsOperation_t * pOperation = IotLink_Container( _jobsOperation_t, | |
| pOperationLink, | |
| link ); | |
| _operationMatchParams_t * pParam = ( _operationMatchParams_t * ) pMatch; | |
| _jobsSubscription_t * pSubscription = pOperation->pSubscription; | |
| const char * pClientToken = NULL; | |
| size_t clientTokenLength = 0; | |
| /* Check for matching Thing Name and operation type. */ | |
| bool match = ( pOperation->type == pParam->type ) && | |
| ( pParam->thingNameLength == pSubscription->thingNameLength ) && | |
| ( strncmp( pParam->pThingName, | |
| pSubscription->pThingName, | |
| pParam->thingNameLength ) == 0 ); | |
| if( match == true ) | |
| { | |
| IotLogDebug( "Verifying client tokens for Jobs %s.", | |
| _pAwsIotJobsOperationNames[ pParam->type ] ); | |
| /* Check the response for a client token. */ | |
| match = AwsIot_GetClientToken( pParam->pResponse, | |
| pParam->responseLength, | |
| &pClientToken, | |
| &clientTokenLength ); | |
| /* If the response contains a client token, check for a match. */ | |
| if( match == true ) | |
| { | |
| match = ( pOperation->clientTokenLength == clientTokenLength ) && | |
| ( strncmp( pOperation->pClientToken, pClientToken, clientTokenLength ) == 0 ); | |
| } | |
| else | |
| { | |
| IotLogWarn( "Received a Jobs %s response with no client token. " | |
| "This is possibly a response to a bad JSON document:\r\n%.*s", | |
| _pAwsIotJobsOperationNames[ pParam->type ], | |
| pParam->responseLength, | |
| pParam->pResponse ); | |
| } | |
| } | |
| return match; | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static void _getPendingCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ) | |
| { | |
| /* Silence warnings about unused parameter. */ | |
| ( void ) pArgument; | |
| _commonOperationCallback( JOBS_GET_PENDING, pMessage ); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static void _startNextCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ) | |
| { | |
| /* Silence warnings about unused parameter. */ | |
| ( void ) pArgument; | |
| _commonOperationCallback( JOBS_START_NEXT, pMessage ); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static void _describeCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ) | |
| { | |
| /* Silence warnings about unused parameter. */ | |
| ( void ) pArgument; | |
| _commonOperationCallback( JOBS_DESCRIBE, pMessage ); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static void _updateCallback( void * pArgument, | |
| IotMqttCallbackParam_t * pMessage ) | |
| { | |
| /* Silence warnings about unused parameter. */ | |
| ( void ) pArgument; | |
| _commonOperationCallback( JOBS_UPDATE, pMessage ); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static void _commonOperationCallback( _jobsOperationType_t type, | |
| IotMqttCallbackParam_t * pMessage ) | |
| { | |
| _jobsOperation_t * pOperation = NULL; | |
| IotLink_t * pOperationLink = NULL; | |
| _operationMatchParams_t param = { 0 }; | |
| AwsIotStatus_t status = AWS_IOT_UNKNOWN; | |
| uint32_t flags = 0; | |
| /* Set operation type and response. */ | |
| param.type = type; | |
| param.pResponse = pMessage->u.message.info.pPayload; | |
| param.responseLength = pMessage->u.message.info.payloadLength; | |
| /* Parse the Thing Name from the MQTT topic name. */ | |
| if( AwsIot_ParseThingName( pMessage->u.message.info.pTopicName, | |
| pMessage->u.message.info.topicNameLength, | |
| &( param.pThingName ), | |
| &( param.thingNameLength ) ) == false ) | |
| { | |
| IOT_GOTO_CLEANUP(); | |
| } | |
| /* Lock the pending operations list for exclusive access. */ | |
| IotMutex_Lock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| /* Search for a matching pending operation. */ | |
| pOperationLink = IotListDouble_FindFirstMatch( &_AwsIotJobsPendingOperations, | |
| NULL, | |
| _jobsOperation_match, | |
| ¶m ); | |
| /* Find and remove the first Jobs operation of the given type. */ | |
| if( pOperationLink == NULL ) | |
| { | |
| /* Operation is not pending. It may have already been processed. Return | |
| * without doing anything */ | |
| IotMutex_Unlock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| IotLogWarn( "Jobs %s callback received an unknown operation.", | |
| _pAwsIotJobsOperationNames[ type ] ); | |
| IOT_GOTO_CLEANUP(); | |
| } | |
| else | |
| { | |
| pOperation = IotLink_Container( _jobsOperation_t, pOperationLink, link ); | |
| /* Copy the flags from the Jobs operation. */ | |
| flags = pOperation->flags; | |
| /* Remove a non-waitable operation from the pending operation list. | |
| * Waitable operations are removed by the Wait function. */ | |
| if( ( flags & AWS_IOT_JOBS_FLAG_WAITABLE ) == 0 ) | |
| { | |
| IotListDouble_Remove( &( pOperation->link ) ); | |
| IotMutex_Unlock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| } | |
| } | |
| /* Parse the status from the topic name. */ | |
| status = AwsIot_ParseStatus( pMessage->u.message.info.pTopicName, | |
| pMessage->u.message.info.topicNameLength ); | |
| switch( status ) | |
| { | |
| case AWS_IOT_ACCEPTED: | |
| case AWS_IOT_REJECTED: | |
| _AwsIotJobs_ParseResponse( status, | |
| pMessage->u.message.info.pPayload, | |
| pMessage->u.message.info.payloadLength, | |
| pOperation ); | |
| break; | |
| default: | |
| IotLogWarn( "Unknown status for %s of %.*s Jobs. Ignoring message.", | |
| _pAwsIotJobsOperationNames[ type ], | |
| pOperation->pSubscription->thingNameLength, | |
| pOperation->pSubscription->pThingName ); | |
| pOperation->status = AWS_IOT_JOBS_BAD_RESPONSE; | |
| break; | |
| } | |
| _notifyCompletion( pOperation ); | |
| /* For waitable operations, unlock the pending operation list mutex to allow | |
| * the Wait function to run. */ | |
| if( ( flags & AWS_IOT_JOBS_FLAG_WAITABLE ) == AWS_IOT_JOBS_FLAG_WAITABLE ) | |
| { | |
| IotMutex_Unlock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| } | |
| /* This function has no return value and no cleanup, but uses the cleanup | |
| * label to exit on error. */ | |
| IOT_FUNCTION_CLEANUP_BEGIN(); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static void _notifyCompletion( _jobsOperation_t * pOperation ) | |
| { | |
| AwsIotJobsCallbackParam_t callbackParam = { .callbackType = ( AwsIotJobsCallbackType_t ) 0 }; | |
| _jobsSubscription_t * pSubscription = pOperation->pSubscription, | |
| * pRemovedSubscription = NULL; | |
| /* If the operation is waiting, post to its wait semaphore and return. */ | |
| if( ( pOperation->flags & AWS_IOT_JOBS_FLAG_WAITABLE ) == AWS_IOT_JOBS_FLAG_WAITABLE ) | |
| { | |
| IotSemaphore_Post( &( pOperation->notify.waitSemaphore ) ); | |
| } | |
| else | |
| { | |
| /* Decrement the reference count. This also removes subscriptions if the | |
| * count reaches 0. */ | |
| IotMutex_Lock( &_AwsIotJobsSubscriptionsMutex ); | |
| _AwsIotJobs_DecrementReferences( pOperation, | |
| pSubscription->pTopicBuffer, | |
| &pRemovedSubscription ); | |
| IotMutex_Unlock( &_AwsIotJobsSubscriptionsMutex ); | |
| /* Set the subscription pointer used for the user callback based on whether | |
| * a subscription was removed from the list. */ | |
| if( pRemovedSubscription != NULL ) | |
| { | |
| pSubscription = pRemovedSubscription; | |
| } | |
| AwsIotJobs_Assert( pSubscription != NULL ); | |
| /* Invoke the user callback if provided. */ | |
| if( pOperation->notify.callback.function != NULL ) | |
| { | |
| /* Set the common members of the callback parameter. */ | |
| callbackParam.callbackType = ( AwsIotJobsCallbackType_t ) pOperation->type; | |
| callbackParam.mqttConnection = pOperation->mqttConnection; | |
| callbackParam.pThingName = pSubscription->pThingName; | |
| callbackParam.thingNameLength = pSubscription->thingNameLength; | |
| callbackParam.u.operation.result = pOperation->status; | |
| callbackParam.u.operation.reference = pOperation; | |
| callbackParam.u.operation.pResponse = pOperation->pJobsResponse; | |
| callbackParam.u.operation.responseLength = pOperation->jobsResponseLength; | |
| pOperation->notify.callback.function( pOperation->notify.callback.pCallbackContext, | |
| &callbackParam ); | |
| } | |
| /* Destroy a removed subscription. */ | |
| if( pRemovedSubscription != NULL ) | |
| { | |
| _AwsIotJobs_DestroySubscription( pRemovedSubscription ); | |
| } | |
| _AwsIotJobs_DestroyOperation( pOperation ); | |
| } | |
| } | |
| /*-----------------------------------------------------------*/ | |
| static AwsIotJobsError_t _findSubscription( const AwsIotJobsRequestInfo_t * pRequestInfo, | |
| char * pTopicBuffer, | |
| uint16_t operationTopicLength, | |
| _jobsOperation_t * pOperation, | |
| bool * pFreeTopicBuffer ) | |
| { | |
| AwsIotJobsError_t status = AWS_IOT_JOBS_SUCCESS; | |
| _jobsSubscription_t * pSubscription = NULL; | |
| /* Lookup table for Jobs operation callbacks. */ | |
| const AwsIotMqttCallbackFunction_t jobsCallbacks[ JOBS_OPERATION_COUNT ] = | |
| { | |
| _getPendingCallback, | |
| _startNextCallback, | |
| _describeCallback, | |
| _updateCallback | |
| }; | |
| /* Lock the subscriptions mutex for exclusive access. */ | |
| IotMutex_Lock( &_AwsIotJobsSubscriptionsMutex ); | |
| /* Check for an existing subscription. This function will attempt to allocate | |
| * a new subscription if not found. */ | |
| pSubscription = _AwsIotJobs_FindSubscription( pRequestInfo->pThingName, | |
| pRequestInfo->thingNameLength, | |
| true ); | |
| if( pSubscription == NULL ) | |
| { | |
| status = AWS_IOT_JOBS_NO_MEMORY; | |
| } | |
| else | |
| { | |
| /* Ensure that the subscription Thing Name matches. */ | |
| AwsIotJobs_Assert( pSubscription != NULL ); | |
| AwsIotJobs_Assert( pSubscription->thingNameLength == pRequestInfo->thingNameLength ); | |
| AwsIotJobs_Assert( strncmp( pSubscription->pThingName, | |
| pRequestInfo->pThingName, | |
| pRequestInfo->thingNameLength ) == 0 ); | |
| /* Set the subscription object for the Jobs operation. */ | |
| pOperation->pSubscription = pSubscription; | |
| /* Assign the topic buffer to the subscription to use for unsubscribing if | |
| * the subscription has no topic buffer. */ | |
| if( pSubscription->pTopicBuffer == NULL ) | |
| { | |
| pSubscription->pTopicBuffer = pTopicBuffer; | |
| /* Don't free the topic buffer if it was allocated to the subscription. */ | |
| *pFreeTopicBuffer = false; | |
| } | |
| else | |
| { | |
| *pFreeTopicBuffer = true; | |
| } | |
| /* Increment the reference count for this Jobs operation's | |
| * subscriptions. */ | |
| status = _AwsIotJobs_IncrementReferences( pOperation, | |
| pTopicBuffer, | |
| operationTopicLength, | |
| jobsCallbacks[ pOperation->type ] ); | |
| if( status != AWS_IOT_JOBS_SUCCESS ) | |
| { | |
| /* Failed to add subscriptions for a Jobs operation. The reference | |
| * count was not incremented. Check if this subscription should be | |
| * deleted. */ | |
| _AwsIotJobs_RemoveSubscription( pSubscription, NULL ); | |
| } | |
| } | |
| /* Unlock the Jobs subscription list mutex. */ | |
| IotMutex_Unlock( &_AwsIotJobsSubscriptionsMutex ); | |
| return status; | |
| } | |
| /*-----------------------------------------------------------*/ | |
| AwsIotJobsError_t _AwsIotJobs_CreateOperation( _jobsOperationType_t type, | |
| const AwsIotJobsRequestInfo_t * pRequestInfo, | |
| const _jsonRequestContents_t * pRequestContents, | |
| uint32_t flags, | |
| const AwsIotJobsCallbackInfo_t * pCallbackInfo, | |
| _jobsOperation_t ** pNewOperation ) | |
| { | |
| IOT_FUNCTION_ENTRY( AwsIotJobsError_t, AWS_IOT_JOBS_SUCCESS ); | |
| size_t operationSize = sizeof( _jobsOperation_t ); | |
| _jobsOperation_t * pOperation = NULL; | |
| IotLogDebug( "Creating operation record for Jobs %s.", | |
| _pAwsIotJobsOperationNames[ type ] ); | |
| /* The Job ID must be saved for DESCRIBE and UPDATE operations. */ | |
| if( ( type == JOBS_DESCRIBE ) || ( type == JOBS_UPDATE ) ) | |
| { | |
| /* Ensure a valid Job ID is provided. */ | |
| AwsIotJobs_Assert( pRequestInfo->pJobId != NULL ); | |
| AwsIotJobs_Assert( pRequestInfo->jobIdLength > 1 ); | |
| AwsIotJobs_Assert( pRequestInfo->jobIdLength <= JOBS_MAX_ID_LENGTH ); | |
| operationSize += pRequestInfo->jobIdLength; | |
| } | |
| /* Allocate memory for a new Jobs operation. */ | |
| pOperation = AwsIotJobs_MallocOperation( operationSize ); | |
| if( pOperation == NULL ) | |
| { | |
| IotLogError( "Failed to allocate memory for Jobs %s.", | |
| _pAwsIotJobsOperationNames[ type ] ); | |
| IOT_SET_AND_GOTO_CLEANUP( AWS_IOT_JOBS_NO_MEMORY ); | |
| } | |
| /* Clear the operation data. */ | |
| ( void ) memset( pOperation, 0x00, sizeof( _jobsOperation_t ) ); | |
| /* Set the remaining common members of the Jobs operation. */ | |
| pOperation->type = type; | |
| pOperation->flags = flags; | |
| pOperation->status = AWS_IOT_JOBS_STATUS_PENDING; | |
| pOperation->mallocResponse = pRequestInfo->mallocResponse; | |
| /* Save the Job ID for DESCRIBE and UPDATE operations. */ | |
| if( ( type == JOBS_DESCRIBE ) || ( type == JOBS_UPDATE ) ) | |
| { | |
| pOperation->jobIdLength = pRequestInfo->jobIdLength; | |
| ( void ) memcpy( pOperation->pJobId, pRequestInfo->pJobId, pRequestInfo->jobIdLength ); | |
| } | |
| /* Generate a Jobs request document. */ | |
| status = _AwsIotJobs_GenerateJsonRequest( type, | |
| pRequestInfo, | |
| pRequestContents, | |
| pOperation ); | |
| if( status != AWS_IOT_JOBS_SUCCESS ) | |
| { | |
| IOT_GOTO_CLEANUP(); | |
| } | |
| /* Check if the waitable flag is set. If it is, create a semaphore to | |
| * wait on. */ | |
| if( ( flags & AWS_IOT_JOBS_FLAG_WAITABLE ) == AWS_IOT_JOBS_FLAG_WAITABLE ) | |
| { | |
| if( IotSemaphore_Create( &( pOperation->notify.waitSemaphore ), 0, 1 ) == false ) | |
| { | |
| IotLogError( "Failed to create semaphore for waitable Jobs %s.", | |
| _pAwsIotJobsOperationNames[ type ] ); | |
| IOT_SET_AND_GOTO_CLEANUP( AWS_IOT_JOBS_NO_MEMORY ); | |
| } | |
| } | |
| else | |
| { | |
| /* If the waitable flag isn't set but a callback is, copy the callback | |
| * information. */ | |
| if( pCallbackInfo != NULL ) | |
| { | |
| pOperation->notify.callback = *pCallbackInfo; | |
| } | |
| } | |
| IOT_FUNCTION_CLEANUP_BEGIN(); | |
| /* Clean up on error. */ | |
| if( status != AWS_IOT_JOBS_SUCCESS ) | |
| { | |
| if( pOperation != NULL ) | |
| { | |
| if( pOperation->pJobsRequest != NULL ) | |
| { | |
| AwsIotJobs_FreeString( ( void * ) ( pOperation->pJobsRequest ) ); | |
| } | |
| AwsIotJobs_FreeOperation( pOperation ); | |
| } | |
| } | |
| else | |
| { | |
| /* Set the output parameter. */ | |
| *pNewOperation = pOperation; | |
| } | |
| IOT_FUNCTION_CLEANUP_END(); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| void _AwsIotJobs_DestroyOperation( void * pData ) | |
| { | |
| _jobsOperation_t * pOperation = ( _jobsOperation_t * ) pData; | |
| AwsIotJobs_Assert( pOperation != NULL ); | |
| IotLogDebug( "Destroying Jobs operation %s.", | |
| _pAwsIotJobsOperationNames[ pOperation->type ] ); | |
| /* Check if a wait semaphore was created for this operation. */ | |
| if( ( pOperation->flags & AWS_IOT_JOBS_FLAG_WAITABLE ) == AWS_IOT_JOBS_FLAG_WAITABLE ) | |
| { | |
| /* Destroy the wait semaphore */ | |
| IotSemaphore_Destroy( &( pOperation->notify.waitSemaphore ) ); | |
| } | |
| /* Free any Jobs request documents. */ | |
| if( pOperation->pJobsRequest != NULL ) | |
| { | |
| AwsIotJobs_Assert( pOperation->jobsRequestLength > 0 ); | |
| AwsIotJobs_FreeString( ( void * ) ( pOperation->pJobsRequest ) ); | |
| } | |
| /* Free the memory used to hold operation data. */ | |
| AwsIotJobs_FreeOperation( pOperation ); | |
| } | |
| /*-----------------------------------------------------------*/ | |
| AwsIotJobsError_t _AwsIotJobs_GenerateJobsTopic( _jobsOperationType_t type, | |
| const AwsIotJobsRequestInfo_t * pRequestInfo, | |
| char ** pTopicBuffer, | |
| uint16_t * pOperationTopicLength ) | |
| { | |
| AwsIotJobsError_t status = AWS_IOT_JOBS_SUCCESS; | |
| AwsIotTopicInfo_t topicInfo = { 0 }; | |
| char pJobOperationName[ JOBS_LONGEST_SUFFIX_LENGTH ] = { 0 }; | |
| uint16_t operationNameLength = 0; | |
| /* Lookup table for Jobs operation strings. */ | |
| const char * const pOperationString[ JOBS_OPERATION_COUNT ] = | |
| { | |
| JOBS_GET_PENDING_OPERATION_STRING, /* Jobs get pending operation. */ | |
| JOBS_START_NEXT_OPERATION_STRING, /* Jobs start next operation. */ | |
| JOBS_DESCRIBE_OPERATION_STRING, /* Jobs describe operation */ | |
| JOBS_UPDATE_OPERATION_STRING /* Jobs update operation. */ | |
| }; | |
| /* Lookup table for Jobs operation string lengths. */ | |
| const uint16_t pOperationStringLength[ JOBS_OPERATION_COUNT ] = | |
| { | |
| JOBS_GET_PENDING_OPERATION_STRING_LENGTH, /* Jobs get pending operation */ | |
| JOBS_START_NEXT_OPERATION_STRING_LENGTH, /* Jobs start next operation. */ | |
| JOBS_DESCRIBE_OPERATION_STRING_LENGTH, /* Jobs describe operation */ | |
| JOBS_UPDATE_OPERATION_STRING_LENGTH /* Jobs update operation. */ | |
| }; | |
| /* Ensure type is valid. */ | |
| AwsIotJobs_Assert( ( type == JOBS_GET_PENDING ) || ( type == JOBS_START_NEXT ) || | |
| ( type == JOBS_DESCRIBE ) || ( type == JOBS_UPDATE ) ); | |
| /* Set the members needed to generate an operation topic. */ | |
| topicInfo.pThingName = pRequestInfo->pThingName; | |
| topicInfo.thingNameLength = pRequestInfo->thingNameLength; | |
| topicInfo.longestSuffixLength = JOBS_LONGEST_SUFFIX_LENGTH; | |
| topicInfo.mallocString = AwsIotJobs_MallocString; | |
| /* Job operations that require a Job ID require additional processing to | |
| * create an operation name with the Job ID. */ | |
| if( ( type == JOBS_DESCRIBE ) || ( type == JOBS_UPDATE ) ) | |
| { | |
| /* Ensure Job ID length is valid. */ | |
| AwsIotJobs_Assert( pRequestInfo->jobIdLength > 0 ); | |
| AwsIotJobs_Assert( pRequestInfo->jobIdLength <= JOBS_MAX_ID_LENGTH ); | |
| /* Construct the Jobs operation name with the Job ID. */ | |
| ( void ) memcpy( pJobOperationName, "/jobs/", 6 ); | |
| operationNameLength = 6; | |
| ( void ) memcpy( pJobOperationName + operationNameLength, | |
| pRequestInfo->pJobId, | |
| pRequestInfo->jobIdLength ); | |
| operationNameLength = ( uint16_t ) ( pRequestInfo->jobIdLength + operationNameLength ); | |
| ( void ) memcpy( pJobOperationName + operationNameLength, | |
| pOperationString[ type ], | |
| pOperationStringLength[ type ] ); | |
| operationNameLength = ( uint16_t ) ( operationNameLength + pOperationStringLength[ type ] ); | |
| topicInfo.pOperationName = pJobOperationName; | |
| topicInfo.operationNameLength = operationNameLength; | |
| } | |
| else | |
| { | |
| topicInfo.pOperationName = pOperationString[ type ]; | |
| topicInfo.operationNameLength = pOperationStringLength[ type ]; | |
| } | |
| if( AwsIot_GenerateOperationTopic( &topicInfo, | |
| pTopicBuffer, | |
| pOperationTopicLength ) == false ) | |
| { | |
| status = AWS_IOT_JOBS_NO_MEMORY; | |
| } | |
| return status; | |
| } | |
| /*-----------------------------------------------------------*/ | |
| AwsIotJobsError_t _AwsIotJobs_ProcessOperation( const AwsIotJobsRequestInfo_t * pRequestInfo, | |
| _jobsOperation_t * pOperation ) | |
| { | |
| IOT_FUNCTION_ENTRY( AwsIotJobsError_t, AWS_IOT_JOBS_STATUS_PENDING ); | |
| char * pTopicBuffer = NULL; | |
| uint16_t operationTopicLength = 0; | |
| bool freeTopicBuffer = true; | |
| IotMqttPublishInfo_t publishInfo = IOT_MQTT_PUBLISH_INFO_INITIALIZER; | |
| IotMqttError_t publishStatus = IOT_MQTT_STATUS_PENDING; | |
| IotLogDebug( "Processing Jobs operation %s for Thing %.*s.", | |
| _pAwsIotJobsOperationNames[ pOperation->type ], | |
| pRequestInfo->thingNameLength, | |
| pRequestInfo->pThingName ); | |
| /* Set the operation's MQTT connection. */ | |
| pOperation->mqttConnection = pRequestInfo->mqttConnection; | |
| /* Generate the operation topic buffer. */ | |
| status = _AwsIotJobs_GenerateJobsTopic( pOperation->type, | |
| pRequestInfo, | |
| &pTopicBuffer, | |
| &operationTopicLength ); | |
| if( status != AWS_IOT_JOBS_SUCCESS ) | |
| { | |
| IotLogError( "No memory for Jobs operation topic buffer." ); | |
| IOT_GOTO_CLEANUP(); | |
| } | |
| /* Get a subscription object for this Jobs operation. */ | |
| status = _findSubscription( pRequestInfo, | |
| pTopicBuffer, | |
| operationTopicLength, | |
| pOperation, | |
| &freeTopicBuffer ); | |
| if( status != AWS_IOT_JOBS_SUCCESS ) | |
| { | |
| /* No subscription was found and no subscription could be allocated. */ | |
| IOT_GOTO_CLEANUP(); | |
| } | |
| /* Set the members for PUBLISH retry. */ | |
| publishInfo.qos = pRequestInfo->qos; | |
| publishInfo.retryLimit = pRequestInfo->retryLimit; | |
| publishInfo.retryMs = pRequestInfo->retryMs; | |
| /* Set the payload as the Jobs request. */ | |
| publishInfo.pPayload = pOperation->pJobsRequest; | |
| publishInfo.payloadLength = pOperation->jobsRequestLength; | |
| /* Set the operation topic name. */ | |
| publishInfo.pTopicName = pTopicBuffer; | |
| publishInfo.topicNameLength = operationTopicLength; | |
| IotLogDebug( "Jobs %s message will be published to topic %.*s", | |
| _pAwsIotJobsOperationNames[ pOperation->type ], | |
| publishInfo.topicNameLength, | |
| publishInfo.pTopicName ); | |
| /* Add Jobs operation to the pending operations list. */ | |
| IotMutex_Lock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| IotListDouble_InsertHead( &( _AwsIotJobsPendingOperations ), | |
| &( pOperation->link ) ); | |
| IotMutex_Unlock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| /* Publish to the Jobs topic name. */ | |
| publishStatus = IotMqtt_PublishSync( pOperation->mqttConnection, | |
| &publishInfo, | |
| 0, | |
| _AwsIotJobsMqttTimeoutMs ); | |
| if( publishStatus != IOT_MQTT_SUCCESS ) | |
| { | |
| IotLogError( "Failed to publish MQTT message to %s %.*s Jobs, error %s.", | |
| _pAwsIotJobsOperationNames[ pOperation->type ], | |
| pRequestInfo->thingNameLength, | |
| pRequestInfo->pThingName, | |
| IotMqtt_strerror( publishStatus ) ); | |
| /* Convert the MQTT "NO MEMORY" error to a Jobs "NO MEMORY" error. */ | |
| if( publishStatus == IOT_MQTT_NO_MEMORY ) | |
| { | |
| status = AWS_IOT_JOBS_NO_MEMORY; | |
| } | |
| else | |
| { | |
| status = AWS_IOT_JOBS_MQTT_ERROR; | |
| } | |
| /* If the "keep subscriptions" flag is not set, decrement the reference | |
| * count. */ | |
| if( ( pOperation->flags & AWS_IOT_JOBS_FLAG_KEEP_SUBSCRIPTIONS ) == 0 ) | |
| { | |
| IotMutex_Lock( &_AwsIotJobsSubscriptionsMutex ); | |
| _AwsIotJobs_DecrementReferences( pOperation, | |
| pTopicBuffer, | |
| NULL ); | |
| IotMutex_Unlock( &_AwsIotJobsSubscriptionsMutex ); | |
| } | |
| /* Remove Jobs operation from the pending operations list. */ | |
| IotMutex_Lock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| IotListDouble_Remove( &( pOperation->link ) ); | |
| IotMutex_Unlock( &( _AwsIotJobsPendingOperationsMutex ) ); | |
| } | |
| else | |
| { | |
| IotLogDebug( "Jobs %s PUBLISH message successfully sent.", | |
| _pAwsIotJobsOperationNames[ pOperation->type ] ); | |
| } | |
| IOT_FUNCTION_CLEANUP_BEGIN(); | |
| /* Free the topic buffer used by this function if it was not assigned to a | |
| * subscription. */ | |
| if( ( freeTopicBuffer == true ) && ( pTopicBuffer != NULL ) ) | |
| { | |
| AwsIotJobs_FreeString( pTopicBuffer ); | |
| } | |
| /* Destroy the Jobs operation on failure. */ | |
| if( status != AWS_IOT_JOBS_SUCCESS ) | |
| { | |
| _AwsIotJobs_DestroyOperation( pOperation ); | |
| } | |
| else | |
| { | |
| /* Convert successful return code to "status pending", as the Jobs | |
| * library is now waiting for a response from the service. */ | |
| status = AWS_IOT_JOBS_STATUS_PENDING; | |
| } | |
| IOT_FUNCTION_CLEANUP_END(); | |
| } | |
| /*-----------------------------------------------------------*/ |