in source/main.c [320:490]
static void hello_task( void * pvParameters )
{
MQTTContext_t xMQTTContext = { 0 };
TransportInterface_t xTransport = { 0 };
MQTTFixedBuffer_t xFixedBuffer = { 0 };
MQTTPublishInfo_t xPublishInfo = { 0 };
MQTTOperation_t xPublishOperation = { 0 };
bool bSessionPresent = true;
MQTTConnectInfo_t xMQTTConnectInfo = { 0 };
MQTTStatus_t xMQTTStatus = MQTTSuccess;
NetworkCredentials_t xNetworkCredentials = { 0 };
TlsTransportStatus_t xTransportStatus = TLS_TRANSPORT_SUCCESS;
NetworkContext_t xNetworkContext = { 0 };
HeapStats_t xHeapStats;
CK_ULONG ulTemp = 0;
char * pcEndpoint = NULL;
char * pcThingName = NULL;
uint32_t ulThingNameLength;
CK_RV xPKCS11Result = CKR_OK;
int32_t lCounter = 0;
char cPayload[ 32 ] = { 0 };
size_t xPayloadLength;
BaseType_t xStatus;
xNetworkCredentials.pRootCa = ( const unsigned char * ) democonfigROOT_CA_PEM;
xNetworkCredentials.rootCaSize = sizeof( democonfigROOT_CA_PEM );
/* Configure MQTT Context */
/* Clear context. */
memset( ( void * ) &xMQTTContext, 0x00, sizeof( MQTTContext_t ) );
while( FreeRTOS_IsNetworkUp() == pdFALSE )
{
PRINTF( "No Network yet\r\n" );
vTaskDelay( pdMS_TO_TICKS( 500 ) );
}
/* Set transport interface members. */
xTransport.pNetworkContext = &xNetworkContext;
xTransport.send = TLS_FreeRTOS_send;
xTransport.recv = TLS_FreeRTOS_recv;
ulGlobalEntryTimeMs = getTimeStampMs();
/* Set buffer members. */
xFixedBuffer.pBuffer = ucBuffer;
xFixedBuffer.size = MQTT_INCOMING_BUFFER_SIZE;
xMQTTStatus = MQTT_Init( &xMQTTContext, &xTransport, getTimeStampMs, eventCallback, &xFixedBuffer );
/* Client ID must be unique to broker. This field is required. */
xPKCS11Result = ulGetThingName( &pcThingName, &ulThingNameLength );
xPKCS11Result = ulGetThingEndpoint( &pcEndpoint, &ulTemp );
if( ( xMQTTStatus == MQTTSuccess ) && ( xPKCS11Result == CKR_OK ) )
{
xMQTTConnectInfo.pClientIdentifier = pcThingName;
xMQTTConnectInfo.clientIdentifierLength = ulThingNameLength;
/* True for creating a new session with broker, false if we want to resume an old one. */
xMQTTConnectInfo.cleanSession = true;
/* The following fields are optional. */
/* Value for keep alive. */
xMQTTConnectInfo.keepAliveSeconds = 60;
/* Optional username and password. */
xMQTTConnectInfo.pUserName = "";
xMQTTConnectInfo.userNameLength = strlen( xMQTTConnectInfo.pUserName );
xMQTTConnectInfo.pPassword = "";
xMQTTConnectInfo.passwordLength = strlen( xMQTTConnectInfo.pPassword );
FreeRTOS_debug_printf( ( "Attempting a connection\n" ) );
xTransportStatus = TLS_FreeRTOS_Connect( xMQTTContext.transportInterface.pNetworkContext, pcEndpoint, 8883, &xNetworkCredentials, 4000, 36000 );
if( TLS_TRANSPORT_SUCCESS == xTransportStatus )
{
/* Send the connect packet. Use 100 ms as the timeout to wait for the CONNACK packet. */
xMQTTStatus = MQTT_Connect( &xMQTTContext, &xMQTTConnectInfo, NULL, 100, &bSessionPresent );
TLS_FreeRTOS_SetRecvTimeout( xMQTTContext.transportInterface.pNetworkContext, 500 );
if( xMQTTStatus == MQTTSuccess )
{
xStatus = MQTTAgent_Init( &xMQTTContext );
configASSERT( xStatus == pdTRUE );
xPublishCompleteSemaphore = xSemaphoreCreateBinary();
configASSERT( xPublishCompleteSemaphore != NULL );
#if ( OTA_UPDATE_ENABLED == 1 )
xStatus = xStartOTAUpdateDemo();
configASSERT( xStatus == pdTRUE );
#endif
for( ; ; )
{
xPayloadLength = snprintf( cPayload, sizeof( cPayload ), "Hello %ld", lCounter++ );
/* Since we requested a clean session, this must be false */
assert( bSessionPresent == false );
/* Do something with the connection. Publish some data. */
xPublishInfo.qos = MQTTQoS0;
xPublishInfo.dup = false;
xPublishInfo.retain = false;
xPublishInfo.pTopicName = "Test/Hello";
xPublishInfo.topicNameLength = 10;
xPublishInfo.pPayload = cPayload;
xPublishInfo.payloadLength = xPayloadLength;
xPublishOperation.type = MQTT_OP_PUBLISH;
xPublishOperation.info.pPublishInfo = &xPublishInfo;
xPublishOperation.callback = publishCompleteCallback;
MQTTAgent_Enqueue( &xPublishOperation, portMAX_DELAY );
xSemaphoreTake( xPublishCompleteSemaphore, portMAX_DELAY );
PRINTF( "Published helloworld.\r\n" );
vTaskDelay( pdMS_TO_TICKS( 5000 ) );
}
vPortGetHeapStats( &xHeapStats );
FreeRTOS_debug_printf( ( "Available heap space %d\n", xHeapStats.xAvailableHeapSpaceInBytes ) );
FreeRTOS_debug_printf( ( "Largest Free Block %d\n", xHeapStats.xSizeOfLargestFreeBlockInBytes ) );
FreeRTOS_debug_printf( ( "Smallest Free Block %d\n", xHeapStats.xSizeOfSmallestFreeBlockInBytes ) );
FreeRTOS_debug_printf( ( "Number of Free Blocks %d\n", xHeapStats.xNumberOfFreeBlocks ) );
FreeRTOS_debug_printf( ( "Minimum Ever Free Bytes Remaining %d\n", xHeapStats.xMinimumEverFreeBytesRemaining ) );
FreeRTOS_debug_printf( ( "Number of Successful Allocations %d\n", xHeapStats.xNumberOfSuccessfulAllocations ) );
FreeRTOS_debug_printf( ( "Number of Successful Frees %d\n", xHeapStats.xNumberOfSuccessfulFrees ) );
/* Disconnect */
MQTT_Disconnect( &xMQTTContext );
}
TLS_FreeRTOS_Disconnect( xMQTTContext.transportInterface.pNetworkContext );
}
else if( TLS_TRANSPORT_INVALID_PARAMETER == xTransportStatus )
{
FreeRTOS_debug_printf( ( "Error Connecting to server : bad parameter\n" ) );
}
else if( TLS_TRANSPORT_CONNECT_FAILURE == xTransportStatus )
{
FreeRTOS_debug_printf( ( "Error Connecting to server : connect failure\n" ) );
}
else
{
FreeRTOS_debug_printf( ( "Error Connecting to server : unknown\n" ) );
}
}
for( ; ; )
{
FreeRTOS_debug_printf( ( "Demo FAILURE\r\n" ) );
vTaskDelay( pdMS_TO_TICKS( 1000 ) );
}
}