/*
 * FreeRTOS MQTT V2.1.1
 * Copyright (C) 2020 Amazon.com, Inc. or its affiliates.  All Rights Reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
 * the Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 *
 * http://aws.amazon.com/freertos
 * http://www.FreeRTOS.org
 */

/**
 * @file iot_test_mqtt_agent.c
 * @brief Tests for MQTT Agent.
 */

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

/* FreeRTOS Includes. */
#include "FreeRTOS.h"
#include "semphr.h"
#include "iot_mqtt_agent.h"
#include "task.h"
#include "queue.h"
#include "event_groups.h"
#include "aws_clientcredential.h"
#include "iot_mqtt.h"
#include "iot_init.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 * 6 )
#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. */
MQTTAgentConnectParams_t xDefaultConnectParameters =
{
    clientcredentialMQTT_BROKER_ENDPOINT,
    mqttagentREQUIRE_TLS,
    pdFALSE,
    clientcredentialMQTT_BROKER_PORT,
    NULL,
    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;
    uint8_t * pucTransmittedData;
    TaskHandle_t xHandle;
    StaticSemaphore_t xSemaphore;
    StaticTask_t xTaskBuffer;
    StackType_t xTaskStack[ mqttagenttestMULTI_TASK_TEST_TASKS_STACK_SIZE ];
} MQTTtestAgentTaskParams_t;
/* Information about tasks for the multitask test. */
static MQTTtestAgentTaskParams_t xTaskParams[ mqttagenttestMULTI_TASK_TEST_NUM_TASKS ] = { 0 };

/* 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 multitask test completion.*/
static StaticEventGroup_t xSyncEventGroup;

/**
 * @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 )
{
    xDefaultConnectParameters.pucClientId = ( uint8_t * ) clientcredentialIOT_THING_NAME;
    TEST_ASSERT_EQUAL( true, IotSdk_Init() );
    TEST_ASSERT_EQUAL( IOT_MQTT_SUCCESS, IotMqtt_Init() );
}
TEST_SETUP( Full_MQTT_Agent_Stress_Tests )
{
    xDefaultConnectParameters.pucClientId = ( uint8_t * ) clientcredentialIOT_THING_NAME;
    TEST_ASSERT_EQUAL( true, IotSdk_Init() );
    TEST_ASSERT_EQUAL( IOT_MQTT_SUCCESS, IotMqtt_Init() );
}
TEST_SETUP( Full_MQTT_Agent_ALPN )
{
    xDefaultConnectParameters.pucClientId = ( uint8_t * ) clientcredentialIOT_THING_NAME;
    TEST_ASSERT_EQUAL( true, IotSdk_Init() );
    TEST_ASSERT_EQUAL( IOT_MQTT_SUCCESS, IotMqtt_Init() );
}
/*-----------------------------------------------------------*/

/**
 * @brief Tear down function called after each test in this group is executed.
 */
TEST_TEAR_DOWN( Full_MQTT_Agent )
{
    IotMqtt_Cleanup();
    IotSdk_Cleanup();
}
TEST_TEAR_DOWN( Full_MQTT_Agent_Stress_Tests )
{
    IotMqtt_Cleanup();
    IotSdk_Cleanup();
}
TEST_TEAR_DOWN( Full_MQTT_Agent_ALPN )
{
    IotMqtt_Cleanup();
    IotSdk_Cleanup();
}
/*-----------------------------------------------------------*/

/**
 * @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 priorities.
 *
 * 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 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;
    uint32_t usEventMask;
    static uint8_t pucTransmittedData[ mqttagenttestMULTI_TASK_TEST_PUB_DATA_SIZE ];

    /* Initialize to unallocated. */
    for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
    {
        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. */
        if( xEventGroupCreateStatic( &xSyncEventGroup ) == NULL )
        {
            TEST_FAIL_MESSAGE( "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. */
        }

        /* 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. */
            if( xSemaphoreCreateCountingStatic( mqttagenttestMULTI_TASK_TEST_PUBLISH_PER_LOOP * mqttagenttestMULTI_TASK_TEST_NUM_TX_TASKS,
                                                0, &( xTaskParams[ usIndex ].xSemaphore ) ) == NULL )
            {
                TEST_FAIL_MESSAGE( "Failed to create receive task semaphore." );
            }

            /* 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 ].xHandle = xTaskCreateStatic( 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,
                                                                ( StackType_t * ) &( xTaskParams[ usIndex ].xTaskStack ),
                                                                &( xTaskParams[ usIndex ].xTaskBuffer ) );
            TEST_ASSERT_NOT_NULL_MESSAGE( xTaskParams[ usIndex ].xHandle, "Task creation failed" );
        }

        /* 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 ].xHandle = xTaskCreateStatic( 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,
                                                                ( StackType_t * ) &( xTaskParams[ usIndex ].xTaskStack ),
                                                                &( xTaskParams[ usIndex ].xTaskBuffer ) );
            TEST_ASSERT_NOT_NULL_MESSAGE( xTaskParams[ usIndex ].xHandle, "Task creation failed" );
        }

        /* When all task reaches the sync group, then the test is complete. */
        usEventMask = xEventGroupSync( ( EventGroupHandle_t ) &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 ].xHandle != NULL )
        {
            vTaskDelete( xTaskParams[ usIndex ].xHandle );
        }
    }

    for( usIndex = 0; usIndex < mqttagenttestMULTI_TASK_TEST_NUM_RX_TASKS; usIndex++ )
    {
        if( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx != NULL )
        {
            vEventGroupDelete( MQTTtestAgentMultiTestRxParam[ usIndex ].xSyncEventRx );
        }
    }
}

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

/**
 * @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 receive 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 = ( SemaphoreHandle_t ) &( 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( ( SemaphoreHandle_t ) &( 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( ( EventGroupHandle_t ) &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( ( EventGroupHandle_t ) &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 */
}
