coreMQTT 演示（相互身份验证）

static void prvMQTTDemoTask( void * pvParameters )

{

    uint32_t ulPublishCount = 0U, ulTopicCount = 0U;

    const uint32_t ulMaxPublishCount = 5UL;

    NetworkContext_t xNetworkContext = { 0 };

    NetworkCredentials_t xNetworkCredentials = { 0 };

    MQTTContext_t xMQTTContext = { 0 };

    MQTTStatus_t xMQTTStatus;

    TlsTransportStatus_t xNetworkStatus;



    /* Remove compiler warnings about unused parameters. */

    ( void ) pvParameters;



    /* Set the entry time of the demo application. This entry time will be used


     * to calculate relative time elapsed in the execution of the demo 


     * application, by the timer utility function that is provided to the MQTT 


     * library. */

    ulGlobalEntryTimeMs = prvGetTimeMs();



    for( ; ; )

    {

        /************************** Connect. ****************************/





        /* Attempt to establish TLS session with MQTT broker. If connection 


         * fails, retry after a timeout. Timeout value will be exponentially 


         * increased until  the maximum number of attempts are reached or the 


         * maximum timeout value is reached. The function returns a failure 


         * status if the TCP connection cannot be established to the broker 


         * after the configured number of attempts. */

        xNetworkStatus = prvConnectToServerWithBackoffRetries( 

                                                     &xNetworkCredentials,

                                                     &xNetworkContext );

        configASSERT( xNetworkStatus == TLS_TRANSPORT_SUCCESS );



        /* Sends an MQTT Connect packet over the already established TLS 


        connection and waits for connection acknowledgment (CONNACK) packet. */

        

        LogInfo( ( "Creating an MQTT connection to %s.\r\n", democonfigMQTT_BROKER_ENDPOINT ) );

        prvCreateMQTTConnectionWithBroker( &xMQTTContext, &xNetworkContext );



        /************************* Subscribe. ***************************/



        /* If server rejected the subscription request, attempt to resubscribe 


         * to topic. Attempts are made according to the exponential backoff 


         * retry strategy implemented in retryUtils. */

        prvMQTTSubscribeWithBackoffRetries( &xMQTTContext );



        /*************** Publish and Keep Alive Loop ********************/


        /* Publish messages with QoS1, send and process Keep alive messages. */

        for( ulPublishCount = 0; ulPublishCount < ulMaxPublishCount; 

            ulPublishCount++ )

        {

            LogInfo( ( "Publish to the MQTT topic %s.\r\n", 

                        mqttexampleTOPIC ) );

            prvMQTTPublishToTopic( &xMQTTContext );



            /* Process incoming publish echo, since application subscribed to 


             * the same topic, the broker will send publish message back to the


             * application. */

            LogInfo( ( "Attempt to receive publish message from broker.\r\n" ) 

            );

            xMQTTStatus = MQTT_ProcessLoop( &xMQTTContext, 

            mqttexamplePROCESS_LOOP_TIMEOUT_MS );

            configASSERT( xMQTTStatus == MQTTSuccess );



            /* Leave Connection Idle for some time. */

            LogInfo( ( "Keeping Connection Idle...\r\n\r\n" ) );

            vTaskDelay( mqttexampleDELAY_BETWEEN_PUBLISHES_TICKS );

        }



        /***************** Unsubscribe from the topic. ******************/

        LogInfo( ( "Unsubscribe from the MQTT topic %s.\r\n", mqttexampleTOPIC 

        ) );

        prvMQTTUnsubscribeFromTopic( &xMQTTContext );



        /* Process incoming UNSUBACK packet from the broker. */

        xMQTTStatus = MQTT_ProcessLoop( &xMQTTContext, 

                                        mqttexamplePROCESS_LOOP_TIMEOUT_MS );

        configASSERT( xMQTTStatus == MQTTSuccess );



        /************************* Disconnect. **************************/



        /* Send an MQTT Disconnect packet over the already connected TLS over


         * TCP connection. There is no corresponding response for the 


         * disconnect packet. After sending disconnect, client must close the 


         * network connection. */

        LogInfo( ( "Disconnecting the MQTT connection with %s.\r\n",

                    democonfigMQTT_BROKER_ENDPOINT ) );

        xMQTTStatus = MQTT_Disconnect( &xMQTTContext );

        configASSERT( xMQTTStatus == MQTTSuccess );



        /* Close the network connection.  */

        TLS_FreeRTOS_Disconnect( &xNetworkContext );



        /* Reset SUBACK status for each topic filter after completion of


         * subscription request cycle. */

        for( ulTopicCount = 0; ulTopicCount < mqttexampleTOPIC_COUNT; 

        ulTopicCount++ )

        {

            xTopicFilterContext[ ulTopicCount ].xSubAckStatus = 

            MQTTSubAckFailure;

        }



        /* Wait for some time between two iterations to ensure that we do not


         * bombard the broker. */

        LogInfo( ( "prvMQTTDemoTask() completed an iteration successfully. "

                   "Total free heap is %u.\r\n",

                    xPortGetFreeHeapSize() ) );

        LogInfo( ( "Demo completed successfully.\r\n" ) );

        LogInfo( ( "Short delay before starting the next iteration.... 

                    \r\n\r\n" ) );

        vTaskDelay( mqttexampleDELAY_BETWEEN_DEMO_ITERATIONS_TICKS );

    }

}

static TlsTransportStatus_t prvConnectToServerWithBackoffRetries( 

                                NetworkCredentials_t * pxNetworkCredentials,

                                NetworkContext_t * pxNetworkContext )

{

    TlsTransportStatus_t xNetworkStatus;

    RetryUtilsStatus_t xRetryUtilsStatus = RetryUtilsSuccess;

    RetryUtilsParams_t xReconnectParams;



    #ifdef democonfigUSE_AWS_IOT_CORE_BROKER



        /* ALPN protocols must be a NULL-terminated list of strings. Therefore,


         * the first entry will contain the actual ALPN protocol string while 


         * the second entry must remain NULL. */

        char * pcAlpnProtocols[] = { NULL, NULL };



        /* The ALPN string changes depending on whether username/password 


        authentication is used. */

        #ifdef democonfigCLIENT_USERNAME

            pcAlpnProtocols[ 0 ] = AWS_IOT_CUSTOM_AUTH_ALPN;

        #else

            pcAlpnProtocols[ 0 ] = AWS_IOT_MQTT_ALPN;

        #endif

        pxNetworkCredentials->pAlpnProtos = pcAlpnProtocols;

    #endif /* ifdef democonfigUSE_AWS_IOT_CORE_BROKER */



    pxNetworkCredentials->disableSni = democonfigDISABLE_SNI;

    /* Set the credentials for establishing a TLS connection. */

    pxNetworkCredentials->pRootCa = ( const unsigned char * ) 

    democonfigROOT_CA_PEM;

    pxNetworkCredentials->rootCaSize = sizeof( democonfigROOT_CA_PEM );

    #ifdef democonfigCLIENT_CERTIFICATE_PEM

        pxNetworkCredentials->pClientCert = ( const unsigned char * ) 

                                    democonfigCLIENT_CERTIFICATE_PEM;

        pxNetworkCredentials->clientCertSize = sizeof( 

                                    democonfigCLIENT_CERTIFICATE_PEM );

        pxNetworkCredentials->pPrivateKey = ( const unsigned char * ) 

        democonfigCLIENT_PRIVATE_KEY_PEM;

        pxNetworkCredentials->privateKeySize = sizeof( 

        democonfigCLIENT_PRIVATE_KEY_PEM );

    #endif

    /* Initialize reconnect attempts and interval. */

    RetryUtils_ParamsReset( &xReconnectParams );

    xReconnectParams.maxRetryAttempts = MAX_RETRY_ATTEMPTS;



    /* Attempt to connect to MQTT broker. If connection fails, retry after


     * a timeout. Timeout value will exponentially increase till maximum


     * attempts are reached.*/

    do

    {

        /* Establish a TLS session with the MQTT broker. This example connects 


         * to the MQTT broker as specified in democonfigMQTT_BROKER_ENDPOINT 


         * and democonfigMQTT_BROKER_PORT at the top of this file. */

        LogInfo( ( "Creating a TLS connection to %s:%u.\r\n",

                   democonfigMQTT_BROKER_ENDPOINT,

                   democonfigMQTT_BROKER_PORT ) );

        /* Attempt to create a mutually authenticated TLS connection. */

        xNetworkStatus = TLS_FreeRTOS_Connect( 

                                    pxNetworkContext,

                                    democonfigMQTT_BROKER_ENDPOINT,

                                    democonfigMQTT_BROKER_PORT,

                                    pxNetworkCredentials,

                                    mqttexampleTRANSPORT_SEND_RECV_TIMEOUT_MS,

                                    mqttexampleTRANSPORT_SEND_RECV_TIMEOUT_MS

                                              );



        if( xNetworkStatus != TLS_TRANSPORT_SUCCESS )

        {

            LogWarn( ( "Connection to the broker failed. Retrying connection 

            with backoff and jitter." ) );

            xRetryUtilsStatus = RetryUtils_BackoffAndSleep( &xReconnectParams 

            );

        }



        if( xRetryUtilsStatus == RetryUtilsRetriesExhausted )

        {

            LogError( ( "Connection to the broker failed, all attempts 

            exhausted." ) );

            xNetworkStatus = TLS_TRANSPORT_CONNECT_FAILURE;

        }

    } while( ( xNetworkStatus != TLS_TRANSPORT_SUCCESS ) && 

             ( xRetryUtilsStatus == RetryUtilsSuccess ) );



    return xNetworkStatus;

}

static void prvCreateMQTTConnectionWithBroker( 

                                    MQTTContext_t * pxMQTTContext,

                                    NetworkContext_t * pxNetworkContext )

{

    MQTTStatus_t xResult;

    MQTTConnectInfo_t xConnectInfo;

    bool xSessionPresent;

    TransportInterface_t xTransport;



    /***


     * For readability, error handling in this function is restricted to the 


     * use of asserts().


     ***/



    /* Fill in Transport Interface send and receive function pointers. */

    xTransport.pNetworkContext = pxNetworkContext;

    xTransport.send = TLS_FreeRTOS_send;

    xTransport.recv = TLS_FreeRTOS_recv;



    /* Initialize MQTT library. */

    xResult = MQTT_Init( pxMQTTContext, &xTransport, prvGetTimeMs, 

                         prvEventCallback, &xBuffer );

    configASSERT( xResult == MQTTSuccess );



    /* Some fields are not used in this demo so start with everything at 0. */

    ( void ) memset( ( void * ) &xConnectInfo, 0x00, sizeof( xConnectInfo ) );



    /* Start with a clean session i.e. direct the MQTT broker to discard any


     * previous session data. Also, establishing a connection with clean 


     * session will ensure that the broker does not store any data when this 


     * client gets disconnected. */

    xConnectInfo.cleanSession = true;



    /* The client identifier is used to uniquely identify this MQTT client to


     * the MQTT broker. In a production device the identifier can be something


     * unique, such as a device serial number. */

    xConnectInfo.pClientIdentifier = democonfigCLIENT_IDENTIFIER;

    xConnectInfo.clientIdentifierLength = ( uint16_t ) strlen( 

                                                democonfigCLIENT_IDENTIFIER );



    /* Set MQTT keep-alive period. If the application does not send packets at 


     * an interval less than the keep-alive period, the MQTT library will send 


     * PINGREQ packets. */

    xConnectInfo.keepAliveSeconds = mqttexampleKEEP_ALIVE_TIMEOUT_SECONDS;



    /* Append metrics when connecting to the AWS IoT Core broker. */

    #ifdef democonfigUSE_AWS_IOT_CORE_BROKER

        #ifdef democonfigCLIENT_USERNAME

            xConnectInfo.pUserName = CLIENT_USERNAME_WITH_METRICS;

            xConnectInfo.userNameLength = ( uint16_t ) strlen( 

                                        CLIENT_USERNAME_WITH_METRICS );

            xConnectInfo.pPassword = democonfigCLIENT_PASSWORD;

            xConnectInfo.passwordLength = ( uint16_t ) strlen( 

            democonfigCLIENT_PASSWORD );

        #else

            xConnectInfo.pUserName = AWS_IOT_METRICS_STRING;

            xConnectInfo.userNameLength = AWS_IOT_METRICS_STRING_LENGTH;

            /* Password for authentication is not used. */

            xConnectInfo.pPassword = NULL;

            xConnectInfo.passwordLength = 0U;

        #endif

    #else /* ifdef democonfigUSE_AWS_IOT_CORE_BROKER */

        #ifdef democonfigCLIENT_USERNAME

            xConnectInfo.pUserName = democonfigCLIENT_USERNAME;

            xConnectInfo.userNameLength = ( uint16_t ) strlen( 

                                            democonfigCLIENT_USERNAME );

            xConnectInfo.pPassword = democonfigCLIENT_PASSWORD;

            xConnectInfo.passwordLength = ( uint16_t ) strlen( 

                                            democonfigCLIENT_PASSWORD );

        #endif /* ifdef democonfigCLIENT_USERNAME */

    #endif /* ifdef democonfigUSE_AWS_IOT_CORE_BROKER */



    /* Send MQTT CONNECT packet to broker. LWT is not used in this demo, so it


     * is passed as NULL. */

    xResult = MQTT_Connect( pxMQTTContext,

                            &xConnectInfo,

                            NULL,

                            mqttexampleCONNACK_RECV_TIMEOUT_MS,

                            &xSessionPresent );

    configASSERT( xResult == MQTTSuccess );



    /* Successfully established and MQTT connection with the broker. */

    LogInfo( ( "An MQTT connection is established with %s.", 

                democonfigMQTT_BROKER_ENDPOINT ) );

}