coreMQTT 演示（无 TLS）

static void prvMQTTDemoTask( void * pvParameters )

{

    uint32_t ulPublishCount = 0U, ulTopicCount = 0U;

    const uint32_t ulMaxPublishCount = 5UL;



    NetworkContext_t xNetworkContext = { 0 };

    MQTTContext_t xMQTTContext;

    MQTTStatus_t xMQTTStatus;

    PlaintextTransportStatus_t xNetworkStatus;



    /***


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


     * the use of asserts().


     ***/



for( ; ; )

{

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



    /* Attempt to connect to the MQTT broker. The socket is returned in


     * the network context structure. */

    xNetworkStatus = prvConnectToServerWithBackoffRetries( &xNetworkContext );

    configASSERT( xNetworkStatus == PLAINTEXT_TRANSPORT_SUCCESS );



    /* Connect to the MQTT broker using the already connected TCP socket. */

    prvCreateMQTTConnectionWithBroker( &xMQTTContext, &xNetworkContext );



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



    /* Subscribe to the test topic. */

    prvMQTTSubscribeWithBackoffRetries( &xMQTTContext );



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



    /* Publish messages with QoS0, then send and process Keep Alive


     * messages. */

    for( ulPublishCount = 0; ulPublishCount < ulMaxPublishCount; ulPublishCount++ )

    {

        prvMQTTPublishToTopic( &xMQTTContext );



        /* Process the incoming publish echo. Since the application subscribed


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


         * back to the application. Note there is a separate demo that


         * shows how to use coreMQTT in a thread safe way - in which case the


         * MQTT protocol runs in the background and this call is not


         * required. */

        xMQTTStatus = MQTT_ProcessLoop(

            &xMQTTContext,

            mqttexamplePROCESS_LOOP_TIMEOUT_MS );

            configASSERT( xMQTTStatus == MQTTSuccess );



        /* Leave the connection idle for some time. */

        vTaskDelay( mqttexampleDELAY_BETWEEN_PUBLISHES );

    }



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



    prvMQTTUnsubscribeFromTopic( &xMQTTContext );



    /* Process the incoming packet from the broker. Note there is a separate


     * demo that shows how to use coreMQTT in a thread safe way - in which case


     * the MQTT protocol runs in the background and this call is not required. */

    xMQTTStatus = MQTT_ProcessLoop( &xMQTTContext,

        mqttexamplePROCESS_LOOP_TIMEOUT_MS );

        configASSERT( xMQTTStatus == MQTTSuccess );



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



    xMQTTStatus = MQTT_Disconnect( &xMQTTContext );



    /* Close the network connection. */

    xNetworkStatus = Plaintext_FreeRTOS_Disconnect( &xNetworkContext );

    configASSERT( xNetworkStatus == PLAINTEXT_TRANSPORT_SUCCESS );



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


     * bombard the MQTT broker. */

    vTaskDelay( mqttexampleDELAY_BETWEEN_DEMO_ITERATIONS );

}

}

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 = Plaintext_FreeRTOS_send;

    xTransport.recv = Plaintext_FreeRTOS_recv;



    /* Initialize MQTT library. */

    xResult = MQTT_Init( 

        pxMQTTContext,

        &xTransport,

        prvGetTimeMs,

        prvEventCallback,

        &xBuffer );

configASSERT( xResult == MQTTSuccess );



/* Many fields 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. It is the responsibility of the application


* to ensure that the interval between Control Packets being sent does not


* exceed the Keep Alive value. In the absence of sending any other


* Control Packets, the Client MUST send a PINGREQ Packet. */

xConnectInfo.keepAliveSeconds = mqttexampleKEEP_ALIVE_TIMEOUT_SECONDS;



/* 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 );

}

static const char *const pcExampleTopic = "/example/topic";



static void prvMQTTSubscribeWithBackoffRetries( MQTTContext_t * pxMQTTContext )

{

    MQTTStatus_t xResult = MQTTSuccess;

    RetryUtilsStatus_t xRetryUtilsStatus = RetryUtilsSuccess;

    RetryUtilsParams_t xRetryParams;

    MQTTSubscribeInfo_t xMQTTSubscription[ mqttexampleTOPIC_COUNT ];

    bool xFailedSubscribeToTopic = false;



    /***


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


     * the use of asserts().


     ***/



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

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



    /* Each packet requires a unique ID. */

    usSubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );



    /* Subscribe to the pcExampleTopic topic filter. This example subscribes


     * to only one topic and uses QoS0. */

    xMQTTSubscription[ 0 ].qos = MQTTQoS0;

    xMQTTSubscription[ 0 ].pTopicFilter = pcExampleTopic;

    xMQTTSubscription[ 0 ].topicFilterLength = strlen( pcExampleTopic );



    /* Initialize retry attempts and interval. */

    RetryUtils_ParamsReset( &xRetryParams );

    xRetryParams.maxRetryAttempts = MAX_RETRY_ATTEMPTS;



    do

    {

        /* The client is already connected to the broker. Subscribe to the topic


         * as specified in pcExampleTopic by sending a subscribe packet then


         * waiting for a subscribe acknowledgment (SUBACK). */

        xResult = MQTT_Subscribe(

                pxMQTTContext,

                xMQTTSubscription,

                1, /* Only subscribing to one topic. */

                usSubscribePacketIdentifier );

        configASSERT( xResult == MQTTSuccess );



        /* Process incoming packet from the broker. After sending the


         * subscribe, the client may receive a publish before it receives a


         * subscribe ack. Therefore, call generic incoming packet processing


         * function. Since this demo is subscribing to the topic to which no


         * one is publishing, probability of receiving Publish message before


         * subscribe ack is zero; but application must be ready to receive any


         * packet. This demo uses the generic packet processing function


         * everywhere to highlight this fact. Note there is a separate demo that


         * shows how to use coreMQTT in a thread safe way – in which case the


         * MQTT protocol runs in the background and this call is not required. */

        xResult = MQTT_ProcessLoop( pxMQTTContext, mqttexamplePROCESS_LOOP_TIMEOUT_MS );

        configASSERT( xResult == MQTTSuccess );



        /* Reset flag before checking suback responses. */

        xFailedSubscribeToTopic = false;



        /* Check if recent subscription request has been rejected.


         * #xTopicFilterContext is updated in the event callback (shown in a


         * code block below) to reflect the status of the SUBACK sent by the


         * broker. It represents either the QoS level granted by the server upon


         * subscription, or acknowledgment of server rejection of the


         * subscription request. */

        if( xTopicFilterContext.xSubAckStatus == MQTTSubAckFailure )

        {

            xFailedSubscribeToTopic = true;

            xRetryUtilsStatus = RetryUtils_BackoffAndSleep( &xRetryParams );

            break;

        }



        configASSERT( xRetryUtilsStatus != RetryUtilsRetriesExhausted );

    } while( ( xFailedSubscribeToTopic == true ) &&

             ( xRetryUtilsStatus == RetryUtilsSuccess ) );

}

static const char *const pcExampleTopic = "/example/topic";



static void prvMQTTPublishToTopic( MQTTContext_t * pxMQTTContext )

{

    MQTTStatus_t xResult;

    MQTTPublishInfo_t xMQTTPublishInfo;



    /***


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


     * use of asserts().


     ***/



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

    ( void ) memset( ( void * ) &xMQTTPublishInfo, 0x00, sizeof(

        xMQTTPublishInfo ) );



    /* This demo uses QoS0. */

    xMQTTPublishInfo.qos = MQTTQoS0;

    xMQTTPublishInfo.retain = false;

    xMQTTPublishInfo.pTopicName = pcExampleTopic;

    xMQTTPublishInfo.topicNameLength = ( uint16_t ) strlen( pcExampleTopic );

    xMQTTPublishInfo.pPayload = mqttexampleMESSAGE;

    xMQTTPublishInfo.payloadLength = strlen( mqttexampleMESSAGE );



    /* Send PUBLISH packet. Packet ID is not used for a QoS0 publish. */

    xResult = MQTT_Publish( 

            pxMQTTContext,

            &xMQTTPublishInfo, 0U );

            configASSERT( xResult == MQTTSuccess );

}

static void prvEventCallback( MQTTContext_t * pxMQTTContext,

                              MQTTPacketInfo_t * pxPacketInfo,

                              MQTTDeserializedInfo_t * pxDeserializedInfo )

{

    /* The MQTT context is not used for this demo. */

    ( void ) pxMQTTContext;



    if( ( pxPacketInfo->type & 0xF0U ) == MQTT_PACKET_TYPE_PUBLISH )

    {

        prvMQTTProcessIncomingPublish( pxDeserializedInfo->pPublishInfo );

    }

    else

    {

        prvMQTTProcessResponse( pxPacketInfo, pxDeserializedInfo->packetIdentifier );

    }

}

static const char *const pcExampleTopic = "/example/topic";



static void prvMQTTProcessIncomingPublish( MQTTPublishInfo_t * pxPublishInfo )

{

    /* Verify the received publish is for the we have subscribed to. */

    if( ( pxPublishInfo->topicNameLength == strlen( pcExampleTopic ) ) &&

        ( 0 == strcmp( pcExampleTopic, pxPublishInfo->pTopicName ) ) )

    {

        LogInfo( ( "Incoming Publish Topic Name: %.*s matches subscribed topic.\r\n"

                   "Incoming Publish Message : %.*s",

                   pxPublishInfo->topicNameLength,

                   pxPublishInfo->pTopicName,

                   pxPublishInfo->payloadLength,

                   pxPublishInfo->pPayload ) );

    }

    else

    {

        LogInfo( ( "Incoming Publish Topic Name: %.*s does not match subscribed topic.",

        pxPublishInfo->topicNameLength,

        pxPublishInfo->pTopicName ) );

    }

}



static void prvMQTTProcessResponse( MQTTPacketInfo_t * pxIncomingPacket,

                                    uint16_t usPacketId )

{

    MQTTStatus_t xResult = MQTTSuccess;

    uint8_t * pucPayload = NULL;

    size_t ulSize = 0;



    switch( pxIncomingPacket->type )

    {

        case MQTT_PACKET_TYPE_SUBACK:



            /* A SUBACK from the broker, containing the server response to our


             * subscription request, has been received. It contains the status


             * code indicating server approval/rejection for the subscription to


             * the single topic requested. The SUBACK will be parsed to obtain


             * the status code, and this status code will be stored in


             * #xTopicFilterContext. */

            xResult = MQTT_GetSubAckStatusCodes( 

                     pxIncomingPacket,

                     &pucPayload,

                     &ulSize );



            /* MQTT_GetSubAckStatusCodes always returns success if called with


             * packet info from the event callback and non-NULL parameters. */

            configASSERT( xResult == MQTTSuccess );



            /* This should be the QOS leve, 0 in this case. */

            xTopicFilterContext.xSubAckStatus = *pucPayload;



            /* Make sure ACK packet identifier matches with Request packet


             * identifier. */

            configASSERT( usSubscribePacketIdentifier == usPacketId );

            break;



        case MQTT_PACKET_TYPE_UNSUBACK:

            LogInfo( ( "Unsubscribed from the topic %s.", mqttexampleTOPIC ) );

            /* Make sure ACK packet identifier matches with Request packet


             * identifier. */

            configASSERT( usUnsubscribePacketIdentifier == usPacketId );

            break;



        case MQTT_PACKET_TYPE_PINGRESP:

            /* Nothing to be done from application as library handles


             * PINGRESP with the use of MQTT_ProcessLoop API function. */

            LogWarn( ( "PINGRESP should not be handled by the application "

                       "callback when using MQTT_ProcessLoop.\n" ) );

            break;



        /* Any other packet type is invalid. */

        default:

            LogWarn( ( "prvMQTTProcessResponse() called with unknown packet type:(%02X).",

                pxIncomingPacket->type ) );

    }

}

static const char *const pcExampleTopic = "/example/topic";



static void prvMQTTUnsubscribeFromTopic( MQTTContext_t * pxMQTTContext )

{

    MQTTStatus_t xResult;

    MQTTSubscribeInfo_t xMQTTSubscription[ mqttexampleTOPIC_COUNT ];



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

    ( void ) memset( ( void * ) &xMQTTSubscription, 0x00, sizeof(

        xMQTTSubscription ) );



    /* Subscribe to the pcExampleTopic topic filter. This example subscribes


     * to only one topic and uses QoS0. */

    xMQTTSubscription[ 0 ].qos = MQTTQoS0;

    xMQTTSubscription[ 0 ].pTopicFilter = pcExampleTopic;

    xMQTTSubscription[ 0 ].topicFilterLength = ( uint16_t ) strlen(

        pcExampleTopic);



    /* Each packet requires a unique ID. */

    usUnsubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );



    /* Send UNSUBSCRIBE packet. */

    xResult = MQTT_Unsubscribe( pxMQTTContext,

                                xMQTTSubscription,

                                sizeof( xMQTTSubscription ) / sizeof(

                                MQTTSubscribeInfo_t ),

                                usUnsubscribePacketIdentifier );



    configASSERT( xResult == MQTTSuccess );

}