/*
* Amazon FreeRTOS MQTT AFQP V1.1.4
* Copyright (C) 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*
* http://aws.amazon.com/freertos
* http://www.FreeRTOS.org
*/
/**
* @file aws_test_mqtt_agent.c
* @brief Tests for MQTT Agent.
*/
/* Standard includes. */
#include <stdio.h>
#include <string.h>
/* Amazon FreeRTOS includes. */
#include "FreeRTOS.h"
#include "semphr.h"
#include "aws_mqtt_agent.h"
#include "task.h"
#include "queue.h"
#include "event_groups.h"
#include "aws_clientcredential.h"
/* Unity framework includes. */
#include "unity_fixture.h"
/* MQTT agent connection timeout. */
#define mqttagenttestTIMEOUT pdMS_TO_TICKS( 10000UL )
/* Topic name for ping-pong testing. */
#define mqttagenttestTOPIC_NAME ( ( const uint8_t * ) "freertos/tests/echo" )
#define mqttagenttestMESSAGE "Hello from the test."
#define mqttagenttestFAILUREPRINTF( x ) vLoggingPrintf x
/* The parameters below are definable so the test can run on most target. */
/* Size in bytes of the messages published. */
#ifndef mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE
#define mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE ( 256 )
#endif
/* Number of Receive Tasks that are created. */
#ifndef mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS
#define mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS ( 2 )
#endif
/* Number of Transmit Tasks that are created. */
#ifndef mqttagenttestMULTI_TASK_TEST_NUM_TX_TASKS
#define mqttagenttestMULTI_TASK_TEST_NUM_TX_TASKS ( 2 )
#endif
/* Total number of tasks. */
#define mqttagenttestMULTI_TASK_TEST_NUM_TASKS ( mqttagenttestMULTI_TASK_TEST_NUM_TX_TASKS + mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS )
/* TCP Echo Client tasks multitask example parameters. */
#define mqttagenttestMULTI_TASK_TEST_TASKS_STACK_SIZE ( configMINIMAL_STACK_SIZE * 4 )
#define mqttagenttestMULTI_TASK_TEST_TASKS_PRIORITY ( tskIDLE_PRIORITY )
/* Timeout for sync between Transmit and Receive Tasks. */
#ifndef mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT
#define mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT ( 40000 )
#endif
/* This represents the time to wait to complete the multitask test.
* It will vary from one platform to another. */
#ifndef mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT
#define mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT ( 400000 )
#endif
/* Event mask used for the sync group. It represents all Tasks.*/
#define mqttagenttestMULTI_TASK_TEST_EVENT_MASK ( ( 1 << mqttagenttestMULTI_TASK_TEST_NUM_TASKS ) - 1 )
/* Event mask used for the sync group. It represents all Transmit Tasks. */
#define mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK ( ( ( 1 << mqttagenttestMULTI_TASK_TEST_NUM_TASKS ) - 1 ) - ( ( 1 << mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS ) - 1 ) )
#define mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT_TICKS ( pdMS_TO_TICKS( mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT ) )
#define mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT_TICKS ( pdMS_TO_TICKS( mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT ) )
/* Receive task priority is changed on bit 0, while Transmit is changed on bit 1.
* Loop needs to be at least 4 to cover every cases. */
#define mqttagenttestMULTI_TASK_TEST_MAX_LOOP ( 4 )
#define mqttagenttestMULTI_TASK_TEST_LOW_PRIORITY tskIDLE_PRIORITY
#define mqttagenttestMULTI_TASK_TEST_HIGH_PRIORITY ( configMAX_PRIORITIES - 1 )
/* Receive task toggle priority on bit 0 while Transmit does on bit 1. This allows covering all possible scenario of low/high priority. */
#define mqttagenttestMULTI_TASK_TEST_RX_LOWP_SWITCH_MASK ( 0 )
#define mqttagenttestMULTI_TASK_TEST_TX_LOWP_SWITCH_MASK ( 1 )
/* Every Receive task will create one client. */
#define mqttagenttestMULTI_TASK_TEST_CLIENT_ID "ClientId%d"
#define mqttagenttestMULTI_TASK_TEST_MAX_CLIENT_ID_SIZE ( 14 )
/* Number of publishing made by the Transmit task every loop. */
#define mqttagenttestMULTI_TASK_TEST_PUBLISH_PER_LOOP ( 10 )
/* Topic name for ping-pong testing. */
#define mqttagenttestMULTI_TASK_TEST_TOPIC_NAME ( ( const uint8_t * ) "freertos/tests/multiTask/%d" )
#define mqttagenttestMULTI_TASK_TEST_MAX_TOPIC_NAME_SIZE ( 30 )
/* Default connection parameters. */
static const MQTTAgentConnectParams_t xDefaultConnectParameters =
{
clientcredentialMQTT_BROKER_ENDPOINT,
mqttagentREQUIRE_TLS,
pdFALSE,
clientcredentialMQTT_BROKER_PORT,
( const uint8_t * ) clientcredentialIOT_THING_NAME,
0,
pdTRUE,
NULL,
NULL,
NULL,
0
};
/* Parameters used in Muti task call back function. Data is checked and status is passed down. */
typedef struct
{
SemaphoreHandle_t xSemaphore;
BaseType_t xStatus;
} MQTTtestAgentCbParam_t;
/* Task parameters for multitask test. */
typedef struct
{
uint16_t usTaskTag;
BaseType_t xStatus;
TaskHandle_t xTaskHandle;
uint8_t * pucTransmittedData;
SemaphoreHandle_t xSemaphore;
BaseType_t xTaskCreated;
} MQTTtestAgentTaskParams_t;
/* Parameters for the Receive tasks. They are bundles into one structure to improve readability. */
typedef struct
{
MQTTAgentHandle_t xMQTTHandle;
EventGroupHandle_t xSyncEventRx;
uint8_t cTopic[ mqttagenttestMULTI_TASK_TEST_MAX_TOPIC_NAME_SIZE ];
uint16_t usTopicLength;
uint8_t cClientID[ mqttagenttestMULTI_TASK_TEST_MAX_CLIENT_ID_SIZE ];
uint16_t usClientIDLength;
} MQTTtestAgentMultiTestRxParam_t;
/* Bundle of global static used by Receive tasks. Note that Transmit task will also use the mqtt handle and the sync semaphore. */
static MQTTtestAgentMultiTestRxParam_t MQTTtestAgentMultiTestRxParam[ mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS ];
static void prvMultiTaskTest_Rx_Task( void * pvParameters );
static void prvMultiTaskTest_Tx_Task( void * pvParameters );
/* The event group used to wait for the multitaks test completion.*/
static EventGroupHandle_t xSyncEventGroup = NULL;
/**
* @brief Callback for MQTT subscription.
*/
static MQTTBool_t prvMultiTaskTestMQTTCallback( void * pvUserData,
const MQTTPublishData_t * const pxPublishParameters )
{
MQTTtestAgentCbParam_t * pxMQTTtestAgentCbParam;
BaseType_t xDataIndex;
pxMQTTtestAgentCbParam = ( MQTTtestAgentCbParam_t * ) pvUserData;
pxMQTTtestAgentCbParam->xStatus = pdPASS;
if( pxPublishParameters->ulDataLength != mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE )
{
pxMQTTtestAgentCbParam->xStatus = pdFAIL;
}
else
{
for( xDataIndex = 0; xDataIndex < mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE; xDataIndex++ )
{
if( ( ( uint8_t * ) ( pxPublishParameters->pvData ) )[ xDataIndex ] != xDataIndex )
{
pxMQTTtestAgentCbParam->xStatus = pdFAIL;
break;
}
}
}
/* Give the semaphore to signal message receipt. */
xSemaphoreGive( pxMQTTtestAgentCbParam->xSemaphore );
return eMQTTFalse;
}
/**
* @brief Callback for MQTT subscription.
*/
static MQTTBool_t prvMQTTCallback( void * pvUserData,
const MQTTPublishData_t * const pxPublishParameters )
{
/* Give the semaphore to signal message receipt. */
xSemaphoreGive( ( SemaphoreHandle_t ) pvUserData );
return eMQTTFalse;
}
/*-----------------------------------------------------------*/
/**
* @brief Test helper routine for MQTT connect, subcribe, publish, and
* tear-down.
*/
static BaseType_t prvCreateClientAndConnectToBroker( BaseType_t xUseAlpn )
{
MQTTAgentReturnCode_t xReturned = eMQTTAgentFailure;
StaticSemaphore_t xSemaphore = { 0 };
MQTTAgentHandle_t xMQTTHandle = NULL;
MQTTAgentSubscribeParams_t xSubscribeParams;
MQTTAgentPublishParams_t xPublishParameters;
BaseType_t xMQTTAgentCreated = pdFALSE;
MQTTAgentConnectParams_t xConnectParameters;
memcpy( &xConnectParameters, &xDefaultConnectParameters, sizeof( MQTTAgentConnectParams_t ) );
/* Initialize the semaphore as unavailable. */
TEST_ASSERT_NOT_NULL( xSemaphoreCreateCountingStatic( 1, 0, &xSemaphore ) );
/* Fill in the MQTTAgentConnectParams_t member that is not const. */
xConnectParameters.usClientIdLength = ( uint16_t ) strlen(
( char * ) xConnectParameters.pucClientId );
/* Switch ports if requested. */
if( pdTRUE == xUseAlpn )
{
xConnectParameters.xFlags |= mqttagentUSE_AWS_IOT_ALPN_443;
}
if( TEST_PROTECT() )
{
/* The MQTT client object must be created before it can be used. */
xReturned = MQTT_AGENT_Create( &xMQTTHandle );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
xMQTTAgentCreated = pdTRUE;
/* Connect to the broker. */
xReturned = MQTT_AGENT_Connect( xMQTTHandle,
&xConnectParameters,
mqttagenttestTIMEOUT );
TEST_ASSERT_EQUAL_INT_MESSAGE( xReturned, eMQTTAgentSuccess, "Failed to connect to the MQTT broker with MQTT_AGENT_Connect()." );
/* Setup subscribe parameters to subscribe to echo topic. */
xSubscribeParams.pucTopic = mqttagenttestTOPIC_NAME;
xSubscribeParams.pvPublishCallbackContext = &xSemaphore;
xSubscribeParams.pxPublishCallback = prvMQTTCallback;
xSubscribeParams.usTopicLength = ( uint16_t ) strlen( ( const char * ) mqttagenttestTOPIC_NAME );
xSubscribeParams.xQoS = eMQTTQoS1;
/* Subscribe to the topic. */
xReturned = MQTT_AGENT_Subscribe( xMQTTHandle,
&xSubscribeParams,
mqttagenttestTIMEOUT );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
/* Setup the publish parameters. */
memset( &( xPublishParameters ), 0x00, sizeof( xPublishParameters ) );
xPublishParameters.pucTopic = mqttagenttestTOPIC_NAME;
xPublishParameters.pvData = mqttagenttestMESSAGE;
xPublishParameters.usTopicLength = ( uint16_t ) strlen( ( const char * ) mqttagenttestTOPIC_NAME );
xPublishParameters.ulDataLength = ( uint32_t ) strlen( mqttagenttestMESSAGE );
xPublishParameters.xQoS = eMQTTQoS1;
/* Publish the message. */
xReturned = MQTT_AGENT_Publish( xMQTTHandle,
&( xPublishParameters ),
mqttagenttestTIMEOUT );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
/* Take the semaphore to ensure the message is Received. */
if( pdFALSE == xSemaphoreTake( ( QueueHandle_t ) &( xSemaphore ), mqttagenttestTIMEOUT ) )
{
TEST_FAIL();
}
/* Disconnect the client. */
xReturned = MQTT_AGENT_Disconnect( xMQTTHandle, mqttagenttestTIMEOUT );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
}
else
{
TEST_FAIL();
}
/*Don't forget to reset the flag, since connect parameters are global, all test afterwards would use ALPN. */
xConnectParameters.xFlags &= ~mqttagentUSE_AWS_IOT_ALPN_443;
if( xMQTTAgentCreated == pdTRUE )
{
/* Delete the MQTT client. */
xReturned = MQTT_AGENT_Delete( xMQTTHandle );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
}
return eMQTTAgentSuccess == xReturned;
}
/*-----------------------------------------------------------*/
/* Define Test Group. */
TEST_GROUP( Full_MQTT_Agent );
TEST_GROUP( Full_MQTT_Agent_ALPN );
TEST_GROUP( Full_MQTT_Agent_Stress_Tests );
/*-----------------------------------------------------------*/
/**
* @brief Setup function called before each test in this group is executed.
*/
TEST_SETUP( Full_MQTT_Agent )
{
}
TEST_SETUP( Full_MQTT_Agent_Stress_Tests )
{
}
TEST_SETUP( Full_MQTT_Agent_ALPN )
{
}
/*-----------------------------------------------------------*/
/**
* @brief Tear down function called after each test in this group is executed.
*/
TEST_TEAR_DOWN( Full_MQTT_Agent )
{
}
TEST_TEAR_DOWN( Full_MQTT_Agent_Stress_Tests )
{
}
TEST_TEAR_DOWN( Full_MQTT_Agent_ALPN )
{
}
/*-----------------------------------------------------------*/
/**
* @brief Function to define which tests to execute as part of this group.
*/
TEST_GROUP_RUNNER( Full_MQTT_Agent )
{
RUN_TEST_CASE( Full_MQTT_Agent, AFQP_MQTT_Agent_SubscribePublishDefaultPort );
RUN_TEST_CASE( Full_MQTT_Agent, AFQP_MQTT_Agent_InvalidCredentials );
}
TEST_GROUP_RUNNER( Full_MQTT_Agent_Stress_Tests )
{
RUN_TEST_CASE( Full_MQTT_Agent_Stress_Tests, MQTT_Agent_MultiTaskTest );
}
TEST_GROUP_RUNNER( Full_MQTT_Agent_ALPN )
{
RUN_TEST_CASE( Full_MQTT_Agent_ALPN, MQTT_Agent_SubscribePublishAlpn );
}
/*-----------------------------------------------------------*/
/* Test for ping-ponging a message using the default AWS IoT port for MQTT. */
TEST( Full_MQTT_Agent, AFQP_MQTT_Agent_SubscribePublishDefaultPort )
{
prvCreateClientAndConnectToBroker( pdFALSE );
}
/* Test for ping-ponging a message using the default AWS IoT port for MQTT. */
TEST( Full_MQTT_Agent, AFQP_MQTT_Agent_InvalidCredentials )
{
MQTTAgentReturnCode_t xReturned;
MQTTAgentHandle_t xMQTTHandle = NULL;
BaseType_t xClientCreated, xClientConnected;
MQTTAgentConnectParams_t xConnectParameters;
memcpy( &xConnectParameters, &xDefaultConnectParameters, sizeof( MQTTAgentConnectParams_t ) );
/* Initialize the semaphore. */
if( TEST_PROTECT() )
{
xClientCreated = pdFALSE;
/* Arbitrary invalid credentials. */
xConnectParameters.pcCertificate = "abcd";
xConnectParameters.ulCertificateSize = ( uint16_t ) strlen( xConnectParameters.pcCertificate );
/* Fill in the MQTTAgentConnectParams_t member that is not const. */
xConnectParameters.usClientIdLength = ( uint16_t ) strlen(
( char * ) xConnectParameters.pucClientId );
/* The MQTT client object must be created before it can be used. */
xReturned = MQTT_AGENT_Create( &xMQTTHandle );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
xClientCreated = pdTRUE;
/* Connect to the broker. */
xClientConnected = pdTRUE;
xReturned = MQTT_AGENT_Connect( xMQTTHandle,
&xConnectParameters,
mqttagenttestTIMEOUT );
TEST_ASSERT_NOT_EQUAL( xReturned, eMQTTAgentSuccess );
xClientConnected = pdFALSE;
}
/* Connect to the broker. */
if( xClientConnected == pdTRUE )
{
/* If enter here, test has already failed. */
xReturned = MQTT_AGENT_Disconnect( xMQTTHandle,
mqttagenttestTIMEOUT );
if( xReturned != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Could not disconnect client.\r\n", __FUNCTION__ ) );
}
}
if( xClientCreated == pdTRUE )
{
/* Delete the MQTT client. */
xReturned = MQTT_AGENT_Delete( xMQTTHandle );
TEST_ASSERT_EQUAL_INT( xReturned, eMQTTAgentSuccess );
}
}
/*-----------------------------------------------------------*/
/* Test for ping-ponging a message using AWS IoT MQTT broker support for port 443. */
TEST( Full_MQTT_Agent_ALPN, MQTT_Agent_SubscribePublishAlpn )
{
prvCreateClientAndConnectToBroker( pdTRUE );
}
/*-----------------------------------------------------------*/
/**
* @brief multitask test for MQTT
*
* This test create x number of Receive tasks that will create x different clients and
* subscribe to x different topic.
* The Receive tasks will take care of create, connecting, subscribing, disconnecting and deleting clients.
* Also, y number of Transmit task are going to be created, each tasks are going to publish n number of MQTT messages
* to each one of the x tasks.
* Once each y Transmit tasks have submitted n messages to each of the x Receive task, it starts all over again. What varies from cycle to
* cycle are the priorities. Receive and Transmit tasks will swich priorities in such a way as to cover all possible combination of
* high and low prioritie.
*
* At the end of the test, Tasks do not self delete. They get suspended. The main test function will first wait for every
* task to complete through Amazon FreeRTOS synchronization mechanism then it will delete all the created tasks.
*
* For simplicity sake, memory is not allocated in the Receive or Transmit tasks. The allocation is centralized in one place, the main task.
*/
TEST( Full_MQTT_Agent_Stress_Tests, MQTT_Agent_MultiTaskTest )
{
uint16_t usIndex;
MQTTtestAgentTaskParams_t xTaskParams[ mqttagenttestMULTI_TASK_TEST_NUM_TASKS ];
uint32_t usEventMask;
BaseType_t xResult;
uint8_t * pucTransmittedData;
xSyncEventGroup = NULL;
pucTransmittedData = NULL;
/* Initialize to unallocated. */
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
xTaskParams[ usIndex ].xSemaphore = NULL;
MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx = NULL;
}
if( TEST_PROTECT() )
{
/* Create the event group used by the Transmit and Receive tasks to synchronize prior
* to commencing a cycle using a new socket. */
xSyncEventGroup = xEventGroupCreate();
TEST_ASSERT_NOT_NULL_MESSAGE( xSyncEventGroup, "Couldn't create Sync event." );
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx = xEventGroupCreate();
TEST_ASSERT_NOT_NULL_MESSAGE( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx, "Couldn't create Sync event." );
/* Flag memory as allocated for later Free. */
}
pucTransmittedData = pvPortMalloc( mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE * sizeof( uint8_t ) );
TEST_ASSERT_NOT_NULL_MESSAGE( pucTransmittedData, "Couldn't allocate data." );
/* Data sent by Transmit tasks. And a counter to check integrity of the data when Received by Receive task. */
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE; usIndex++ )
{
pucTransmittedData[ usIndex ] = ( uint8_t ) ( usIndex & 0xFF );
}
/* Create Receive tasks. */
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
xTaskParams[ usIndex ].usTaskTag = usIndex;
xTaskParams[ usIndex ].xStatus = pdFAIL;
xTaskParams[ usIndex ].pucTransmittedData = pucTransmittedData;
/* Each Receive task will Receive mqttagenttestMULTI_TASK_TEST_PUBLISH_PER_LOOP * mqttagenttestMULTI_TASK_TEST_NUM_TX_TASKS messages.
* It is possible that all of them may be Received before callback can trigger, so make semaphore count enough. */
xTaskParams[ usIndex ].xSemaphore = xSemaphoreCreateCounting( mqttagenttestMULTI_TASK_TEST_PUBLISH_PER_LOOP * mqttagenttestMULTI_TASK_TEST_NUM_TX_TASKS, 0 );
TEST_ASSERT_NOT_NULL( xTaskParams[ usIndex ].xSemaphore );
/* Create topic for each Receive task. */
MQTTtestAgentMultiTestRxParam[ usIndex ].usTopicLength =
snprintf( ( char * ) &MQTTtestAgentMultiTestRxParam[ usIndex ].cTopic,
mqttagenttestMULTI_TASK_TEST_MAX_TOPIC_NAME_SIZE, /* Limit size, will vary depending of the number of tasks. */
( char * ) mqttagenttestMULTI_TASK_TEST_TOPIC_NAME,
usIndex );
/* Create client ID for each Receive task. */
MQTTtestAgentMultiTestRxParam[ usIndex ].usClientIDLength =
snprintf( ( char * ) &MQTTtestAgentMultiTestRxParam[ usIndex ].cClientID,
mqttagenttestMULTI_TASK_TEST_MAX_CLIENT_ID_SIZE, /* Limit size, will vary depending of the number of tasks. */
( char * ) mqttagenttestMULTI_TASK_TEST_CLIENT_ID,
usIndex );
/* Create Receive Task. When created, flag it for later deletion. */
xTaskParams[ usIndex ].xTaskCreated = pdFALSE;
xResult = xTaskCreate( prvMultiTaskTest_Rx_Task, /* The function that implements the task. */
"ClientTask", /* Just a text name for the task to aid debugging. */
mqttagenttestMULTI_TASK_TEST_TASKS_STACK_SIZE,
&( xTaskParams[ usIndex ] ),
mqttagenttestMULTI_TASK_TEST_TASKS_PRIORITY,
&( xTaskParams[ usIndex ].xTaskHandle ) );
TEST_ASSERT_EQUAL_MESSAGE( pdPASS, xResult, "Task creation failed" );
/* Flag Task as created for later delete. */
xTaskParams[ usIndex ].xTaskCreated = pdTRUE;
}
/* Create Transmit tasks. */
for( usIndex = mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_TASKS; usIndex++ )
{
xTaskParams[ usIndex ].usTaskTag = usIndex;
xTaskParams[ usIndex ].xStatus = pdFAIL;
xTaskParams[ usIndex ].pucTransmittedData = pucTransmittedData;
xTaskParams[ usIndex ].xTaskCreated = pdFALSE;
xResult = xTaskCreate( prvMultiTaskTest_Tx_Task, /* The function that implements the task. */
"ClientTask", /* Just a text name for the task to aid debugging. */
mqttagenttestMULTI_TASK_TEST_TASKS_STACK_SIZE,
&( xTaskParams[ usIndex ] ),
mqttagenttestMULTI_TASK_TEST_TASKS_PRIORITY,
&( xTaskParams[ usIndex ].xTaskHandle ) );
TEST_ASSERT_EQUAL_MESSAGE( pdPASS, xResult, "Task creation failed" );
/* Flag Task as created for later delete. */
xTaskParams[ usIndex ].xTaskCreated = pdTRUE;
}
/* When all task reaches the sync group, then the test is complete. */
usEventMask = xEventGroupSync( xSyncEventGroup, /* The event group used for the rendezvous. */
0,
mqttagenttestMULTI_TASK_TEST_EVENT_MASK,
mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT_TICKS );
TEST_ASSERT_EQUAL_MESSAGE( mqttagenttestMULTI_TASK_TEST_EVENT_MASK, usEventMask, "Timeout waiting for tasks" );
/* Check which tasks have failed. If one failed, the test is failed. */
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_TASKS; usIndex++ )
{
if( xTaskParams[ usIndex ].xStatus != pdPASS )
{
TEST_FAIL_MESSAGE( "Not all tasks completed successfully." );
}
}
}
/* The code below will delete all the memory that has been successfully allocated. */
/* For each task not completed, delete the task. */
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_TASKS; usIndex++ )
{
/* Check task has been created before deleting it. */
if( xTaskParams[ usIndex ].xTaskCreated == pdTRUE )
{
vTaskDelete( xTaskParams[ usIndex ].xTaskHandle );
}
}
/* Delete global group event. */
if( xSyncEventGroup != NULL )
{
vEventGroupDelete( xSyncEventGroup );
}
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
if( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx != NULL )
{
vEventGroupDelete( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx );
}
if( xTaskParams[ usIndex ].xSemaphore != NULL )
{
vSemaphoreDelete( xTaskParams[ usIndex ].xSemaphore );
}
}
if( pucTransmittedData != NULL )
{
vPortFree( pucTransmittedData );
}
}
/*-----------------------------------------------------------*/
/**
* @brief Receive Task
*
* The Receive tasks will take care of create, connecting, subscribing, disconnecting and deleting clients.
* Also, y number of Transmit task are going to be created, each tasks are going to publish n number of MQTT messages
* to each one of the x tasks.
*
* Once the task has connected, it will synch with the Transmit tasks. Since the task will receice y messages (from y Transmit task),
* it will pend on Receive semaphore y times. Data is a simple counter and is checked in the callback.
*/
static void prvMultiTaskTest_Rx_Task( void * pvParameters )
{
MQTTAgentSubscribeParams_t xSubscribeParams;
MQTTAgentUnsubscribeParams_t xUnSubscribeParams;
BaseType_t xResult, xStatus, xAgentCreated = pdFALSE;
size_t xRecvLoop, xPubSubIndex;
MQTTtestAgentTaskParams_t * pxTcptestEchoClientsTaskParams;
EventGroupHandle_t * pxSyncEventRx;
MQTTAgentHandle_t * pxMQTTHandle;
MQTTAgentConnectParams_t xConnectParametersMultiTaskTest;
MQTTtestAgentCbParam_t MQTTtestAgentCbParam;
uint16_t usTaskTag;
/* Copy over default parameters, then change for specific clientID connection. */
memcpy( &xConnectParametersMultiTaskTest, &xDefaultConnectParameters, sizeof( MQTTAgentConnectParams_t ) );
pxTcptestEchoClientsTaskParams = ( ( MQTTtestAgentTaskParams_t * ) pvParameters );
usTaskTag = pxTcptestEchoClientsTaskParams->usTaskTag;
pxSyncEventRx = &MQTTtestAgentMultiTestRxParam[ usTaskTag ].xSyncEventRx;
pxMQTTHandle = &MQTTtestAgentMultiTestRxParam[ usTaskTag ].xMQTTHandle;
xConnectParametersMultiTaskTest.pucClientId = MQTTtestAgentMultiTestRxParam[ usTaskTag ].cClientID;
xConnectParametersMultiTaskTest.usClientIdLength = MQTTtestAgentMultiTestRxParam[ usTaskTag ].usClientIDLength;
/* Setup subscribe parameters to subscribe to echo topic. */
xSubscribeParams.pvPublishCallbackContext = &MQTTtestAgentCbParam;
xSubscribeParams.pxPublishCallback = prvMultiTaskTestMQTTCallback;
xSubscribeParams.xQoS = eMQTTQoS1;
xSubscribeParams.pucTopic = MQTTtestAgentMultiTestRxParam[ usTaskTag ].cTopic;
xSubscribeParams.usTopicLength = MQTTtestAgentMultiTestRxParam[ usTaskTag ].usTopicLength;
xUnSubscribeParams.pucTopic = xSubscribeParams.pucTopic;
xUnSubscribeParams.usTopicLength = xSubscribeParams.usTopicLength;
MQTTtestAgentCbParam.xSemaphore = pxTcptestEchoClientsTaskParams->xSemaphore;
/* Default is pass to avoid nesting. */
xStatus = pdPASS;
for( xRecvLoop = 0; xRecvLoop < mqttagenttestMULTI_TASK_TEST_MAX_LOOP; xRecvLoop++ )
{
if( mqttagenttestMULTI_TASK_TEST_RX_LOWP_SWITCH_MASK & xRecvLoop )
{
vTaskPrioritySet( NULL, mqttagenttestMULTI_TASK_TEST_LOW_PRIORITY );
}
else
{
vTaskPrioritySet( NULL, mqttagenttestMULTI_TASK_TEST_HIGH_PRIORITY );
}
/* The MQTT client object must be created before it can be used. */
xResult = MQTT_AGENT_Create( pxMQTTHandle );
if( xResult != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Could not create MQTT client.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
xAgentCreated = pdTRUE;
/* Connect to the broker. */
xResult = MQTT_AGENT_Connect( *pxMQTTHandle,
&xConnectParametersMultiTaskTest,
mqttagenttestTIMEOUT );
if( xResult != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Could not connect MQTT client.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
/* Subscribe to the topic. */
xResult = MQTT_AGENT_Subscribe( *pxMQTTHandle,
&xSubscribeParams,
mqttagenttestTIMEOUT );
if( xResult != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Could not subscribe to topic.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
if( xEventGroupSync( *pxSyncEventRx, /* The event group used for the rendezvous. */
( 1 << usTaskTag ),
( 1 << usTaskTag ) | mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK,
mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT_TICKS ) != ( ( 1 << usTaskTag ) | mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK ) )
{
mqttagenttestFAILUREPRINTF( ( "%s: Timed out waiting for Transmit tasks.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
/* Take the semaphore to ensure the message is Received. */
for( xPubSubIndex = 0; xPubSubIndex < mqttagenttestMULTI_TASK_TEST_PUBLISH_PER_LOOP * mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; xPubSubIndex++ )
{
if( pdFALSE == xSemaphoreTake( pxTcptestEchoClientsTaskParams->xSemaphore, mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT_TICKS ) )
{
mqttagenttestFAILUREPRINTF( ( "%s: Timed out waiting for pub from Transmit tasks.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
else /* Get status from the callback*/
{
if( MQTTtestAgentCbParam.xStatus != pdPASS )
{
mqttagenttestFAILUREPRINTF( ( "%s: Error, corrupted data Received.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
}
}
/*If fail happened in for loop above, break out. */
if( xStatus == pdFAIL )
{
break;
}
if( xEventGroupSync( *pxSyncEventRx, /* The event group used for the rendezvous. */
( 1 << usTaskTag ),
( 1 << usTaskTag ) | mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK,
mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT_TICKS ) != ( ( 1 << usTaskTag ) | mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK ) )
{
mqttagenttestFAILUREPRINTF( ( "%s: Timed out waiting for Transmit tasks.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
/* Unsubscribe to the topic. */
xResult = MQTT_AGENT_Unsubscribe( *pxMQTTHandle,
&xUnSubscribeParams,
mqttagenttestTIMEOUT );
if( xResult != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Failed to unsubscribe.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
/* Disconnect. */
xResult = MQTT_AGENT_Disconnect( *pxMQTTHandle, mqttagenttestTIMEOUT );
if( xResult != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Failed to disconnect.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
xResult = MQTT_AGENT_Delete( *pxMQTTHandle );
if( xResult != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Failed to delete MQTT client.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
xAgentCreated = pdFALSE;
}
if( xAgentCreated == pdTRUE )
{
( void ) MQTT_AGENT_Delete( *pxMQTTHandle );
}
pxTcptestEchoClientsTaskParams->xStatus = xStatus;
xEventGroupSync( xSyncEventGroup, /* The event group used for the rendezvous. */
( 1 << usTaskTag ),
mqttagenttestMULTI_TASK_TEST_EVENT_MASK,
mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT_TICKS );
vTaskSuspend( NULL ); /* Suspend this task. Will be deleted by the mainTask */
}
/*-----------------------------------------------------------*/
/**
* @brief Transmit Task
*
* y number of Transmit task are going to be created, each tasks are going to publish n number of MQTT messages
* to each one of the x tasks.
*/
static void prvMultiTaskTest_Tx_Task( void * pvParameters )
{
size_t xSendLoop;
BaseType_t xStatus, xReturn;
size_t xPubSubIndex, usIndex;
MQTTAgentPublishParams_t xPublishParameters;
MQTTtestAgentTaskParams_t * xTcptestEchoClientsTaskParams;
xTcptestEchoClientsTaskParams = ( ( MQTTtestAgentTaskParams_t * ) pvParameters );
xStatus = pdPASS;
/* Transmit and Receive loop switch priorities on a different bit, so all combination of high and low priorities will be covered. */
for( xSendLoop = 1; xSendLoop < mqttagenttestMULTI_TASK_TEST_MAX_LOOP + 1; xSendLoop++ )
{
if( mqttagenttestMULTI_TASK_TEST_TX_LOWP_SWITCH_MASK & xSendLoop )
{
vTaskPrioritySet( NULL, mqttagenttestMULTI_TASK_TEST_LOW_PRIORITY );
}
else
{
vTaskPrioritySet( NULL, mqttagenttestMULTI_TASK_TEST_HIGH_PRIORITY );
}
/* Wait for the Receive tasks to create and connect the agent. */
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
if( xEventGroupSync( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx, /* The event group used for the rendezvous. */
( 1 << xTcptestEchoClientsTaskParams->usTaskTag ), /* The bit representing the Transmit task reaching the rendezvous. */
( mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK | ( 1 << usIndex ) ), /* Also wait for the Receive tasks. */
mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT_TICKS ) != ( mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK | ( 1 << usIndex ) ) )
{
mqttagenttestFAILUREPRINTF( ( "%s: Failed synching with Transmit and Receive tasks.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
}
if( xStatus == pdFAIL )
{
break;
}
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
/* Setup the publish parameters. */
memset( &( xPublishParameters ), 0x00, sizeof( xPublishParameters ) );
xPublishParameters.pucTopic = MQTTtestAgentMultiTestRxParam[ usIndex ].cTopic;
xPublishParameters.pvData = xTcptestEchoClientsTaskParams->pucTransmittedData;
xPublishParameters.usTopicLength = MQTTtestAgentMultiTestRxParam[ usIndex ].usTopicLength;
xPublishParameters.ulDataLength = ( uint32_t ) mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE;
xPublishParameters.xQoS = eMQTTQoS1;
/* Publish the message. */
for( xPubSubIndex = 0; xPubSubIndex < mqttagenttestMULTI_TASK_TEST_PUBLISH_PER_LOOP; xPubSubIndex++ )
{
xReturn = MQTT_AGENT_Publish( MQTTtestAgentMultiTestRxParam[ usIndex ].xMQTTHandle,
&( xPublishParameters ),
mqttagenttestTIMEOUT );
if( xReturn != eMQTTAgentSuccess )
{
mqttagenttestFAILUREPRINTF( ( "%s: Failed publishing to topic.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
}
if( xStatus == pdFAIL )
{
break;
}
}
if( xStatus == pdFAIL )
{
break;
}
for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
{
if( xEventGroupSync( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx, /* The event group used for the rendezvous. */
( 1 << xTcptestEchoClientsTaskParams->usTaskTag ), /* The bit representing the Transmit task reaching the rendezvous. */
( mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK | ( 1 << usIndex ) ), /* Also wait for the Receive tasks. */
mqttagenttestMULTI_TASK_TEST_SYNC_TIMEOUT_TICKS ) != ( mqttagenttestMULTI_TASK_TEST_TX_EVENT_MASK | ( 1 << usIndex ) ) )
{
mqttagenttestFAILUREPRINTF( ( "%s: Failed synching with Transmit and Receive tasks.\r\n", __FUNCTION__ ) );
xStatus = pdFAIL;
break;
}
}
}
xTcptestEchoClientsTaskParams->xStatus = xStatus;
xEventGroupSync( xSyncEventGroup, /* The event group used for the rendezvous. */
( 1 << xTcptestEchoClientsTaskParams->usTaskTag ),
mqttagenttestMULTI_TASK_TEST_EVENT_MASK,
mqttagenttestMULTI_TASK_TEST_COMPLETE_TIMEOUT_TICKS );
vTaskSuspend( NULL ); /* Suspend this task. Will be deleted by the mainTask */
}