github-forks
/
FreeRTOS
mirror of https://github.com/FreeRTOS/FreeRTOS.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add retries to demos in case of a failure. (#435)

Browse Source
pull/439/head
leegeth 4 years ago committed by GitHub
parent d2fcf20caf
commit 4651c46479
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 987 additions and 500 deletions
Show Stats Download Patch File Download Diff File

339
FreeRTOS-Plus/Demo/AWS/Device_Defender_Windows_Simulator/Device_Defender_Demo/DemoTasks/DefenderDemoExample.c
Unescape Escape View File

@ -107,6 +107,22 @@
*/ */
#define DEFENDER_RESPONSE_REPORT_ID_FIELD_LENGTH ( sizeof( DEFENDER_RESPONSE_REPORT_ID_FIELD ) - 1 ) #define DEFENDER_RESPONSE_REPORT_ID_FIELD_LENGTH ( sizeof( DEFENDER_RESPONSE_REPORT_ID_FIELD ) - 1 )
/**
* @brief The maximum number of times to run the loop in this demo.
*
* @note The demo loop is attempted to re-run only if it fails in an iteration.
* Once the demo loop succeeds in an iteration, the demo exits successfully.
*/
#ifndef DEFENDER_MAX_DEMO_LOOP_COUNT
#define DEFENDER_MAX_DEMO_LOOP_COUNT ( 3 )
#endif
/**
* @brief Time in ticks to wait between retries of the demo loop if
* demo loop fails.
*/
#define DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS ( pdMS_TO_TICKS( 5000U ) )
/** /**
* @brief Status values of the device defender report. * @brief Status values of the device defender report.
*/ */
@ -633,6 +649,7 @@ void prvDefenderDemoTask( void * pvParameters )
bool xStatus = false; bool xStatus = false;
BaseType_t xExitStatus = EXIT_FAILURE; BaseType_t xExitStatus = EXIT_FAILURE;
uint32_t ulReportLength = 0UL, i, ulMqttSessionEstablished = 0UL; uint32_t ulReportLength = 0UL, i, ulMqttSessionEstablished = 0UL;
UBaseType_t uxDemoRunCount = 0UL;
/* Remove compiler warnings about unused parameters. */ /* Remove compiler warnings about unused parameters. */
( void ) pvParameters; ( void ) pvParameters;
@ -643,188 +660,220 @@ void prvDefenderDemoTask( void * pvParameters )
/* Start with report not received. */ /* Start with report not received. */
xReportStatus = ReportStatusNotReceived; xReportStatus = ReportStatusNotReceived;
/* Set a report Id to be used. /* This demo runs a single loop unless there are failures in the demo execution.
* * In case of failures in the demo execution, demo loop will be retried for up to
* !!!NOTE!!! * DEFENDER_MAX_DEMO_LOOP_COUNT times. */
* This demo sets the report ID to xTaskGetTickCount(), which may collide do
* if the device is reset. Reports for a Thing with a previously used
* report ID will be assumed to be duplicates and discarded by the Device
* Defender service. The report ID needs to be unique per report sent with
* a given Thing. We recommend using an increasing unique id such as the
* current timestamp. */
ulReportId = ( uint32_t ) xTaskGetTickCount();
/****************************** Connect. ******************************/
/* Attempts to connect to the AWS IoT MQTT broker over TCP. If the
* connection fails, retries after a timeout. Timeout value will
* exponentially increase until maximum attempts are reached. */
LogInfo( ( "Establishing MQTT session..." ) );
xStatus = xEstablishMqttSession( &xMqttContext,
&xNetworkContext,
&xBuffer,
prvPublishCallback );
if( xStatus != true )
{
LogError( ( "Failed to establish MQTT session." ) );
}
else
{
ulMqttSessionEstablished = 1;
}
/******************** Subscribe to Defender topics. *******************/
/* Attempt to subscribe to the AWS IoT Device Defender topics.
* Since this demo is using JSON, in prvSubscribeToDefenderTopics() we
* subscribe to the topics to which accepted and rejected responses are
* received from after publishing a JSON report.
*
* This demo uses a constant #democonfigTHING_NAME known at compile time
* therefore we use macros to assemble defender topic strings.
* If the thing name is known at run time, then we could use the API
* #Defender_GetTopic instead.
*
* For example, for the JSON accepted responses topic:
*
* #define TOPIC_BUFFER_LENGTH ( 256U )
*
* // Every device should have a unique thing name registered with AWS IoT Core.
* // This example assumes that the device has a unique serial number which is
* // registered as the thing name with AWS IoT Core.
* const char * pThingName = GetDeviceSerialNumber();
* uint16_t thingNameLength = ( uint16_t )strlen( pThingname );
* char topicBuffer[ TOPIC_BUFFER_LENGTH ] = { 0 };
* uint16_t topicLength = 0;
* DefenderStatus_t status = DefenderSuccess;
*
* status = Defender_GetTopic( &( topicBuffer[ 0 ] ),
* TOPIC_BUFFER_LENGTH,
* pThingName,
* thingNameLength,
* DefenderJsonReportAccepted,
* &( topicLength ) );
*/
if( xStatus == true )
{ {
LogInfo( ( "Subscribing to defender topics..." ) ); /* Set a report Id to be used.
xStatus = prvSubscribeToDefenderTopics(); *
* !!!NOTE!!!
* This demo sets the report ID to xTaskGetTickCount(), which may collide
* if the device is reset. Reports for a Thing with a previously used
* report ID will be assumed to be duplicates and discarded by the Device
* Defender service. The report ID needs to be unique per report sent with
* a given Thing. We recommend using an increasing unique id such as the
* current timestamp. */
ulReportId = ( uint32_t ) xTaskGetTickCount();
/****************************** Connect. ******************************/
/* Attempts to connect to the AWS IoT MQTT broker over TCP. If the
* connection fails, retries after a timeout. Timeout value will
* exponentially increase until maximum attempts are reached. */
LogInfo( ( "Establishing MQTT session..." ) );
xStatus = xEstablishMqttSession( &xMqttContext,
&xNetworkContext,
&xBuffer,
prvPublishCallback );
if( xStatus != true ) if( xStatus != true )
{ {
LogError( ( "Failed to subscribe to defender topics." ) ); LogError( ( "Failed to establish MQTT session." ) );
}
else
{
ulMqttSessionEstablished = 1;
} }
}
/*********************** Collect device metrics. **********************/ /******************** Subscribe to Defender topics. *******************/
/* Attempt to subscribe to the AWS IoT Device Defender topics.
* Since this demo is using JSON, in prvSubscribeToDefenderTopics() we
* subscribe to the topics to which accepted and rejected responses are
* received from after publishing a JSON report.
*
* This demo uses a constant #democonfigTHING_NAME known at compile time
* therefore we use macros to assemble defender topic strings.
* If the thing name is known at run time, then we could use the API
* #Defender_GetTopic instead.
*
* For example, for the JSON accepted responses topic:
*
* #define TOPIC_BUFFER_LENGTH ( 256U )
*
* // Every device should have a unique thing name registered with AWS IoT Core.
* // This example assumes that the device has a unique serial number which is
* // registered as the thing name with AWS IoT Core.
* const char * pThingName = GetDeviceSerialNumber();
* uint16_t thingNameLength = ( uint16_t )strlen( pThingname );
* char topicBuffer[ TOPIC_BUFFER_LENGTH ] = { 0 };
* uint16_t topicLength = 0;
* DefenderStatus_t status = DefenderSuccess;
*
* status = Defender_GetTopic( &( topicBuffer[ 0 ] ),
* TOPIC_BUFFER_LENGTH,
* pThingName,
* thingNameLength,
* DefenderJsonReportAccepted,
* &( topicLength ) );
*/
if( xStatus == true )
{
LogInfo( ( "Subscribing to defender topics..." ) );
xStatus = prvSubscribeToDefenderTopics();
/* We then need to collect the metrics that will be sent to the AWS IoT if( xStatus != true )
* Device Defender service. This demo uses the functions declared in {
* in metrics_collector.h to collect network metrics. For this demo, the LogError( ( "Failed to subscribe to defender topics." ) );
* implementation of these functions are in metrics_collector.c and }
* collects metrics using tcp_netstat utility for FreeRTOS+TCP. */ }
if( xStatus == true )
{
LogInfo( ( "Collecting device metrics..." ) );
xStatus = prvCollectDeviceMetrics();
if( xStatus != true ) /*********************** Collect device metrics. **********************/
/* We then need to collect the metrics that will be sent to the AWS IoT
* Device Defender service. This demo uses the functions declared in
* in metrics_collector.h to collect network metrics. For this demo, the
* implementation of these functions are in metrics_collector.c and
* collects metrics using tcp_netstat utility for FreeRTOS+TCP. */
if( xStatus == true )
{ {
LogError( ( "Failed to collect device metrics." ) ); LogInfo( ( "Collecting device metrics..." ) );
xStatus = prvCollectDeviceMetrics();
if( xStatus != true )
{
LogError( ( "Failed to collect device metrics." ) );
}
} }
}
/********************** Generate defender report. *********************/ /********************** Generate defender report. *********************/
/* The data needs to be incorporated into a JSON formatted report, /* The data needs to be incorporated into a JSON formatted report,
* which follows the format expected by the Device Defender service. * which follows the format expected by the Device Defender service.
* This format is documented here: * This format is documented here:
* https://docs.aws.amazon.com/iot/latest/developerguide/detect-device-side-metrics.html * https://docs.aws.amazon.com/iot/latest/developerguide/detect-device-side-metrics.html
*/ */
if( xStatus == true ) if( xStatus == true )
{ {
LogInfo( ( "Generating device defender report..." ) ); LogInfo( ( "Generating device defender report..." ) );
xStatus = prvGenerateDeviceMetricsReport( &( ulReportLength ) ); xStatus = prvGenerateDeviceMetricsReport( &( ulReportLength ) );
if( xStatus != true ) if( xStatus != true )
{
LogError( ( "Failed to generate device defender report." ) );
}
}
/********************** Publish defender report. **********************/
/* The report is then published to the Device Defender service. This report
* is published to the MQTT topic for publishing JSON reports. As before,
* we use the defender library macros to create the topic string, though
* #Defender_GetTopic could be used if the Thing name is acquired at
* run time */
if( xStatus == true )
{ {
LogError( ( "Failed to generate device defender report." ) ); LogInfo( ( "Publishing device defender report..." ) );
xStatus = prvPublishDeviceMetricsReport( ulReportLength );
if( xStatus != true )
{
LogError( ( "Failed to publish device defender report." ) );
}
} }
}
/********************** Publish defender report. **********************/ /* Wait for the response to our report. Response will be handled by the
* callback passed to xEstablishMqttSession() earlier.
* The callback will verify that the MQTT messages received are from the
* defender service's topic. Based on whether the response comes from
* the accepted or rejected topics, it updates xReportStatus. */
if( xStatus == true )
{
for( i = 0; i < DEFENDER_RESPONSE_WAIT_SECONDS; i++ )
{
( void ) xProcessLoop( &xMqttContext );
/* xReportStatus is updated in the prvPublishCallback. */
if( xReportStatus != ReportStatusNotReceived )
{
break;
}
/* The report is then published to the Device Defender service. This report /* Wait for sometime between consecutive executions of ProcessLoop. */
* is published to the MQTT topic for publishing JSON reports. As before, vTaskDelay( pdMS_TO_TICKS( 1000U ) );
* we use the defender library macros to create the topic string, though }
* #Defender_GetTopic could be used if the Thing name is acquired at }
* run time */
if( xStatus == true )
{
LogInfo( ( "Publishing device defender report..." ) );
xStatus = prvPublishDeviceMetricsReport( ulReportLength );
if( xStatus != true ) if( xReportStatus == ReportStatusNotReceived )
{ {
LogError( ( "Failed to publish device defender report." ) ); LogError( ( "Failed to receive response from AWS IoT Device Defender Service." ) );
xStatus = false;
} }
}
/* Wait for the response to our report. Response will be handled by the /**************************** Disconnect. *****************************/
* callback passed to xEstablishMqttSession() earlier.
* The callback will verify that the MQTT messages received are from the /* Unsubscribe and disconnect if MQTT session was established. Per the MQTT
* defender service's topic. Based on whether the response comes from * protocol spec, it is okay to send UNSUBSCRIBE even if no corresponding
* the accepted or rejected topics, it updates xReportStatus. */ * subscription exists on the broker. Therefore, it is okay to attempt
if( xStatus == true ) * unsubscribe even if one more subscribe failed earlier. */
{ if( ulMqttSessionEstablished == 1 )
for( i = 0; i < DEFENDER_RESPONSE_WAIT_SECONDS; i++ )
{ {
( void ) xProcessLoop( &xMqttContext ); LogInfo( ( "Unsubscribing from defender topics..." ) );
xStatus = prvUnsubscribeFromDefenderTopics();
/* xReportStatus is updated in the prvPublishCallback. */ if( xStatus != true )
if( xReportStatus != ReportStatusNotReceived )
{ {
break; LogError( ( "Failed to unsubscribe from defender topics." ) );
} }
/* Wait for sometime between consecutive executions of ProcessLoop. */ LogInfo( ( "Closing MQTT session..." ) );
vTaskDelay( 1000 / portTICK_PERIOD_MS ); ( void ) xDisconnectMqttSession( &xMqttContext,
&xNetworkContext );
} }
}
if( xReportStatus == ReportStatusNotReceived ) if( ( xStatus == true ) && ( xReportStatus == ReportStatusAccepted ) )
{ {
LogError( ( "Failed to receive response from AWS IoT Device Defender Service." ) ); xExitStatus = EXIT_SUCCESS;
xStatus = false; }
}
/**************************** Disconnect. *****************************/ /*********************** Retry in case of failure. ************************/
/* Unsubscribe and disconnect if MQTT session was established. Per the MQTT /* Increment the demo run count. */
* protocol spec, it is okay to send UNSUBSCRIBE even if no corresponding uxDemoRunCount++;
* subscription exists on the broker. Therefore, it is okay to attempt
* unsubscribe even if one more subscribe failed earlier. */
if( ulMqttSessionEstablished == 1 )
{
LogInfo( ( "Unsubscribing from defender topics..." ) );
xStatus = prvUnsubscribeFromDefenderTopics();
if( xStatus != true ) if( xExitStatus == EXIT_SUCCESS )
{ {
LogError( ( "Failed to unsubscribe from defender topics." ) ); LogInfo( ( "Demo iteration %lu is successful.", uxDemoRunCount ) );
} }
/* Attempt to retry a failed iteration of demo for up to #DEFENDER_MAX_DEMO_LOOP_COUNT times. */
LogInfo( ( "Closing MQTT session..." ) ); else if( uxDemoRunCount < DEFENDER_MAX_DEMO_LOOP_COUNT )
( void ) xDisconnectMqttSession( &xMqttContext, {
&xNetworkContext ); LogWarn( ( "Demo iteration %lu failed. Retrying...", uxDemoRunCount ) );
} vTaskDelay( DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS );
}
/* Failed all #DEFENDER_MAX_DEMO_LOOP_COUNT demo iterations. */
else
{
LogError( ( "All %d demo iterations failed.", DEFENDER_MAX_DEMO_LOOP_COUNT ) );
break;
}
} while( xExitStatus != EXIT_SUCCESS );
/****************************** Finish. ******************************/ /****************************** Finish. ******************************/
if( ( xStatus == true ) && ( xReportStatus == ReportStatusAccepted ) ) if( xExitStatus == EXIT_SUCCESS )
{ {
xExitStatus = EXIT_SUCCESS;
LogInfo( ( "Demo completed successfully." ) ); LogInfo( ( "Demo completed successfully." ) );
} }
else else

612
FreeRTOS-Plus/Demo/AWS/Device_Shadow_Windows_Simulator/Device_Shadow_Demo/DemoTasks/ShadowDemoMainExample.c
Unescape Escape View File

@ -148,6 +148,55 @@
*/ */
#define SHADOW_REPORTED_JSON_LENGTH ( sizeof( SHADOW_REPORTED_JSON ) - 3 ) #define SHADOW_REPORTED_JSON_LENGTH ( sizeof( SHADOW_REPORTED_JSON ) - 3 )
/**
* @brief The maximum number of times to run the loop in this demo.
*
* @note The demo loop is attempted to re-run only if it fails in an iteration.
* Once the demo loop succeeds in an iteration, the demo exits successfully.
*/
#ifndef SHADOW_MAX_DEMO_LOOP_COUNT
#define SHADOW_MAX_DEMO_LOOP_COUNT ( 3 )
#endif
/**
* @brief Time in ticks to wait between retries of the demo loop if
* demo loop fails.
*/
#define DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS ( pdMS_TO_TICKS( 5000U ) )
/**
* @brief The maximum number of times to call MQTT_ProcessLoop() when waiting
* for a response for Shadow delete operation.
*/
#define MQTT_PROCESS_LOOP_DELETE_RESPONSE_COUNT_MAX ( 30U )
/**
* @brief Timeout for MQTT_ProcessLoop in milliseconds.
*/
#define MQTT_PROCESS_LOOP_TIMEOUT_MS ( 700U )
/**
* @brief JSON key for response code that indicates the type of error in
* the error document received on topic `/delete/rejected`.
*/
#define SHADOW_DELETE_REJECTED_ERROR_CODE_KEY "code"
/**
* @brief Length of #SHADOW_DELETE_REJECTED_ERROR_CODE_KEY.
*/
#define SHADOW_DELETE_REJECTED_ERROR_CODE_KEY_LENGTH ( ( uint16_t ) ( sizeof( SHADOW_DELETE_REJECTED_ERROR_CODE_KEY ) - 1 ) )
/**
* @brief Error response code sent from AWS IoT Shadow service when an attempt
* is made to delete a Shadow document that doesn't exist.
*/
#define SHADOW_NO_SHADOW_EXISTS_ERROR_CODE "404"
/**
* @brief Length of #SHADOW_NO_SHADOW_EXISTS_ERROR_CODE.
*/
#define SHADOW_NO_SHADOW_EXISTS_ERROR_CODE_LENGTH ( ( uint16_t ) ( sizeof( SHADOW_NO_SHADOW_EXISTS_ERROR_CODE ) - 1 ) )
/*------------- Demo configurations -------------------------*/ /*------------- Demo configurations -------------------------*/
#ifndef democonfigTHING_NAME #ifndef democonfigTHING_NAME
@ -225,6 +274,24 @@ static BaseType_t xUpdateDeltaReturn = pdPASS;
*/ */
static BaseType_t xUpdateAcceptedReturn = pdPASS; static BaseType_t xUpdateAcceptedReturn = pdPASS;
/**
* @brief Status of the response of Shadow delete operation from AWS IoT
* message broker.
*/
static BaseType_t xDeleteResponseReceived = pdFALSE;
/**
* @brief Status of the Shadow delete operation.
*
* The Shadow delete status will be updated by the incoming publishes on the
* MQTT topics for delete acknowledgement from AWS IoT message broker
* (accepted/rejected). Shadow document is considered to be deleted if an
* incoming publish is received on `/delete/accepted` topic or an incoming
* publish is received on `/delete/rejected` topic with error code 404. Code 404
* indicates that the Shadow document does not exist for the Thing yet.
*/
static BaseType_t xShadowDeleted = pdFALSE;
/*-----------------------------------------------------------*/ /*-----------------------------------------------------------*/
/** /**
@ -271,6 +338,126 @@ static void prvUpdateAcceptedHandler( MQTTPublishInfo_t * pxPublishInfo );
*/ */
static void prvShadowDemoTask( void * pvParameters ); static void prvShadowDemoTask( void * pvParameters );
/**
* @brief Process payload from `/delete/rejected` topic.
*
* This handler examines the rejected message to look for the reject reason code.
* If the reject reason code is `404`, an attempt was made to delete a shadow
* document which was not present yet. This is considered to be success for this
* demo application.
*
* @param[in] pxPublishInfo Deserialized publish info pointer for the incoming
* packet.
*/
static void prvDeleteRejectedHandler( MQTTPublishInfo_t * pxPublishInfo );
/**
* @brief Helper function to wait for a response for Shadow delete operation.
*
* @param[in] pxMQTTContext MQTT context pointer.
*
* @return pdPASS if successfully received a response for Shadow delete
* operation; pdFAIL otherwise.
*/
static BaseType_t prvWaitForDeleteResponse( MQTTContext_t * pxMQTTContext );
/*-----------------------------------------------------------*/
static BaseType_t prvWaitForDeleteResponse( MQTTContext_t * pxMQTTContext )
{
uint8_t ucCount = 0U;
MQTTStatus_t xMQTTStatus = MQTTSuccess;
BaseType_t xReturnStatus = pdPASS;
configASSERT( pxMQTTContext != NULL );
while( ( xDeleteResponseReceived != pdTRUE ) &&
( ucCount++ < MQTT_PROCESS_LOOP_DELETE_RESPONSE_COUNT_MAX ) &&
( xMQTTStatus == MQTTSuccess ) )
{
/* Event callback will set #xDeleteResponseReceived when receiving an
* incoming publish on either `/delete/accepted` or `/delete/rejected`
* Shadow topics. */
xMQTTStatus = MQTT_ProcessLoop( pxMQTTContext, MQTT_PROCESS_LOOP_TIMEOUT_MS );
}
if( ( xMQTTStatus != MQTTSuccess ) || ( xDeleteResponseReceived != pdTRUE ) )
{
LogError( ( "MQTT_ProcessLoop failed to receive a response for Shadow delete operation:"
" LoopDuration=%u, MQTT Status=%s.",
( MQTT_PROCESS_LOOP_TIMEOUT_MS * ucCount ),
MQTT_Status_strerror( xMQTTStatus ) ) );
xReturnStatus = pdFAIL;
}
return xReturnStatus;
}
/*-----------------------------------------------------------*/
static void prvDeleteRejectedHandler( MQTTPublishInfo_t * pxPublishInfo )
{
JSONStatus_t result = JSONSuccess;
char * pcOutValue = NULL;
uint32_t ulOutValueLength = 0UL;
configASSERT( pxPublishInfo != NULL );
configASSERT( pxPublishInfo->pPayload != NULL );
LogInfo( ( "/delete/rejected json payload:%s.", ( const char * ) pxPublishInfo->pPayload ) );
/* The payload will look similar to this:
* {
* "code": error-code,
* "message": "error-message",
* "timestamp": timestamp,
* "clientToken": "token"
* }
*/
/* Make sure the payload is a valid json document. */
result = JSON_Validate( pxPublishInfo->pPayload,
pxPublishInfo->payloadLength );
if( result == JSONSuccess )
{
/* Then we start to get the version value by JSON keyword "version". */
result = JSON_SearchConst( pxPublishInfo->pPayload,
pxPublishInfo->payloadLength,
SHADOW_DELETE_REJECTED_ERROR_CODE_KEY,
SHADOW_DELETE_REJECTED_ERROR_CODE_KEY_LENGTH,
&pcOutValue,
( size_t * ) &ulOutValueLength,
NULL );
}
else
{
LogError( ( "The json document is invalid!!" ) );
}
if( result == JSONSuccess )
{
LogInfo( ( "Error code is: %.*s.",
ulOutValueLength,
pcOutValue ) );
/* Check if error code is `404`. An error code `404` indicates that an
* attempt was made to delete a Shadow document that didn't exist. */
if( ulOutValueLength == SHADOW_NO_SHADOW_EXISTS_ERROR_CODE_LENGTH )
{
if( strncmp( pcOutValue, SHADOW_NO_SHADOW_EXISTS_ERROR_CODE,
SHADOW_NO_SHADOW_EXISTS_ERROR_CODE_LENGTH ) == 0 )
{
xShadowDeleted = pdTRUE;
}
}
}
else
{
LogError( ( "No error code in json document!!" ) );
}
}
/*-----------------------------------------------------------*/ /*-----------------------------------------------------------*/
static void prvUpdateDeltaHandler( MQTTPublishInfo_t * pxPublishInfo ) static void prvUpdateDeltaHandler( MQTTPublishInfo_t * pxPublishInfo )
@ -536,6 +723,18 @@ static void prvEventCallback( MQTTContext_t * pxMqttContext,
{ {
LogInfo( ( "/update/rejected json payload:%s.", ( const char * ) pxDeserializedInfo->pPublishInfo->pPayload ) ); LogInfo( ( "/update/rejected json payload:%s.", ( const char * ) pxDeserializedInfo->pPublishInfo->pPayload ) );
} }
else if( messageType == ShadowMessageTypeDeleteAccepted )
{
LogInfo( ( "Received an MQTT incoming publish on /delete/accepted topic." ) );
xShadowDeleted = pdTRUE;
xDeleteResponseReceived = pdTRUE;
}
else if( messageType == ShadowMessageTypeDeleteRejected )
{
/* Handler function to process payload. */
prvDeleteRejectedHandler( pxDeserializedInfo->pPublishInfo );
xDeleteResponseReceived = pdTRUE;
}
else else
{ {
LogInfo( ( "Other message type:%d !!", messageType ) ); LogInfo( ( "Other message type:%d !!", messageType ) );
@ -593,7 +792,8 @@ void vStartShadowDemo( void )
*/ */
void prvShadowDemoTask( void * pvParameters ) void prvShadowDemoTask( void * pvParameters )
{ {
BaseType_t demoStatus = pdPASS; BaseType_t xDemoStatus = pdPASS;
UBaseType_t uxDemoRunCount = 0UL;
/* A buffer containing the update document. It has static duration to prevent /* A buffer containing the update document. It has static duration to prevent
* it from being placed on the call stack. */ * it from being placed on the call stack. */
@ -607,205 +807,291 @@ void prvShadowDemoTask( void * pvParameters )
/****************************** Connect. ******************************/ /****************************** Connect. ******************************/
demoStatus = xEstablishMqttSession( &xMqttContext, /* This demo runs a single loop unless there are failures in the demo execution.
&xNetworkContext, * In case of failures in the demo execution, demo loop will be retried for up to
&xBuffer, * SHADOW_MAX_DEMO_LOOP_COUNT times. */
prvEventCallback ); do
if( demoStatus == pdFAIL )
{ {
/* Log error to indicate connection failure. */ /****************************** Connect. ******************************/
LogError( ( "Failed to connect to MQTT broker." ) );
}
if( demoStatus == pdPASS ) xDemoStatus = xEstablishMqttSession( &xMqttContext,
{ &xNetworkContext,
/* Try to delete any Shadow document in the cloud. This is done to &xBuffer,
* ensure possbile previous Shadow documents do not affect the state prvEventCallback );
* expected in the current demo. */
demoStatus = xPublishToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_DELETE( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_DELETE( THING_NAME_LENGTH ),
pcUpdateDocument,
0U );
}
/********************* Subscribe to Shadow topics. ************************/ if( xDemoStatus == pdFAIL )
{
/* Log error to indicate connection failure. */
LogError( ( "Failed to connect to MQTT broker." ) );
}
/* Then try to subscribe the Shadow topics. */ /**************** Attempt to delete Shadow document. ******************/
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
demoStatus = xSubscribeToTopic( &xMqttContext, /* Reset the shadow delete status flags. */
SHADOW_TOPIC_STRING_UPDATE_DELTA( democonfigTHING_NAME ), xDeleteResponseReceived = pdFALSE;
SHADOW_TOPIC_LENGTH_UPDATE_DELTA( THING_NAME_LENGTH ) ); xShadowDeleted = pdFALSE;
}
/* First of all, try to delete any Shadow document in the cloud.
* Try to subscribe to `/delete/accepted` and `/delete/rejected` topics. */
xDemoStatus = xSubscribeToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_DELETE_ACCEPTED( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_DELETE_ACCEPTED( THING_NAME_LENGTH ) );
}
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
demoStatus = xSubscribeToTopic( &xMqttContext, /* Try to subscribe to `/delete/rejected` topic. */
SHADOW_TOPIC_STRING_UPDATE_ACCEPTED( democonfigTHING_NAME ), xDemoStatus = xSubscribeToTopic( &xMqttContext,
SHADOW_TOPIC_LENGTH_UPDATE_ACCEPTED( THING_NAME_LENGTH ) ); SHADOW_TOPIC_STRING_DELETE_REJECTED( democonfigTHING_NAME ),
} SHADOW_TOPIC_LENGTH_DELETE_REJECTED( THING_NAME_LENGTH ) );
}
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
demoStatus = xSubscribeToTopic( &xMqttContext, /* Publish to Shadow `delete` topic to attempt to delete the
SHADOW_TOPIC_STRING_UPDATE_REJECTED( democonfigTHING_NAME ), * Shadow document if exists. */
SHADOW_TOPIC_LENGTH_UPDATE_REJECTED( THING_NAME_LENGTH ) ); xDemoStatus = xPublishToTopic( &xMqttContext,
} SHADOW_TOPIC_STRING_DELETE( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_DELETE( THING_NAME_LENGTH ),
pcUpdateDocument,
0U );
}
/********************* Publish to Shadow topics. **********************/ /* Wait for an incoming publish on `/delete/accepted` or `/delete/rejected`
* topics, if not already received a publish. */
/* This demo uses a constant #democonfigTHING_NAME known at compile time if( ( xDemoStatus == pdPASS ) && ( xDeleteResponseReceived != pdTRUE ) )
* therefore we can use macros to assemble shadow topic strings. {
* If the thing name is known at run time, then we could use the API xDemoStatus = prvWaitForDeleteResponse( &xMqttContext );
* #Shadow_GetTopicString to assemble shadow topic strings, here is the }
* example for /update/delta:
*
* For /update/delta:
*
* #define SHADOW_TOPIC_MAX_LENGTH (256U)
*
* ShadowStatus_t shadowStatus = SHADOW_STATUS_SUCCESS;
* char cTopicBuffer[ SHADOW_TOPIC_MAX_LENGTH ] = { 0 };
* uint16_t usBufferSize = SHADOW_TOPIC_MAX_LENGTH;
* uint16_t usOutLength = 0;
* const char * pcThingName = "TestThingName";
* uint16_t usThingNameLength = ( sizeof( pcThingName ) - 1U );
*
* shadowStatus = Shadow_GetTopicString( SHADOW_TOPIC_STRING_TYPE_UPDATE_DELTA,
* pcThingName,
* usThingNameLength,
* & ( cTopicBuffer[ 0 ] ),
* usBufferSize,
* & usOutLength );
*/
/* Then we publish a desired state to the /update topic. Since we've deleted /* Unsubscribe from the `/delete/accepted` and 'delete/rejected` topics.*/
* the device shadow at the beginning of the demo, this will cause a delta if( xDemoStatus == pdPASS )
* message to be published, which we have subscribed to. {
* In many real applications, the desired state is not published by xDemoStatus = xUnsubscribeFromTopic( &xMqttContext,
* the device itself. But for the purpose of making this demo self-contained, SHADOW_TOPIC_STRING_DELETE_ACCEPTED( democonfigTHING_NAME ),
* we publish one here so that we can receive a delta message later. SHADOW_TOPIC_LENGTH_DELETE_ACCEPTED( THING_NAME_LENGTH ) );
*/ }
if( demoStatus == pdPASS )
{
/* Desired power on state . */
LogInfo( ( "Send desired power state with 1." ) );
( void ) memset( pcUpdateDocument,
0x00,
sizeof( pcUpdateDocument ) );
snprintf( pcUpdateDocument,
SHADOW_DESIRED_JSON_LENGTH + 1,
SHADOW_DESIRED_JSON,
( int ) 1,
( long unsigned ) ( xTaskGetTickCount() % 1000000 ) );
demoStatus = xPublishToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE( THING_NAME_LENGTH ),
pcUpdateDocument,
( SHADOW_DESIRED_JSON_LENGTH + 1 ) );
}
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
/* Note that PublishToTopic already called MQTT_ProcessLoop, xDemoStatus = xUnsubscribeFromTopic( &xMqttContext,
* therefore responses may have been received and the prvEventCallback SHADOW_TOPIC_STRING_DELETE_REJECTED( democonfigTHING_NAME ),
* may have been called, which may have changed the stateChanged flag. SHADOW_TOPIC_LENGTH_DELETE_REJECTED( THING_NAME_LENGTH ) );
* Check if the state change flag has been modified or not. If it's modified, }
* then we publish reported state to update topic.
/* Check if Shadow document delete was successful. A delete can be
* successful in cases listed below.
* 1. If an incoming publish is received on `/delete/accepted` topic.
* 2. If an incoming publish is received on `/delete/rejected` topic
* with error code 404. This indicates that a Shadow document was
* not present for the Thing. */
if( xShadowDeleted == pdFALSE )
{
LogError( ( "Shadow delete operation failed." ) );
xDemoStatus = pdFAIL;
}
/********************* Subscribe to Shadow topics. ************************/
/* Then try to subscribe the Shadow topics. */
if( xDemoStatus == pdPASS )
{
xDemoStatus = xSubscribeToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_DELTA( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_DELTA( THING_NAME_LENGTH ) );
}
if( xDemoStatus == pdPASS )
{
xDemoStatus = xSubscribeToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_ACCEPTED( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_ACCEPTED( THING_NAME_LENGTH ) );
}
if( xDemoStatus == pdPASS )
{
xDemoStatus = xSubscribeToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_REJECTED( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_REJECTED( THING_NAME_LENGTH ) );
}
/********************* Publish to Shadow topics. **********************/
/* This demo uses a constant #democonfigTHING_NAME known at compile time
* therefore we can use macros to assemble shadow topic strings.
* If the thing name is known at run time, then we could use the API
* #Shadow_GetTopicString to assemble shadow topic strings, here is the
* example for /update/delta:
*
* For /update/delta:
*
* #define SHADOW_TOPIC_MAX_LENGTH (256U)
*
* ShadowStatus_t shadowStatus = SHADOW_STATUS_SUCCESS;
* char cTopicBuffer[ SHADOW_TOPIC_MAX_LENGTH ] = { 0 };
* uint16_t usBufferSize = SHADOW_TOPIC_MAX_LENGTH;
* uint16_t usOutLength = 0;
* const char * pcThingName = "TestThingName";
* uint16_t usThingNameLength = ( sizeof( pcThingName ) - 1U );
*
* shadowStatus = Shadow_GetTopicString( SHADOW_TOPIC_STRING_TYPE_UPDATE_DELTA,
* pcThingName,
* usThingNameLength,
* & ( cTopicBuffer[ 0 ] ),
* usBufferSize,
* & usOutLength );
*/
/* Then we publish a desired state to the /update topic. Since we've deleted
* the device shadow at the beginning of the demo, this will cause a delta
* message to be published, which we have subscribed to.
* In many real applications, the desired state is not published by
* the device itself. But for the purpose of making this demo self-contained,
* we publish one here so that we can receive a delta message later.
*/ */
if( stateChanged == true ) if( xDemoStatus == pdPASS )
{ {
/* Report the latest power state back to device shadow. */ /* Desired power on state . */
LogInfo( ( "Report to the state change: %d", ulCurrentPowerOnState ) ); LogInfo( ( "Send desired power state with 1." ) );
( void ) memset( pcUpdateDocument, ( void ) memset( pcUpdateDocument,
0x00, 0x00,
sizeof( pcUpdateDocument ) ); sizeof( pcUpdateDocument ) );
/* Keep the client token in global variable used to compare if
* the same token in /update/accepted. */
ulClientToken = ( xTaskGetTickCount() % 1000000 );
snprintf( pcUpdateDocument, snprintf( pcUpdateDocument,
SHADOW_REPORTED_JSON_LENGTH + 1, SHADOW_DESIRED_JSON_LENGTH + 1,
SHADOW_REPORTED_JSON, SHADOW_DESIRED_JSON,
( int ) ulCurrentPowerOnState, ( int ) 1,
( long unsigned ) ulClientToken ); ( long unsigned ) ( xTaskGetTickCount() % 1000000 ) );
demoStatus = xPublishToTopic( &xMqttContext, xDemoStatus = xPublishToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE( democonfigTHING_NAME ), SHADOW_TOPIC_STRING_UPDATE( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE( THING_NAME_LENGTH ), SHADOW_TOPIC_LENGTH_UPDATE( THING_NAME_LENGTH ),
pcUpdateDocument, pcUpdateDocument,
( SHADOW_DESIRED_JSON_LENGTH + 1 ) ); ( SHADOW_DESIRED_JSON_LENGTH + 1 ) );
} }
else
if( xDemoStatus == pdPASS )
{ {
LogInfo( ( "No change from /update/delta, unsubscribe all shadow topics and disconnect from MQTT.\r\n" ) ); /* Note that PublishToTopic already called MQTT_ProcessLoop,
* therefore responses may have been received and the prvEventCallback
* may have been called, which may have changed the stateChanged flag.
* Check if the state change flag has been modified or not. If it's modified,
* then we publish reported state to update topic.
*/
if( stateChanged == true )
{
/* Report the latest power state back to device shadow. */
LogInfo( ( "Report to the state change: %d", ulCurrentPowerOnState ) );
( void ) memset( pcUpdateDocument,
0x00,
sizeof( pcUpdateDocument ) );
/* Keep the client token in global variable used to compare if
* the same token in /update/accepted. */
ulClientToken = ( xTaskGetTickCount() % 1000000 );
snprintf( pcUpdateDocument,
SHADOW_REPORTED_JSON_LENGTH + 1,
SHADOW_REPORTED_JSON,
( int ) ulCurrentPowerOnState,
( long unsigned ) ulClientToken );
xDemoStatus = xPublishToTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE( THING_NAME_LENGTH ),
pcUpdateDocument,
( SHADOW_DESIRED_JSON_LENGTH + 1 ) );
}
else
{
LogInfo( ( "No change from /update/delta, unsubscribe all shadow topics and disconnect from MQTT.\r\n" ) );
}
} }
}
/****************** Unsubscribe from Shadow topics. *******************/ /****************** Unsubscribe from Shadow topics. *******************/
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
LogInfo( ( "Start to unsubscribe shadow topics and disconnect from MQTT. \r\n" ) ); LogInfo( ( "Start to unsubscribe shadow topics and disconnect from MQTT. \r\n" ) );
demoStatus = xUnsubscribeFromTopic( &xMqttContext, xDemoStatus = xUnsubscribeFromTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_DELTA( democonfigTHING_NAME ), SHADOW_TOPIC_STRING_UPDATE_DELTA( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_DELTA( THING_NAME_LENGTH ) ); SHADOW_TOPIC_LENGTH_UPDATE_DELTA( THING_NAME_LENGTH ) );
if( demoStatus != pdPASS ) if( xDemoStatus != pdPASS )
{ {
LogError( ( "Failed to unsubscribe the topic %s", LogError( ( "Failed to unsubscribe the topic %s",
SHADOW_TOPIC_STRING_UPDATE_DELTA( democonfigTHING_NAME ) ) ); SHADOW_TOPIC_STRING_UPDATE_DELTA( democonfigTHING_NAME ) ) );
}
} }
}
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
demoStatus = xUnsubscribeFromTopic( &xMqttContext, xDemoStatus = xUnsubscribeFromTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_ACCEPTED( democonfigTHING_NAME ), SHADOW_TOPIC_STRING_UPDATE_ACCEPTED( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_ACCEPTED( THING_NAME_LENGTH ) ); SHADOW_TOPIC_LENGTH_UPDATE_ACCEPTED( THING_NAME_LENGTH ) );
if( xDemoStatus != pdPASS )
{
LogError( ( "Failed to unsubscribe the topic %s",
SHADOW_TOPIC_STRING_UPDATE_ACCEPTED( democonfigTHING_NAME ) ) );
}
}
if( demoStatus != pdPASS ) if( xDemoStatus == pdPASS )
{ {
LogError( ( "Failed to unsubscribe the topic %s", xDemoStatus = xUnsubscribeFromTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_ACCEPTED( democonfigTHING_NAME ) ) ); SHADOW_TOPIC_STRING_UPDATE_REJECTED( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_REJECTED( THING_NAME_LENGTH ) );
if( xDemoStatus != pdPASS )
{
LogError( ( "Failed to unsubscribe the topic %s",
SHADOW_TOPIC_STRING_UPDATE_REJECTED( democonfigTHING_NAME ) ) );
}
} }
}
if( demoStatus == pdPASS ) /****************************** Disconnect. *******************************/
{
demoStatus = xUnsubscribeFromTopic( &xMqttContext,
SHADOW_TOPIC_STRING_UPDATE_REJECTED( democonfigTHING_NAME ),
SHADOW_TOPIC_LENGTH_UPDATE_REJECTED( THING_NAME_LENGTH ) );
if( demoStatus != pdPASS ) /* The MQTT session is always disconnected, even if there were prior failures. */
xDemoStatus = xDisconnectMqttSession( &xMqttContext, &xNetworkContext );
/* This demo performs only Device Shadow operations. If matching the Shadow
* MQTT topic fails or there are failure in parsing the received JSON document,
* then this demo was not successful. */
if( ( xUpdateAcceptedReturn != pdPASS ) || ( xUpdateDeltaReturn != pdPASS ) )
{ {
LogError( ( "Failed to unsubscribe the topic %s", LogError( ( "Callback function failed." ) );
SHADOW_TOPIC_STRING_UPDATE_REJECTED( democonfigTHING_NAME ) ) );
} }
}
/****************************** Disconnect. *******************************/ /*********************** Retry in case of failure. ************************/
/* The MQTT session is always disconnected, even if there were prior failures. */ /* Increment the demo run count. */
demoStatus = xDisconnectMqttSession( &xMqttContext, &xNetworkContext ); uxDemoRunCount++;
/* This demo performs only Device Shadow operations. If matching the Shadow if( xDemoStatus == pdPASS )
* MQTT topic fails or there are failure in parsing the received JSON document, {
* then this demo was not successful. */ LogInfo( ( "Demo iteration %lu is successful.", uxDemoRunCount ) );
if( ( xUpdateAcceptedReturn != pdPASS ) || ( xUpdateDeltaReturn != pdPASS ) ) }
{ /* Attempt to retry a failed iteration of demo for up to #SHADOW_MAX_DEMO_LOOP_COUNT times. */
LogError( ( "Callback function failed." ) ); else if( uxDemoRunCount < SHADOW_MAX_DEMO_LOOP_COUNT )
} {
LogWarn( ( "Demo iteration %lu failed. Retrying...", uxDemoRunCount ) );
vTaskDelay( DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS );
}
/* Failed all #SHADOW_MAX_DEMO_LOOP_COUNT demo iterations. */
else
{
LogError( ( "All %d demo iterations failed.", SHADOW_MAX_DEMO_LOOP_COUNT ) );
break;
}
} while( xDemoStatus != pdPASS );
if( demoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
LogInfo( ( "Demo completed successfully." ) ); LogInfo( ( "Demo completed successfully." ) );
} }

260
FreeRTOS-Plus/Demo/AWS/Jobs_Windows_Simulator/Jobs_Demo/DemoTasks/JobsDemoExample.c
Unescape Escape View File

@ -208,6 +208,22 @@
*/ */
#define MAKE_STATUS_REPORT( x ) "{\"status\":\"" x "\"}" #define MAKE_STATUS_REPORT( x ) "{\"status\":\"" x "\"}"
/**
* @brief The maximum number of times to run the loop in this demo.
*
* @note The demo loop is attempted to re-run only if it fails in an iteration.
* Once the demo loop succeeds in an iteration, the demo exits successfully.
*/
#ifndef JOBS_MAX_DEMO_LOOP_COUNT
#define JOBS_MAX_DEMO_LOOP_COUNT ( 3 )
#endif
/**
* @brief Time in ticks to wait between retries of the demo loop if
* demo loop fails.
*/
#define DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS ( pdMS_TO_TICKS( 5000U ) )
/*-----------------------------------------------------------*/ /*-----------------------------------------------------------*/
/** /**
@ -735,6 +751,8 @@ void vStartJobsDemo( void )
void prvJobsDemoTask( void * pvParameters ) void prvJobsDemoTask( void * pvParameters )
{ {
BaseType_t xDemoStatus = pdPASS; BaseType_t xDemoStatus = pdPASS;
UBaseType_t uxDemoRunCount = 0UL;
BaseType_t retryDemoLoop = pdFALSE;
/* Remove compiler warnings about unused parameters. */ /* Remove compiler warnings about unused parameters. */
( void ) pvParameters; ( void ) pvParameters;
@ -742,119 +760,163 @@ void prvJobsDemoTask( void * pvParameters )
/* Set the pParams member of the network context with desired transport. */ /* Set the pParams member of the network context with desired transport. */
xNetworkContext.pParams = &xTlsTransportParams; xNetworkContext.pParams = &xTlsTransportParams;
/* Establish an MQTT connection with AWS IoT over a mutually authenticated TLS session. */ /* This demo runs a single loop unless there are failures in the demo execution.
xDemoStatus = xEstablishMqttSession( &xMqttContext, * In case of failures in the demo execution, demo loop will be retried for up to
&xNetworkContext, * JOBS_MAX_DEMO_LOOP_COUNT times. */
&xBuffer, do
prvEventCallback );
if( xDemoStatus == pdFAIL )
{ {
/* Log error to indicate connection failure. */ /* Establish an MQTT connection with AWS IoT over a mutually authenticated TLS session. */
LogError( ( "Failed to connect to AWS IoT broker." ) ); xDemoStatus = xEstablishMqttSession( &xMqttContext,
} &xNetworkContext,
else &xBuffer,
{ prvEventCallback );
/* Print out a short user guide to the console. The default logging
* limit of 255 characters can be changed in demo_logging.c, but breaking if( xDemoStatus == pdFAIL )
* up the only instance of a 1000+ character string is more practical. */
LogInfo( ( "\r\n"
"/*-----------------------------------------------------------*/\r\n"
"\r\n"
"The Jobs demo is now ready to accept Jobs.\r\n"
"Jobs may be created using the AWS IoT console or AWS CLI.\r\n"
"See the following link for more information.\r\n" ) );
LogInfo( ( "\r"
"https://docs.aws.amazon.com/cli/latest/reference/iot/create-job.html\r\n"
"\r\n"
"This demo expects Job documents to have an \"action\" JSON key.\r\n"
"The following actions are currently supported:\r\n" ) );
LogInfo( ( "\r"
" - print \r\n"
" Logs a message to the local console. The Job document must also contain a \"message\".\r\n"
" For example: { \"action\": \"print\", \"message\": \"Hello world!\"} will cause\r\n"
" \"Hello world!\" to be printed on the console.\r\n" ) );
LogInfo( ( "\r"
" - publish \r\n"
" Publishes a message to an MQTT topic. The Job document must also contain a \"message\" and \"topic\".\r\n" ) );
LogInfo( ( "\r"
" For example: { \"action\": \"publish\", \"topic\": \"demo/jobs\", \"message\": \"Hello world!\"} will cause\r\n"
" \"Hello world!\" to be published to the topic \"demo/jobs\".\r\n" ) );
LogInfo( ( "\r"
" - exit \r\n"
" Exits the demo program. This program will run until { \"action\": \"exit\" } is received.\r\n"
"\r\n"
"/*-----------------------------------------------------------*/\r\n" ) );
/* Subscribe to the NextJobExecutionChanged API topic to receive notifications about the next pending
* job in the queue for the Thing resource used by this demo. */
if( xSubscribeToTopic( &xMqttContext,
NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ),
sizeof( NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) - 1 ) != pdPASS )
{ {
xDemoStatus = pdFAIL; /* Log error to indicate connection failure. */
LogError( ( "Failed to subscribe to NextJobExecutionChanged API of AWS IoT Jobs service: Topic=%s", LogError( ( "Failed to connect to AWS IoT broker." ) );
NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) ); }
else
{
/* Print out a short user guide to the console. The default logging
* limit of 255 characters can be changed in demo_logging.c, but breaking
* up the only instance of a 1000+ character string is more practical. */
LogInfo( ( "\r\n"
"/*-----------------------------------------------------------*/\r\n"
"\r\n"
"The Jobs demo is now ready to accept Jobs.\r\n"
"Jobs may be created using the AWS IoT console or AWS CLI.\r\n"
"See the following link for more information.\r\n" ) );
LogInfo( ( "\r"
"https://docs.aws.amazon.com/cli/latest/reference/iot/create-job.html\r\n"
"\r\n"
"This demo expects Job documents to have an \"action\" JSON key.\r\n"
"The following actions are currently supported:\r\n" ) );
LogInfo( ( "\r"
" - print \r\n"
" Logs a message to the local console. The Job document must also contain a \"message\".\r\n"
" For example: { \"action\": \"print\", \"message\": \"Hello world!\"} will cause\r\n"
" \"Hello world!\" to be printed on the console.\r\n" ) );
LogInfo( ( "\r"
" - publish \r\n"
" Publishes a message to an MQTT topic. The Job document must also contain a \"message\" and \"topic\".\r\n" ) );
LogInfo( ( "\r"
" For example: { \"action\": \"publish\", \"topic\": \"demo/jobs\", \"message\": \"Hello world!\"} will cause\r\n"
" \"Hello world!\" to be published to the topic \"demo/jobs\".\r\n" ) );
LogInfo( ( "\r"
" - exit \r\n"
" Exits the demo program. This program will run until { \"action\": \"exit\" } is received.\r\n"
"\r\n"
"/*-----------------------------------------------------------*/\r\n" ) );
/* Subscribe to the NextJobExecutionChanged API topic to receive notifications about the next pending
* job in the queue for the Thing resource used by this demo. */
if( xSubscribeToTopic( &xMqttContext,
NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ),
sizeof( NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) - 1 ) != pdPASS )
{
xDemoStatus = pdFAIL;
LogError( ( "Failed to subscribe to NextJobExecutionChanged API of AWS IoT Jobs service: Topic=%s",
NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) );
}
} }
}
if( xDemoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{
/* Publish to AWS IoT Jobs on the StartNextPendingJobExecution API to request the next pending job.
*
* Note: It is not required to make MQTT subscriptions to the response topics of the
* StartNextPendingJobExecution API because the AWS IoT Jobs service sends responses for
* the PUBLISH commands on the same MQTT connection irrespective of whether the client has subscribed
* to the response topics or not.
* This demo processes incoming messages from the response topics of the API in the prvEventCallback()
* handler that is supplied to the coreMQTT library. */
if( xPublishToTopic( &xMqttContext,
START_NEXT_JOB_TOPIC( democonfigTHING_NAME ),
sizeof( START_NEXT_JOB_TOPIC( democonfigTHING_NAME ) ) - 1,
NULL,
0 ) != pdPASS )
{ {
xDemoStatus = pdFAIL; /* Publish to AWS IoT Jobs on the StartNextPendingJobExecution API to request the next pending job.
LogError( ( "Failed to publish to StartNextPendingJobExecution API of AWS IoT Jobs service: " *
"Topic=%s", START_NEXT_JOB_TOPIC( democonfigTHING_NAME ) ) ); * Note: It is not required to make MQTT subscriptions to the response topics of the
* StartNextPendingJobExecution API because the AWS IoT Jobs service sends responses for
* the PUBLISH commands on the same MQTT connection irrespective of whether the client has subscribed
* to the response topics or not.
* This demo processes incoming messages from the response topics of the API in the prvEventCallback()
* handler that is supplied to the coreMQTT library. */
if( xPublishToTopic( &xMqttContext,
START_NEXT_JOB_TOPIC( democonfigTHING_NAME ),
sizeof( START_NEXT_JOB_TOPIC( democonfigTHING_NAME ) ) - 1,
NULL,
0 ) != pdPASS )
{
xDemoStatus = pdFAIL;
LogError( ( "Failed to publish to StartNextPendingJobExecution API of AWS IoT Jobs service: "
"Topic=%s", START_NEXT_JOB_TOPIC( democonfigTHING_NAME ) ) );
}
} }
}
/* Keep on running the demo until we receive a job for the "exit" action to exit the demo. */ /* Keep on running the demo until we receive a job for the "exit" action to exit the demo. */
while( ( xExitActionJobReceived == pdFALSE ) && while( ( xExitActionJobReceived == pdFALSE ) &&
( xDemoEncounteredError == pdFALSE ) && ( xDemoEncounteredError == pdFALSE ) &&
( xDemoStatus == pdPASS ) ) ( xDemoStatus == pdPASS ) )
{ {
MQTTStatus_t xMqttStatus = MQTTSuccess; MQTTStatus_t xMqttStatus = MQTTSuccess;
/* Check if we have notification for the next pending job in the queue from the /* Check if we have notification for the next pending job in the queue from the
* NextJobExecutionChanged API of the AWS IoT Jobs service. */ * NextJobExecutionChanged API of the AWS IoT Jobs service. */
xMqttStatus = MQTT_ProcessLoop( &xMqttContext, 300U ); xMqttStatus = MQTT_ProcessLoop( &xMqttContext, 300U );
if( xMqttStatus != MQTTSuccess ) if( xMqttStatus != MQTTSuccess )
{
xDemoStatus = pdFAIL;
LogError( ( "Failed to receive notification about next pending job: "
"MQTT_ProcessLoop failed" ) );
}
}
/* Increment the demo run count. */
uxDemoRunCount++;
/* Retry demo loop only if there is a failure before completing
* the processing of any pending jobs. Any failure in MQTT unsubscribe
* or disconnect is considered a failure in demo execution and retry
* will not be attempted since a retry without any pending jobs will
* make this demo indefinitely wait. */
if( ( xDemoStatus == pdFAIL ) || ( xDemoEncounteredError == pdTRUE ) )
{
if( uxDemoRunCount < JOBS_MAX_DEMO_LOOP_COUNT )
{
LogWarn( ( "Demo iteration %lu failed. Retrying...", uxDemoRunCount ) );
retryDemoLoop = pdTRUE;
}
else
{
LogError( ( "All %d demo iterations failed.", JOBS_MAX_DEMO_LOOP_COUNT ) );
retryDemoLoop = pdFALSE;
}
}
else
{
/* Reset the flag for demo retry. */
retryDemoLoop = pdFALSE;
}
/* Unsubscribe from the NextJobExecutionChanged API topic. */
if( xUnsubscribeFromTopic( &xMqttContext,
NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ),
sizeof( NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) - 1 ) != pdPASS )
{ {
xDemoStatus = pdFAIL; xDemoStatus = pdFAIL;
LogError( ( "Failed to receive notification about next pending job: " LogError( ( "Failed to subscribe unsubscribe from the NextJobExecutionChanged API of AWS IoT Jobs service: "
"MQTT_ProcessLoop failed" ) ); "Topic=%s", NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) );
} }
}
/* Unsubscribe from the NextJobExecutionChanged API topic. */ /* Disconnect the MQTT and network connections with AWS IoT. */
if( xUnsubscribeFromTopic( &xMqttContext, if( xDisconnectMqttSession( &xMqttContext, &xNetworkContext ) != pdPASS )
NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ), {
sizeof( NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) - 1 ) != pdPASS ) xDemoStatus = pdFAIL;
{ LogError( ( "Disconnection from AWS IoT failed..." ) );
xDemoStatus == pdFAIL; }
LogError( ( "Failed to subscribe unsubscribe from the NextJobExecutionChanged API of AWS IoT Jobs service: "
"Topic=%s", NEXT_JOB_EXECUTION_CHANGED_TOPIC( democonfigTHING_NAME ) ) );
}
/* Disconnect the MQTT and network connections with AWS IoT. */ /* Add a delay if a retry is required. */
if( xDisconnectMqttSession( &xMqttContext, &xNetworkContext ) != pdPASS ) if( retryDemoLoop == pdTRUE )
{ {
xDemoStatus == pdFAIL; /* Clear the flag that indicates that indicates the demo error
LogError( ( "Disconnection from AWS Iot failed..." ) ); * before attempting a retry. */
} xDemoEncounteredError = pdFALSE;
LogInfo( ( "A short delay before the next demo iteration." ) );
vTaskDelay( DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS );
}
} while( retryDemoLoop == pdTRUE );
if( ( xDemoEncounteredError == pdFALSE ) && ( xDemoStatus == pdPASS ) ) if( ( xDemoEncounteredError == pdFALSE ) && ( xDemoStatus == pdPASS ) )
{ {

141
FreeRTOS-Plus/Demo/coreHTTP_Windows_Simulator/HTTP_Mutual_Auth/DemoTasks/MutualAuthHTTPExample.c
Unescape Escape View File

@ -128,6 +128,22 @@ struct NetworkContext
TlsTransportParams_t * pParams; TlsTransportParams_t * pParams;
}; };
/**
* @brief The maximum number of times to run the loop in this demo.
*
* @note The demo loop is attempted to re-run only if it fails in an iteration.
* Once the demo loop succeeds in an iteration, the demo exits successfully.
*/
#ifndef HTTP_MAX_DEMO_LOOP_COUNT
#define HTTP_MAX_DEMO_LOOP_COUNT ( 3 )
#endif
/**
* @brief Time in ticks to wait between retries of the demo loop if
* demo loop fails.
*/
#define DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS ( pdMS_TO_TICKS( 5000U ) )
/** /**
* @brief A buffer used in the demo for storing HTTP request headers and * @brief A buffer used in the demo for storing HTTP request headers and
* HTTP response headers and body. * HTTP response headers and body.
@ -215,6 +231,7 @@ static void prvHTTPDemoTask( void * pvParameters )
NetworkContext_t xNetworkContext = { 0 }; NetworkContext_t xNetworkContext = { 0 };
TlsTransportParams_t xTlsTransportParams = { 0 }; TlsTransportParams_t xTlsTransportParams = { 0 };
BaseType_t xIsConnectionEstablished = pdFALSE; BaseType_t xIsConnectionEstablished = pdFALSE;
UBaseType_t uxDemoRunCount = 0UL;
/* The user of this demo must check the logs for any failure codes. */ /* The user of this demo must check the logs for any failure codes. */
BaseType_t xDemoStatus = pdPASS; BaseType_t xDemoStatus = pdPASS;
@ -225,55 +242,83 @@ static void prvHTTPDemoTask( void * pvParameters )
/* Set the pParams member of the network context with desired transport. */ /* Set the pParams member of the network context with desired transport. */
xNetworkContext.pParams = &xTlsTransportParams; xNetworkContext.pParams = &xTlsTransportParams;
/**************************** Connect. ******************************/ /* This demo runs a single loop unless there are failures in the demo execution.
* In case of failures in the demo execution, demo loop will be retried for up to
/* Attempt to connect to the HTTP server. If connection fails, retry after a * HTTP_MAX_DEMO_LOOP_COUNT times. */
* timeout. The timeout value will be exponentially increased until either the do
* maximum number of attempts or the maximum timeout value is reached. The
* function returns pdFAIL if the TCP connection cannot be established with
* the broker after configured number of attempts. */
xDemoStatus = connectToServerWithBackoffRetries( prvConnectToServer,
&xNetworkContext );
if( xDemoStatus == pdPASS )
{ {
/* Set a flag indicating that a TLS connection exists. */ /**************************** Connect. ******************************/
xIsConnectionEstablished = pdTRUE;
/* Attempt to connect to the HTTP server. If connection fails, retry after a
/* Define the transport interface. */ * timeout. The timeout value will be exponentially increased until either the
xTransportInterface.pNetworkContext = &xNetworkContext; * maximum number of attempts or the maximum timeout value is reached. The
xTransportInterface.send = TLS_FreeRTOS_send; * function returns pdFAIL if the TCP connection cannot be established with
xTransportInterface.recv = TLS_FreeRTOS_recv; * the broker after configured number of attempts. */
} xDemoStatus = connectToServerWithBackoffRetries( prvConnectToServer,
else &xNetworkContext );
{
/* Log error to indicate connection failure after all if( xDemoStatus == pdPASS )
* reconnect attempts are over. */ {
LogError( ( "Failed to connect to HTTP server %.*s.", /* Set a flag indicating that a TLS connection exists. */
( int32_t ) AWS_IOT_ENDPOINT_LENGTH, xIsConnectionEstablished = pdTRUE;
democonfigAWS_IOT_ENDPOINT ) );
} /* Define the transport interface. */
xTransportInterface.pNetworkContext = &xNetworkContext;
/*********************** Send HTTP request.************************/ xTransportInterface.send = TLS_FreeRTOS_send;
xTransportInterface.recv = TLS_FreeRTOS_recv;
if( xDemoStatus == pdPASS ) }
{ else
xDemoStatus = prvSendHttpRequest( &xTransportInterface, {
HTTP_METHOD_POST, /* Log error to indicate connection failure after all
( sizeof( HTTP_METHOD_POST ) - 1 ), * reconnect attempts are over. */
democonfigPOST_PATH, LogError( ( "Failed to connect to HTTP server %.*s.",
( sizeof( democonfigPOST_PATH ) - 1 ) ); ( int32_t ) AWS_IOT_ENDPOINT_LENGTH,
} democonfigAWS_IOT_ENDPOINT ) );
}
/**************************** Disconnect. ******************************/
/*********************** Send HTTP request.************************/
/* Close the network connection to clean up any system resources that the
* demo may have consumed. */ if( xDemoStatus == pdPASS )
if( xIsConnectionEstablished == pdTRUE ) {
{ xDemoStatus = prvSendHttpRequest( &xTransportInterface,
/* Close the network connection. */ HTTP_METHOD_POST,
TLS_FreeRTOS_Disconnect( &xNetworkContext ); ( sizeof( HTTP_METHOD_POST ) - 1 ),
} democonfigPOST_PATH,
( sizeof( democonfigPOST_PATH ) - 1 ) );
}
/**************************** Disconnect. ******************************/
/* Close the network connection to clean up any system resources that the
* demo may have consumed. */
if( xIsConnectionEstablished == pdTRUE )
{
/* Close the network connection. */
TLS_FreeRTOS_Disconnect( &xNetworkContext );
}
/*********************** Retry in case of failure. ************************/
/* Increment the demo run count. */
uxDemoRunCount++;
if( xDemoStatus == pdPASS )
{
LogInfo( ( "Demo iteration %lu was successful.", uxDemoRunCount ) );
}
/* Attempt to retry a failed demo iteration for up to #HTTP_MAX_DEMO_LOOP_COUNT times. */
else if( uxDemoRunCount < HTTP_MAX_DEMO_LOOP_COUNT )
{
LogWarn( ( "Demo iteration %lu failed. Retrying...", uxDemoRunCount ) );
vTaskDelay( DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS );
}
/* Failed all #HTTP_MAX_DEMO_LOOP_COUNT demo iterations. */
else
{
LogError( ( "All %d demo iterations failed.", HTTP_MAX_DEMO_LOOP_COUNT ) );
break;
}
} while( xDemoStatus != pdPASS );
if( xDemoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {

133
FreeRTOS-Plus/Demo/coreHTTP_Windows_Simulator/HTTP_Plaintext/DemoTasks/PlainTextHTTPExample.c
Unescape Escape View File

@ -170,6 +170,22 @@ struct NetworkContext
PlaintextTransportParams_t * pParams; PlaintextTransportParams_t * pParams;
}; };
/**
* @brief The maximum number of times to run the loop in this demo.
*
* @note The demo loop is attempted to re-run only if it fails in an iteration.
* Once the demo loop succeeds in an iteration, the demo exits successfully.
*/
#ifndef HTTP_MAX_DEMO_LOOP_COUNT
#define HTTP_MAX_DEMO_LOOP_COUNT ( 3 )
#endif
/**
* @brief Time in ticks to wait between retries of the demo loop if
* demo loop fails.
*/
#define DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS ( pdMS_TO_TICKS( 5000U ) )
/** /**
* @brief A pair containing a path string of the URI and its length. * @brief A pair containing a path string of the URI and its length.
*/ */
@ -294,6 +310,7 @@ static void prvHTTPDemoTask( void * pvParameters )
}; };
BaseType_t xIsConnectionEstablished = pdFALSE; BaseType_t xIsConnectionEstablished = pdFALSE;
UBaseType_t uxHttpPathCount = 0U; UBaseType_t uxHttpPathCount = 0U;
UBaseType_t uxDemoRunCount = 0UL;
/* The user of this demo must check the logs for any failure codes. */ /* The user of this demo must check the logs for any failure codes. */
BaseType_t xDemoStatus = pdPASS; BaseType_t xDemoStatus = pdPASS;
@ -304,62 +321,90 @@ static void prvHTTPDemoTask( void * pvParameters )
/* Set the pParams member of the network context with desired transport. */ /* Set the pParams member of the network context with desired transport. */
xNetworkContext.pParams = &xPlaintextTransportParams; xNetworkContext.pParams = &xPlaintextTransportParams;
/**************************** Connect. ******************************/ /* This demo runs a single loop unless there are failures in the demo execution.
* In case of failures in the demo execution, demo loop will be retried for up to
* HTTP_MAX_DEMO_LOOP_COUNT times. */
do
{
/**************************** Connect. ******************************/
/* Attempt to connect to the HTTP server. If connection fails, retry after a /* Attempt to connect to the HTTP server. If connection fails, retry after a
* timeout. The timeout value will be exponentially increased until either the * timeout. The timeout value will be exponentially increased until either the
* maximum number of attempts or the maximum timeout value is reached. The * maximum number of attempts or the maximum timeout value is reached. The
* function returns pdFAIL if the TCP connection cannot be established with * function returns pdFAIL if the TCP connection cannot be established with
* the broker after configured number of attempts. */ * the broker after configured number of attempts. */
xDemoStatus = connectToServerWithBackoffRetries( prvConnectToServer, xDemoStatus = connectToServerWithBackoffRetries( prvConnectToServer,
&xNetworkContext ); &xNetworkContext );
if( xDemoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
/* Set a flag indicating that a TCP connection has been established. */ /* Set a flag indicating that a TCP connection has been established. */
xIsConnectionEstablished = pdTRUE; xIsConnectionEstablished = pdTRUE;
/* Define the transport interface. */ /* Define the transport interface. */
xTransportInterface.pNetworkContext = &xNetworkContext; xTransportInterface.pNetworkContext = &xNetworkContext;
xTransportInterface.send = Plaintext_FreeRTOS_send; xTransportInterface.send = Plaintext_FreeRTOS_send;
xTransportInterface.recv = Plaintext_FreeRTOS_recv; xTransportInterface.recv = Plaintext_FreeRTOS_recv;
} }
else else
{ {
/* Log error to indicate connection failure after all /* Log error to indicate connection failure after all
* reconnect attempts are over. */ * reconnect attempts are over. */
LogError( ( "Failed to connect to HTTP server %.*s.", LogError( ( "Failed to connect to HTTP server %.*s.",
( int32_t ) httpexampleSERVER_HOSTNAME_LENGTH, ( int32_t ) httpexampleSERVER_HOSTNAME_LENGTH,
democonfigSERVER_HOSTNAME ) ); democonfigSERVER_HOSTNAME ) );
} }
/*********************** Send HTTP request.************************/ /*********************** Send HTTP request.************************/
for( uxHttpPathCount = 0; uxHttpPathCount < httpexampleNUMBER_HTTP_PATHS; ++uxHttpPathCount )
{
if( xDemoStatus == pdPASS )
{
xDemoStatus = prvSendHttpRequest( &xTransportInterface,
xHttpMethods[ uxHttpPathCount ].pcHttpMethod,
xHttpMethods[ uxHttpPathCount ].ulHttpMethodLength,
xHttpMethodPaths[ uxHttpPathCount ].pcHttpPath,
xHttpMethodPaths[ uxHttpPathCount ].ulHttpPathLength );
}
else
{
break;
}
}
/**************************** Disconnect. ******************************/
/* Close the network connection to clean up any system resources that the
* demo may have consumed. */
if( xIsConnectionEstablished == pdTRUE )
{
/* Close the network connection. */
Plaintext_FreeRTOS_Disconnect( &xNetworkContext );
}
/*********************** Retry in case of failure. ************************/
/* Increment the demo run count. */
uxDemoRunCount++;
for( uxHttpPathCount = 0; uxHttpPathCount < httpexampleNUMBER_HTTP_PATHS; ++uxHttpPathCount )
{
if( xDemoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {
xDemoStatus = prvSendHttpRequest( &xTransportInterface, LogInfo( ( "Demo iteration %lu was successful.", uxDemoRunCount ) );
xHttpMethods[ uxHttpPathCount ].pcHttpMethod,
xHttpMethods[ uxHttpPathCount ].ulHttpMethodLength,
xHttpMethodPaths[ uxHttpPathCount ].pcHttpPath,
xHttpMethodPaths[ uxHttpPathCount ].ulHttpPathLength );
} }
/* Attempt to retry a failed demo iteration for up to #HTTP_MAX_DEMO_LOOP_COUNT times. */
else if( uxDemoRunCount < HTTP_MAX_DEMO_LOOP_COUNT )
{
LogWarn( ( "Demo iteration %lu failed. Retrying...", uxDemoRunCount ) );
vTaskDelay( DELAY_BETWEEN_DEMO_RETRY_ITERATIONS_TICKS );
}
/* Failed all #HTTP_MAX_DEMO_LOOP_COUNT demo iterations. */
else else
{ {
LogError( ( "All %d demo iterations failed.", HTTP_MAX_DEMO_LOOP_COUNT ) );
break; break;
} }
} } while( xDemoStatus != pdPASS );
/**************************** Disconnect. ******************************/
/* Close the network connection to clean up any system resources that the
* demo may have consumed. */
if( xIsConnectionEstablished == pdTRUE )
{
/* Close the network connection. */
Plaintext_FreeRTOS_Disconnect( &xNetworkContext );
}
if( xDemoStatus == pdPASS ) if( xDemoStatus == pdPASS )
{ {

2
FreeRTOS-Plus/Source/coreJSON

@ -1 +1 @@
Subproject commit 5bfadf5f86a038986ebdf7114939d73de2681f7f Subproject commit c3ed2a23a64390178f6332589aac6994cb944f48
Write Preview
Loading…
Cancel
Save