coreMQTT 存活演示

static void prvMQTTDemoTask( void * pvParameters )

{

uint32_t ulTopicCount = 0U;

NetworkContext_t xNetworkContext = { 0 };

MQTTContext_t xMQTTContext;

MQTTStatus_t xMQTTStatus;

PlaintextTransportStatus_t xNetworkStatus;

BaseType_t xTimerStatus;



/* Remove compiler warnings about unused parameters. */

( void ) pvParameters;



/* Serialize a PINGREQ packet to send upon invoking the keep-alive timer


* callback. */

xMQTTStatus = MQTT_SerializePingreq( &xPingReqBuffer );

configASSERT( xMQTTStatus == MQTTSuccess );



for( ; ; )

{

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



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


* after a timeout. The timeout value will be exponentially increased


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


* timeout value is reached. The function below returns a failure status


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


* the configured number of attempts. */

xNetworkStatus = prvConnectToServerWithBackoffRetries( &xNetworkContext );

configASSERT( xNetworkStatus == PLAINTEXT_TRANSPORT_SUCCESS );



/* Sends an MQTT Connect packet over the already connected TCP socket,


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

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

prvCreateMQTTConnectionWithBroker( &xMQTTContext, &xNetworkContext );



/* Create timers to handle keep-alive. */

xPingReqTimer = xTimerCreateStatic( "PingReqTimer",

                                    mqttexamplePING_REQUEST_DELAY,

                                    pdTRUE,

                                    ( void * ) &xMQTTContext.transportInterface,

                                    prvPingReqTimerCallback,

                                    &xPingReqTimerBuffer );

configASSERT( xPingReqTimer );

xPingRespTimer = xTimerCreateStatic( "PingRespTimer",

                                     mqttexamplePING_RESPONSE_DELAY,

                                     pdFALSE,

                                     NULL,

                                     prvPingRespTimerCallback,

                                     &xPingRespTimerBuffer );

configASSERT( xPingRespTimer );



/* Start the timer to send a PINGREQ. */

xTimerStatus = xTimerStart( xPingReqTimer, 0 );

configASSERT( xTimerStatus == pdPASS );



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



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


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


* strategy declared in retry_utils.h. */

prvMQTTSubscribeWithBackoffRetries( &xMQTTContext );



/************************ Send PINGREQ packet. ************************/



/* Deliberately delay in order for the auto-reload timer to send a PINGREQ to the broker. */

vTaskDelay( mqttexamplePING_REQUEST_DELAY );



/********************* Publish and Receive Loop. **********************/

/* Publish messages with QOS1, send and process keep-alive messages. */

LogInfo( ( "Publish to the MQTT topic %s.", mqttexampleTOPIC ) );

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. */

LogInfo( ( "Attempt to receive publish message from broker." ) );

while( xReceivedPubAck == pdFALSE )

{

    ulReceiveLoopIterations += 1U;

    configASSERT( ulReceiveLoopIterations <= mqttexampleMAX_RECEIVE_LOOP_ITERATIONS );



    vTaskDelay( mqttexampleRECEIVE_LOOP_ITERATION_DELAY );



    xMQTTStatus = MQTT_ReceiveLoop( &xMQTTContext, 0U );

    configASSERT( xMQTTStatus == MQTTSuccess );

}



/* Reset after loop. */

ulReceiveLoopIterations = 0U;

xReceivedPubAck = pdFALSE;



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

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

prvMQTTUnsubscribeFromTopic( &xMQTTContext );



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

while( xReceivedUnsubAck == pdFALSE )

{

    ulReceiveLoopIterations += 1U;

    configASSERT( ulReceiveLoopIterations <= mqttexampleMAX_RECEIVE_LOOP_ITERATIONS );



    vTaskDelay( mqttexampleRECEIVE_LOOP_ITERATION_DELAY );



}



/* Reset after loop. */

ulReceiveLoopIterations = 0U;

xReceivedUnsubAck = pdFALSE;



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



/* Send an MQTT disconnect packet over the connected TCP socket.


* There is no corresponding response for the disconnect packet. After


* sending the disconnect, the client must close the network connection. */

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

democonfigMQTT_BROKER_ENDPOINT ) );

xMQTTStatus = MQTT_Disconnect( &xMQTTContext );

configASSERT( xMQTTStatus == MQTTSuccess );



/* Stop the keep-alive timers for the next iteration. */

xTimerStatus = xTimerStop( xPingReqTimer, 0 );

configASSERT( xTimerStatus == pdPASS );

xTimerStatus = xTimerStop( xPingRespTimer, 0 );

configASSERT( xTimerStatus == pdPASS );



/* Close the network connection. */

xNetworkStatus = Plaintext_FreeRTOS_Disconnect( &xNetworkContext );

configASSERT( xNetworkStatus == PLAINTEXT_TRANSPORT_SUCCESS );



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


* 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.",

xPortGetFreeHeapSize() ) );

LogInfo( ( "Demo completed successfully." ) );

LogInfo( ( "Short delay before starting the next iteration.... \r\n" ) );

vTaskDelay( mqttexampleDELAY_BETWEEN_DEMO_ITERATIONS );

}

}

static PlaintextTransportStatus_t prvConnectToServerWithBackoffRetries( NetworkContext_t * pxNetworkContext )

{

PlaintextTransportStatus_t xNetworkStatus;

RetryUtilsStatus_t xRetryUtilsStatus = RetryUtilsSuccess;

RetryUtilsParams_t xReconnectParams;



/* 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 TCP connection 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( ( "Create a TCP connection to %s:%d.",

democonfigMQTT_BROKER_ENDPOINT,

democonfigMQTT_BROKER_PORT ) );

xNetworkStatus = Plaintext_FreeRTOS_Connect( pxNetworkContext,

democonfigMQTT_BROKER_ENDPOINT,

democonfigMQTT_BROKER_PORT,

mqttexampleTRANSPORT_SEND_RECV_TIMEOUT_MS,

mqttexampleTRANSPORT_SEND_RECV_TIMEOUT_MS );



if( xNetworkStatus != PLAINTEXT_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 = PLAINTEXT_TRANSPORT_CONNECT_FAILURE;

}

} while( ( xNetworkStatus != PLAINTEXT_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 = 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 a 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 void prvPingReqTimerCallback( TimerHandle_t pxTimer )

{

    TransportInterface_t * pxTransport;

    int32_t xTransportStatus;

    BaseType_t xTimerStatus;



    pxTransport = ( TransportInterface_t * ) pvTimerGetTimerID( pxTimer );



    /* Do not resend if waiting on a PINGRESP. */

    if( xWaitingForPingResp == false )

    {

        /* Send PINGREQ to broker */

        LogInfo( ( "Ping the MQTT broker." ) );

        xTransportStatus = pxTransport->send( pxTransport->pNetworkContext,

                                              ( void * ) xPingReqBuffer.pBuffer,

                                              xPingReqBuffer.size );

        configASSERT( ( size_t ) xTransportStatus == xPingReqBuffer.size );



        xWaitingForPingResp = true;

        /* Start the timer to expect a PINGRESP. */

        xTimerStatus = xTimerStart( xPingRespTimer, 0 );

        configASSERT( xTimerStatus == pdPASS );

    }

}

static void prvPingRespTimerCallback( TimerHandle_t pxTimer )

{

    ( void ) pxTimer;



    /* Assert that a pending PINGRESP has been received. */

    configASSERT( xWaitingForPingResp == false );

}

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;

uint32_t ulTopicCount = 0U;



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

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



/* Get a unique packet id. */

usSubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );



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


* only one topic and uses QoS0. */

xMQTTSubscription[ 0 ].qos = MQTTQoS0;

xMQTTSubscription[ 0 ].pTopicFilter = mqttexampleTOPIC;

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



/* Initialize retry attempts and interval. */

RetryUtils_ParamsReset( &xRetryParams );

xRetryParams.maxRetryAttempts = MAX_RETRY_ATTEMPTS;



do

{

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


* as specified in mqttexampleTOPIC at the top of this file by sending a


* subscribe packet then waiting for a subscribe acknowledgment (SUBACK).


* This client will then publish to the same topic it subscribed to, so it


* will expect all the messages it sends to the broker to be sent back to it


* from the broker. This demo uses QOS0 in Subscribe. Therefore, the publish


* messages received from the broker will have QOS0. */

LogInfo( ( "Attempt to subscribe to the MQTT topic %s.", mqttexampleTOPIC ) );

xResult = MQTT_Subscribe( pxMQTTContext,

xMQTTSubscription,

sizeof( xMQTTSubscription ) / sizeof( MQTTSubscribeInfo_t ),

usSubscribePacketIdentifier );

configASSERT( xResult == MQTTSuccess );



LogInfo( ( "SUBSCRIBE sent for topic %s to broker.\n\n", mqttexampleTOPIC ) );



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


* client may receive a publish before it receives a subscribe ack. Therefore,


* call the generic incoming packet processing function. Since this demo is


* subscribing to the topic to which no one is publishing, probability of


* receiving a publish message before a subscribe ack is zero; but the application


* must be ready to receive any packet. This demo uses the generic packet


* processing function everywhere to highlight this fact. */

while( xReceivedSubAck == pdFALSE )

{

    ulReceiveLoopIterations += 1U;

    configASSERT( ulReceiveLoopIterations <= mqttexampleMAX_RECEIVE_LOOP_ITERATIONS );



    vTaskDelay( mqttexampleRECEIVE_LOOP_ITERATION_DELAY );



    xResult = MQTT_ReceiveLoop( pxMQTTContext, 0U );

    configASSERT( xResult == MQTTSuccess );

}



/* Reset in case another attempt to subscribe is needed. */

ulReceiveLoopIterations = 0U;

xReceivedSubAck = pdFALSE;



/* Reset flag before checking suback responses. */

xFailedSubscribeToTopic = false;



/* Check if the recent subscription request has been rejected. #xTopicFilterContext


* is updated in the event callback to reflect the status of the SUBACK


* sent by the broker. It represents either the QoS level granted by the


* server upon subscription or acknowledgement of server rejection of the


* subscription request. */

for( ulTopicCount = 0; ulTopicCount < mqttexampleTOPIC_COUNT; ulTopicCount++ )

{

if( xTopicFilterContext[ ulTopicCount ].xSubAckStatus == MQTTSubAckFailure )

{

LogWarn( ( "Server rejected subscription request. Attempting to re-subscribe to topic %s.",

xTopicFilterContext[ ulTopicCount ].pcTopicFilter ) );

xFailedSubscribeToTopic = true;

xRetryUtilsStatus = RetryUtils_BackoffAndSleep( &xRetryParams );

break;

}

}



configASSERT( xRetryUtilsStatus != RetryUtilsRetriesExhausted );

} while( ( xFailedSubscribeToTopic == true ) && ( xRetryUtilsStatus == RetryUtilsSuccess ) );

}

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 void prvMQTTPublishToTopic( MQTTContext_t * pxMQTTContext )

{

MQTTStatus_t xResult;

MQTTPublishInfo_t xMQTTPublishInfo;

BaseType_t xTimerStatus;



/***


* 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 = mqttexampleTOPIC;

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

xMQTTPublishInfo.pPayload = mqttexampleMESSAGE;

xMQTTPublishInfo.payloadLength = strlen( mqttexampleMESSAGE );



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

xResult = MQTT_Publish( pxMQTTContext, &xMQTTPublishInfo, 0U );

configASSERT( xResult == MQTTSuccess );



/* When a PUBLISH packet has been sent, the keep-alive timer can be reset. */

xTimerStatus = prvCheckTimeoutThenResetTimer( xKeepAliveTimer );

configASSERT( xTimerStatus == pdPASS );

}

static void prvMQTTProcessIncomingPublish( MQTTPublishInfo_t * pxPublishInfo )

{

configASSERT( pxPublishInfo != NULL );



/* Process incoming Publish. */

LogInfo( ( "Incoming QoS : %d\n", pxPublishInfo->qos ) );



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

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

( 0 == strncmp( mqttexampleTOPIC, pxPublishInfo->pTopicName, pxPublishInfo->topicNameLength ) ) )

{

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 prvMQTTUnsubscribeFromTopic( MQTTContext_t * pxMQTTContext )

{

MQTTStatus_t xResult;

MQTTSubscribeInfo_t xMQTTSubscription[ mqttexampleTOPIC_COUNT ];



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

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



/* Get a unique packet id. */

usSubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );



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


* only one topic and uses QoS0. */

xMQTTSubscription[ 0 ].qos = MQTTQoS0;

xMQTTSubscription[ 0 ].pTopicFilter = mqttexampleTOPIC;

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



/* Get the next unique packet identifier. */

usUnsubscribePacketIdentifier = MQTT_GetPacketId( pxMQTTContext );



/* Send the UNSUBSCRIBE packet. */

xResult = MQTT_Unsubscribe( pxMQTTContext,

xMQTTSubscription,

sizeof( xMQTTSubscription ) / sizeof( MQTTSubscribeInfo_t ),

usUnsubscribePacketIdentifier );



configASSERT( xResult == MQTTSuccess );

}