len0rd/FreeRTOS-Kernel
1
0
Fork 0
mirror of https://github.com/FreeRTOS/FreeRTOS-Kernel.git synced 2025-08-29 22:48:37 -04:00
Code Issues Projects Releases Packages Wiki Activity

Add the Labs projects provided in the V10.2.1_191129 zip file.

Browse source
This commit is contained in:
Richard Barry 2019-12-02 23:39:25 +00:00
parent 46e5937529
commit e5708b38e9
801 changed files with 356576 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

853
FreeRTOS-Labs/Demo/FreeRTOS_IoT_Libraries/mqtt/common/DemoTasks/LightWeightMQTTExample.c Normal file
View file

@ -0,0 +1,853 @@
/*
* FreeRTOS Kernel V10.2.1
* Copyright (C) 2019 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://www.FreeRTOS.org
* http://aws.amazon.com/freertos
*
* 1 tab == 4 spaces!
*/
/*
* Proof of Concept for use of MQTT light weight serializer API.
* Light weight serializer API lets user to serialize and
* deserialize MQTT messages into user provided buffer.
* This API allows use of statically allocated buffer.
*
* Example shown below uses this API to create MQTT messages and
* and send them over connection established using FreeRTOS sockets.
* The example is single threaded and uses statically allocated memory.
*
* !!! NOTE !!!
* This is work in progress to show how light weight serializer
* API can be used. This is not a complete demo, and should not
* be treated as production ready code.
*/
/* Standard includes. */
#include <string.h>
#include <stdio.h>
/* Kernel includes. */
#include "FreeRTOS.h"
#include "task.h"
/* FreeRTOS+TCP includes. */
#include "FreeRTOS_IP.h"
#include "FreeRTOS_Sockets.h"
/* IoT SDK includes. */
#include "iot_mqtt.h"
#include "iot_mqtt_serialize.h"
#include "platform/iot_network_freertos.h"
/* Demo Specific configs. */
#include "mqtt_demo_profile.h"
/**
* @brief Time to wait between each cycle of the demo implemented by prvMQTTDemoTask().
*/
#define mqttexampleDELAY_BETWEEN_DEMO_ITERATIONS ( pdMS_TO_TICKS( 5000 ) )
/**
* @brief Time to wait before sending ping request to keep MQTT connection alive.
*/
#define mqttexampleKEEP_ALIVE_DELAY ( pdMS_TO_TICKS( 1000 ) )
/**
* @brief The MQTT client identifier used in this example. Each client identifier
* must be unique so edit as required to ensure no two clients connecting to the
* same broker use the same client identifier.
*/
#define mqttexampleCLIENT_IDENTIFIER mqttdemoprofileCLIENT_IDENTIFIER
/**
* @brief Details of the MQTT broker to connect to.
*/
#define mqttexampleMQTT_BROKER_ENDPOINT mqttdemoprofileBROKER_ENDPOINT
/**
* @brief The port to use for the demo.
*/
#define mqttexampleMQTT_BROKER_PORT mqttdemoprofileBROKER_PORT
/**
* @brief The topic to subscribe and publish to in the example.
*
* The topic starts with the client identifier to ensure that each demo interacts
* with a unique topic.
*/
#define mqttexampleTOPIC mqttexampleCLIENT_IDENTIFIER "/example/topic"
/**
* @brief The MQTT message published in this example.
*/
#define mqttexampleMESSAGE "Hello Light Weight MQTT World!"
/**
* @brief Dimensions a file scope buffer currently used to send and receive MQTT data from a
* socket
*/
#define mqttexampleSHARED_BUFFER_SIZE 500
/*-----------------------------------------------------------*/
/**
* @brief MQTT Protocol constants used by this demo.
* These types are defined in internal MQTT include files.
* For light-weight demo application, only a few are needed, therefore
* they are redefined here so that internal files need not be included.
*/
/* MQTT Control Packet Types.*/
#define MQTT_PACKET_TYPE_CONNACK ( ( uint8_t ) 0x20U ) /**< @brief CONNACK (server-to-client). */
#define MQTT_PACKET_TYPE_PUBLISH ( ( uint8_t ) 0x30U ) /**< @brief PUBLISH (bi-directional). */
#define MQTT_PACKET_TYPE_SUBACK ( ( uint8_t ) 0x90U ) /**< @brief SUBACK (server-to-client). */
#define MQTT_PACKET_TYPE_UNSUBACK ( ( uint8_t ) 0xb0U ) /**< @brief UNSUBACK (server-to-client). */
#define MQTT_PACKET_TYPE_PINGRESP ( ( uint8_t ) 0xd0U ) /**< @brief PINGRESP (server-to-client). */
/* MQTT Fixed Packet Sizes */
#define MQTT_PACKET_DISCONNECT_SIZE ( ( uint8_t ) 2 ) /**< @brief Size of DISCONNECT packet. */
#define MQTT_PACKET_PINGREQ_SIZE ( ( uint8_t ) 2 ) /**< @brief Size of PINGREQ packet. */
/*-----------------------------------------------------------*/
/**
* @brief The task used to demonstrate the MQTT API.
*
* @param[in] pvParameters Parameters as passed at the time of task creation. Not
* used in this example.
*/
static void prvMQTTDemoTask( void * pvParameters );
/**
* @brief Creates a TCP connection to the MQTT broker as specified in
* mqttexampleMQTT_BROKER_ENDPOINT and mqttexampleMQTT_BROKER_PORT.
*
* @return On success the socket connected to the MQTT broker is returned. Otherwise
* FREERTOS_INVALID_SOCKET is returned.
*
*/
static Socket_t prvCreateTCPConnectionToBroker( void );
/**
* @brief Sends an MQTT Connect packet over the already connected TCP socket.
*
* @param xMQTTSocketis a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return IOT_MQTT_SUCCESS is returned if the reply is a valid connection
* acknowledgeable (CONNACK) packet, otherwise an error code is returned.
*/
static IotMqttError_t prvCreateMQTTConnectionWithBroker( Socket_t xMQTTSocket );
/**
* @brief Performs a graceful shutdown and close of the socket passed in as its
* parameter.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*/
static void prvGracefulShutDown( Socket_t xSocket );
/**
* @brief Subscribes to the topic as specified in mqttexampleTOPIC.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return IOT_MQTT_SUCCESS is returned if the
* subscription is successful, otherwise an error code is returned.
*/
static IotMqttError_t prvMQTTSubscribeToTopic( Socket_t xMQTTSocket );
/**
* @brief Publishes a messages mqttexampleMESSAGE on mqttexampleTOPIC topic.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return IOT_MQTT_SUCCESS is returned if Publish is successful,
* otherwise an error code is returned.
*/
static IotMqttError_t prvMQTTPublishToTopic( Socket_t xMQTTSocket );
/**
* @brief Process Incoming Publish.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return #IOT_MQTT_SUCCESS is returned if the processing is successful,
* otherwise an error code is returned.
*/
static IotMqttError_t prvMQTTProcessIncomingPublish( Socket_t xMQTTSocket );
/**
* @brief Unsubscribes from the previously subscribed topic as specified
* in mqttexampleTOPIC.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return IOT_MQTT_SUCCESS is returned if the
* unsubscribe is successful, otherwise an error code is returned.
*/
static IotMqttError_t prvMQTTUnsubscribeFromTopic( Socket_t xMQTTSocket );
/**
* @brief Send MQTT Ping Request to broker and receive response.
* Ping request is used to keep connection to broker alive.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return IOT_MQTT_SUCCESS is returned if the successful Ping Response is received.
* otherwise an error code is returned.
*/
static IotMqttError_t prvMQTTKeepAlive( Socket_t xMQTTSocket );
/**
* @brief Disconnect From MQTT Broker.
*
* @param xMQTTSocket is a TCP socket that is connected to an MQTT broker to which
* an MQTT connection has been established.
*
* @return IOT_MQTT_SUCCESS is returned if the disconnect is successful,
* otherwise an error code is returned.
*/
static IotMqttError_t prvMQTTDisconnect( Socket_t xMQTTSocket );
/*-----------------------------------------------------------*/
/**
* @brief Function to receive next byte from network,
* The declaration must match IotMqttGetNextByte_t.
*
* @param[in] pvContext Network Connection context. Implementation in this
* file uses FreeRTOS socket.
* @param[in, out] pNextBye Pointer to buffer where the byte will be stored.
*
* @return #IOT_MQTT_SUCCESS or #IOT_MQTT_TIMEOUT
*/
IotMqttError_t getNextByte( void * pvContext,
uint8_t * pNextByte );
/*-----------------------------------------------------------*/
/* @brief Static memory buffer used for sending and receiving MQTT messages */
static uint8_t ucSharedBuffer[ mqttexampleSHARED_BUFFER_SIZE ];
/*-----------------------------------------------------------*/
/*
* @brief Task for Light Weight MQTT Serializer API Proof of Concept.
* To run the proof of concept example, in main.c, in function vApplicationIPNetworkEventHook(),
* replace vStartSimpleMQTTDemo() with vApplicationIPNetworkEventHook().
*/
void vStartLightWeightMQTTDemo( void )
{
TickType_t xShortDelay = ( TickType_t ) pdMS_TO_TICKS( ( TickType_t ) 500 );
/* Wait a short time to allow receipt of the ARP replies. */
vTaskDelay( xShortDelay );
/* This example uses a single application task, which in turn is used to
* connect, subscribe, publish, unsubscribe and disconnect from the MQTT
* broker. */
xTaskCreate( prvMQTTDemoTask, /* Function that implements the task. */
"MQTTLWDemo", /* Text name for the task - only used for debugging. */
democonfigDEMO_STACKSIZE, /* Size of stack (in words, not bytes) to allocate for the task. */
NULL, /* Task parameter - not used in this case. */
tskIDLE_PRIORITY, /* Task priority, must be between 0 and configMAX_PRIORITIES - 1. */
NULL ); /* Used to pass out a handle to the created task - not used in this case. */
}
/*-----------------------------------------------------------*/
static void prvGracefulShutDown( Socket_t xSocket )
{
uint8_t ucDummy[ 20 ];
const TickType_t xShortDelay = pdMS_TO_MIN_TICKS( 250 );
if( xSocket != ( Socket_t ) 0 )
{
if( xSocket != FREERTOS_INVALID_SOCKET )
{
/* Initiate graceful shutdown. */
FreeRTOS_shutdown( xSocket, FREERTOS_SHUT_RDWR );
/* Wait for the socket to disconnect gracefully (indicated by FreeRTOS_recv()
* returning a FREERTOS_EINVAL error) before closing the socket. */
while( FreeRTOS_recv( xSocket, ucDummy, sizeof( ucDummy ), 0 ) >= 0 )
{
/* Wait for shutdown to complete. If a receive block time is used then
* this delay will not be necessary as FreeRTOS_recv() will place the RTOS task
* into the Blocked state anyway. */
vTaskDelay( xShortDelay );
/* Note ? real applications should implement a timeout here, not just
* loop forever. */
}
/* The socket has shut down and is safe to close. */
FreeRTOS_closesocket( xSocket );
}
}
}
/*-----------------------------------------------------------*/
IotMqttError_t getNextByte( void * pvContext,
uint8_t * pNextByte )
{
Socket_t xMQTTSocket = ( Socket_t ) pvContext;
BaseType_t receivedBytes;
IotMqttError_t result;
/* Receive one byte from network */
receivedBytes = FreeRTOS_recv( xMQTTSocket, ( void * ) pNextByte, sizeof( uint8_t ), 0 );
if( receivedBytes == sizeof( uint8_t ) )
{
result = IOT_MQTT_SUCCESS;
}
else
{
result = IOT_MQTT_TIMEOUT;
}
return result;
}
/*-----------------------------------------------------------*/
static void prvMQTTDemoTask( void * pvParameters )
{
const TickType_t xNoDelay = ( TickType_t ) 0;
Socket_t xMQTTSocket;
IotMqttError_t xReturned;
uint32_t ulPublishCount = 0;
const uint32_t ulMaxPublishCount = 5UL;
/* Remove compiler warnings about unused parameters. */
( void ) pvParameters;
for( ; ; )
{
/* Don't expect any notifications to be pending yet. */
configASSERT( ulTaskNotifyTake( pdTRUE, xNoDelay ) == 0 );
/****************************** Connect. ******************************/
/* Establish a TCP connection with the MQTT broker. This example connects to
* the MQTT broker as specified in mqttexampleMQTT_BROKER_ENDPOINT and
* mqttexampleMQTT_BROKER_PORT at the top of this file. */
configPRINTF( ( "Create a TCP connection to %s\r\n", mqttexampleMQTT_BROKER_ENDPOINT ) );
xMQTTSocket = prvCreateTCPConnectionToBroker();
configASSERT( xMQTTSocket != FREERTOS_INVALID_SOCKET );
configPRINTF( ( "Connected to %s\r\n", mqttexampleMQTT_BROKER_ENDPOINT ) );
/* Sends an MQTT Connect packet over the already connected TCP socket
* xMQTTSocket, then waits for and interprets the reply. IOT_MQTT_SUCCESS is
* returned if the reply is a valid connection acknowledgeable (CONNACK) packet,
* otherwise an error code is returned. */
configPRINTF( ( "Creating an MQTT connection with %s\r\n", mqttexampleMQTT_BROKER_ENDPOINT ) );
xReturned = prvCreateMQTTConnectionWithBroker( xMQTTSocket );
configASSERT( xReturned == IOT_MQTT_SUCCESS );
configPRINTF( ( "Established an MQTT connection.\r\n" ) );
/**************************** Subscribe. ******************************/
/* 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 will
* expect all the messages it sends to the broker to be sent back to it
* from the broker. */
configPRINTF( ( "Attempt to subscribed to the MQTT topic %s\r\n", mqttexampleTOPIC ) );
xReturned = prvMQTTSubscribeToTopic( xMQTTSocket );
configPRINTF( ( "Subscribed to the topic %s\r\n", mqttexampleTOPIC ) );
/**************************** Publish. ******************************/
/* Send publish for with QOS0, Process Keep alive */
for( ulPublishCount = 0; ulPublishCount < ulMaxPublishCount; ulPublishCount++ )
{
configPRINTF( ( "Attempt to publish to the MQTT topic %s\r\n", mqttexampleTOPIC ) );
xReturned = prvMQTTPublishToTopic( xMQTTSocket );
configASSERT( xReturned == IOT_MQTT_SUCCESS );
configPRINTF( ( "Publish successful to the topic %s\r\n", mqttexampleTOPIC ) );
/* Process incoming publish echo, since application subscribed to the same topic
* broker will send publish message back to the application */
configPRINTF( ( "Attempt to receive publish message from broker\r\n" ) );
xReturned = prvMQTTProcessIncomingPublish( xMQTTSocket );
configASSERT( xReturned == IOT_MQTT_SUCCESS );
configPRINTF( ( "Successfully Received Publish message from broker\r\n" ) );
/* Leave Connection Idle for some time */
configPRINTF( ( "Keeping Connection Idle\r\n" ) );
vTaskDelay( pdMS_TO_TICKS( mqttexampleKEEP_ALIVE_DELAY ) );
/* Send Ping request to broker and receive ping response */
configPRINTF( ( "Sending Ping Request to the broker\r\n" ) );
xReturned = prvMQTTKeepAlive( xMQTTSocket );
configASSERT( xReturned == IOT_MQTT_SUCCESS );
configPRINTF( ( "Ping Response successfully received\r\n" ) );
}
/************************ Unsubscribe from the topic. **************************/
configPRINTF( ( "Attempt to unsubscribe from the MQTT topic %s\r\n", mqttexampleTOPIC ) );
xReturned = prvMQTTUnsubscribeFromTopic( xMQTTSocket );
configASSERT( xReturned == IOT_MQTT_SUCCESS );
configPRINTF( ( "Unsubscribe from the topic %s\r\n", mqttexampleTOPIC ) );
/**************************** Disconnect. ******************************/
/* Sends an MQTT Disconnect packet over the already connected TCP socket
* xMQTTSocket, then waits for and interprets the reply. IOT_MQTT_SUCCESS is
* returned if the reply is a valid connection acknowledgeable (CONNACK) packet,
* otherwise an error code is returned. */
configPRINTF( ( "Creating an MQTT connection with %s\r\n", mqttexampleMQTT_BROKER_ENDPOINT ) );
xReturned = prvMQTTDisconnect( xMQTTSocket );
configASSERT( xReturned == IOT_MQTT_SUCCESS );
configPRINTF( ( "Established an MQTT connection.\r\n" ) );
/* Disconnect from broker. */
prvGracefulShutDown( xMQTTSocket );
/* Wait for some time between two iterations to ensure that we do not
* bombard the public test mosquitto broker. */
configPRINTF( ( "prvMQTTDemoTask() completed an iteration successfully. Total free heap is %u\r\n", xPortGetFreeHeapSize() ) );
configPRINTF( ( "Short delay before starting the next iteration.... \r\n\r\n" ) );
vTaskDelay( pdMS_TO_TICKS( mqttexampleDELAY_BETWEEN_DEMO_ITERATIONS ) );
}
}
/*-----------------------------------------------------------*/
Socket_t prvCreateTCPConnectionToBroker( void )
{
Socket_t xMQTTSocket;
struct freertos_sockaddr xBrokerAddress;
uint32_t ulBrokerIPAddress;
/* This is the socket used to connect to the MQTT broker. */
xMQTTSocket = FreeRTOS_socket( FREERTOS_AF_INET,
FREERTOS_SOCK_STREAM,
FREERTOS_IPPROTO_TCP );
configASSERT( xMQTTSocket != FREERTOS_INVALID_SOCKET );
/* Locate then connect to the MQTT broker. */
ulBrokerIPAddress = FreeRTOS_gethostbyname( mqttexampleMQTT_BROKER_ENDPOINT );
if( ulBrokerIPAddress != 0 )
{
xBrokerAddress.sin_port = FreeRTOS_htons( mqttexampleMQTT_BROKER_PORT );
xBrokerAddress.sin_addr = ulBrokerIPAddress;
if( FreeRTOS_connect( xMQTTSocket, &xBrokerAddress, sizeof( xBrokerAddress ) ) != 0 )
{
/* Could not connect so delete socket and return an error. */
FreeRTOS_closesocket( xMQTTSocket );
xMQTTSocket = FREERTOS_INVALID_SOCKET;
}
}
return xMQTTSocket;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvCreateMQTTConnectionWithBroker( Socket_t xMQTTSocket )
{
IotMqttConnectInfo_t xConnectInfo;
size_t xRemainingLength = 0;
size_t xPacketSize = 0;
IotMqttError_t xResult;
IotMqttPacketInfo_t xIncomingPacket;
/* Many fields not used in this demo so start with everything at 0. */
memset( ( void * ) &xConnectInfo, 0x00, sizeof( xConnectInfo ) );
memset( ( void * ) &xIncomingPacket, 0x00, sizeof( xIncomingPacket ) );
/* 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 = mqttexampleCLIENT_IDENTIFIER;
xConnectInfo.clientIdentifierLength = ( uint16_t ) strlen( mqttexampleCLIENT_IDENTIFIER );
/* Get size requirement for the connect packet */
xResult = IotMqtt_GetConnectPacketSize( &xConnectInfo, &xRemainingLength, &xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
/* Make sure the packet size is less than static buffer size */
configASSERT( xPacketSize < mqttexampleSHARED_BUFFER_SIZE );
/* Serialize MQTT connect packet into provided buffer */
xResult = IotMqtt_SerializeConnect( &xConnectInfo, xRemainingLength, ucSharedBuffer, xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
if( FreeRTOS_send( xMQTTSocket, ( void * ) ucSharedBuffer, xPacketSize, 0 ) == ( BaseType_t ) xPacketSize )
{
/* Wait for the connection ack. TODO check the receive timeout value. */
memset( ( void * ) &xIncomingPacket, 0x00, sizeof( IotMqttPacketInfo_t ) );
/* Get packet type and remaining length of the received packet
* We cannot assume received data is the connection acknowledgment.
* Therefore this function reads type and remaining length of the
* received packet, before processing entire packet.
*/
xResult = IotMqtt_GetIncomingMQTTPacketTypeAndLength( &xIncomingPacket, getNextByte, ( void * ) xMQTTSocket );
configASSERT( xResult == IOT_MQTT_SUCCESS );
configASSERT( xIncomingPacket.type == MQTT_PACKET_TYPE_CONNACK );
configASSERT( xIncomingPacket.remainingLength <= mqttexampleSHARED_BUFFER_SIZE );
if( FreeRTOS_recv( xMQTTSocket, ( void * ) ucSharedBuffer, xIncomingPacket.remainingLength, 0 )
== ( BaseType_t ) xIncomingPacket.remainingLength )
{
xIncomingPacket.pRemainingData = ucSharedBuffer;
if( IotMqtt_DeserializeResponse( &xIncomingPacket ) != IOT_MQTT_SUCCESS )
{
xResult = IOT_MQTT_SERVER_REFUSED;
}
}
else
{
configPRINTF( ( "Receive Failed while receiving MQTT ConnAck\n" ) );
xResult = IOT_MQTT_NETWORK_ERROR;
}
}
else
{
configPRINTF( ( "Send Failed while connecting to MQTT broker\n" ) );
xResult = IOT_MQTT_NETWORK_ERROR;
}
return xResult;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvMQTTSubscribeToTopic( Socket_t xMQTTSocket )
{
IotMqttError_t xResult;
IotMqttSubscription_t xMQTTSubscription[ 1 ];
size_t xRemainingLength = 0;
size_t xPacketSize = 0;
uint16_t usPacketIdentifier;
IotMqttPacketInfo_t xIncomingPacket;
/* Some fields not used by this demo so start with everything at 0. */
memset( ( void * ) &xMQTTSubscription, 0x00, sizeof( xMQTTSubscription ) );
/* Subscribe to the mqttexampleTOPIC topic filter. This example subscribes to only one topic */
xMQTTSubscription[ 0 ].qos = IOT_MQTT_QOS_0;
xMQTTSubscription[ 0 ].pTopicFilter = mqttexampleTOPIC;
xMQTTSubscription[ 0 ].topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
xResult = IotMqtt_GetSubscriptionPacketSize( IOT_MQTT_SUBSCRIBE,
xMQTTSubscription,
sizeof( xMQTTSubscription ) / sizeof( IotMqttSubscription_t ),
&xRemainingLength, &xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
/* Make sure the packet size is less than static buffer size */
configASSERT( xPacketSize < mqttexampleSHARED_BUFFER_SIZE );
/* Serialize subscribe into statically allocated ucSharedBuffer */
xResult = IotMqtt_SerializeSubscribe( xMQTTSubscription,
sizeof( xMQTTSubscription ) / sizeof( IotMqttSubscription_t ),
xRemainingLength,
&usPacketIdentifier,
ucSharedBuffer,
xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
if( FreeRTOS_send( xMQTTSocket, ( void * ) ucSharedBuffer, xPacketSize, 0 ) == ( BaseType_t ) xPacketSize )
{
/* Wait for the subscription ack. The socket is already connected to the MQTT broker, so
* publishes to this client can occur at any time and we cannot assume received
* data is the subscription acknowledgment. Therefore this function is, at this
* time, doing what would otherwise be done wherever incoming packets are
* interpreted (in a callback, or whatever). */
memset( ( void * ) &xIncomingPacket, 0x00, sizeof( IotMqttPacketInfo_t ) );
xResult = IotMqtt_GetIncomingMQTTPacketTypeAndLength( &xIncomingPacket, getNextByte, ( void * ) xMQTTSocket );
configASSERT( xResult == IOT_MQTT_SUCCESS );
configASSERT( xIncomingPacket.type == MQTT_PACKET_TYPE_SUBACK );
configASSERT( xIncomingPacket.remainingLength <= mqttexampleSHARED_BUFFER_SIZE );
/* Receive the remaining bytes. */
if( FreeRTOS_recv( xMQTTSocket, ( void * ) ucSharedBuffer, xIncomingPacket.remainingLength, 0 ) == ( BaseType_t ) xIncomingPacket.remainingLength )
{
xIncomingPacket.pRemainingData = ucSharedBuffer;
if( IotMqtt_DeserializeResponse( &xIncomingPacket ) != IOT_MQTT_SUCCESS )
{
xResult = IOT_MQTT_BAD_RESPONSE;
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
return xResult;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvMQTTPublishToTopic( Socket_t xMQTTSocket )
{
IotMqttError_t xResult;
IotMqttPublishInfo_t xMQTTPublishInfo;
size_t xRemainingLength = 0;
size_t xPacketSize = 0;
uint16_t usPacketIdentifier;
uint8_t * pusPacketIdentifierHigh;
/* Some fields not used by this demo so start with everything at 0. */
memset( ( void * ) &xMQTTPublishInfo, 0x00, sizeof( xMQTTPublishInfo ) );
xMQTTPublishInfo.qos = IOT_MQTT_QOS_0;
xMQTTPublishInfo.retain = false;
xMQTTPublishInfo.pTopicName = mqttexampleTOPIC;
xMQTTPublishInfo.topicNameLength = ( uint16_t ) strlen( mqttexampleTOPIC );
xMQTTPublishInfo.pPayload = mqttexampleMESSAGE;
xMQTTPublishInfo.payloadLength = strlen( mqttexampleMESSAGE );
/* Find out length of Publish packet size. */
xResult = IotMqtt_GetPublishPacketSize( &xMQTTPublishInfo, &xRemainingLength, &xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
/* Make sure the packet size is less than static buffer size */
configASSERT( xPacketSize < mqttexampleSHARED_BUFFER_SIZE );
xResult = IotMqtt_SerializePublish( &xMQTTPublishInfo,
xRemainingLength,
&usPacketIdentifier,
&pusPacketIdentifierHigh,
ucSharedBuffer,
xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
if( FreeRTOS_send( xMQTTSocket, ( void * ) ucSharedBuffer, xPacketSize, 0 ) != ( BaseType_t ) xPacketSize )
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
else
{
/* Send success. Since in this case, we are using IOT_MQTT_QOS_0,
* there will not be any PubAck. Publish will be echoed back, which is processed
* in prvMQTTProcessIncomingPublish() */
xResult = IOT_MQTT_SUCCESS;
}
return xResult;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvMQTTProcessIncomingPublish( Socket_t xMQTTSocket )
{
IotMqttError_t xResult;
IotMqttPacketInfo_t xIncomingPacket;
memset( ( void * ) &xIncomingPacket, 0x00, sizeof( IotMqttPacketInfo_t ) );
xResult = IotMqtt_GetIncomingMQTTPacketTypeAndLength( &xIncomingPacket, getNextByte, ( void * ) xMQTTSocket );
configASSERT( xResult == IOT_MQTT_SUCCESS );
configASSERT( ( xIncomingPacket.type & 0xf0 ) == MQTT_PACKET_TYPE_PUBLISH );
configASSERT( xIncomingPacket.remainingLength <= mqttexampleSHARED_BUFFER_SIZE );
/* Receive the remaining bytes. */
if( FreeRTOS_recv( xMQTTSocket, ( void * ) ucSharedBuffer, xIncomingPacket.remainingLength, 0 ) == ( BaseType_t ) xIncomingPacket.remainingLength )
{
xIncomingPacket.pRemainingData = ucSharedBuffer;
if( IotMqtt_DeserializePublish( &xIncomingPacket ) != IOT_MQTT_SUCCESS )
{
xResult = IOT_MQTT_BAD_RESPONSE;
}
else
{
/* Process incoming Publish */
configPRINTF( ( "Incoming QOS : %d\n", xIncomingPacket.pubInfo.qos ) );
configPRINTF( ( "Incoming Publish Topic Name: %.*s\n", xIncomingPacket.pubInfo.topicNameLength, xIncomingPacket.pubInfo.pTopicName ) );
configPRINTF( ( "Incoming Publish Message : %.*s\n", xIncomingPacket.pubInfo.payloadLength, xIncomingPacket.pubInfo.pPayload ) );
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
return xResult;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvMQTTUnsubscribeFromTopic( Socket_t xMQTTSocket )
{
IotMqttError_t xResult;
IotMqttSubscription_t xMQTTSubscription[ 1 ];
size_t xRemainingLength;
size_t xPacketSize;
uint16_t usPacketIdentifier;
IotMqttPacketInfo_t xIncomingPacket;
/* Some fields not used by this demo so start with everything at 0. */
memset( ( void * ) &xMQTTSubscription, 0x00, sizeof( xMQTTSubscription ) );
/* Unsubscribe to the mqttexampleTOPIC topic filter. The task handle is passed
* as the callback context which is used by the callback to send a task
* notification to this task.*/
xMQTTSubscription[ 0 ].qos = IOT_MQTT_QOS_0;
xMQTTSubscription[ 0 ].pTopicFilter = mqttexampleTOPIC;
xMQTTSubscription[ 0 ].topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
xResult = IotMqtt_GetSubscriptionPacketSize( IOT_MQTT_UNSUBSCRIBE,
xMQTTSubscription,
sizeof( xMQTTSubscription ) / sizeof( IotMqttSubscription_t ),
&xRemainingLength,
&xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
/* Make sure the packet size is less than static buffer size */
configASSERT( xPacketSize < mqttexampleSHARED_BUFFER_SIZE );
xResult = IotMqtt_SerializeUnsubscribe( xMQTTSubscription,
sizeof( xMQTTSubscription ) / sizeof( IotMqttSubscription_t ),
xRemainingLength,
&usPacketIdentifier,
ucSharedBuffer,
xPacketSize );
configASSERT( xResult == IOT_MQTT_SUCCESS );
if( FreeRTOS_send( xMQTTSocket, ( void * ) ucSharedBuffer, xPacketSize, 0 ) == ( BaseType_t ) xPacketSize )
{
/* Wait for the subscription ack. The socket is already connected to the MQTT broker, so
* publishes to this client can occur at any time and we cannot assume received
* data is the subscription acknowledgment. Therefore this function is, at this
* time, doing what would otherwise be done wherever incoming packets are
* interpreted (in a callback, or whatever). */
memset( ( void * ) &xIncomingPacket, 0x00, sizeof( IotMqttPacketInfo_t ) );
xResult = IotMqtt_GetIncomingMQTTPacketTypeAndLength( &xIncomingPacket, getNextByte, ( void * ) xMQTTSocket );
configASSERT( xResult == IOT_MQTT_SUCCESS );
configASSERT( xIncomingPacket.type == MQTT_PACKET_TYPE_UNSUBACK );
configASSERT( xIncomingPacket.remainingLength <= sizeof( ucSharedBuffer ) );
/* Receive the remaining bytes. */
if( FreeRTOS_recv( xMQTTSocket, ( void * ) ucSharedBuffer, xIncomingPacket.remainingLength, 0 ) == ( BaseType_t ) xIncomingPacket.remainingLength )
{
xIncomingPacket.pRemainingData = ucSharedBuffer;
if( IotMqtt_DeserializeResponse( &xIncomingPacket ) != IOT_MQTT_SUCCESS )
{
xResult = IOT_MQTT_BAD_RESPONSE;
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
return xResult;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvMQTTKeepAlive( Socket_t xMQTTSocket )
{
IotMqttError_t xResult;
IotMqttPacketInfo_t xIncomingPacket;
/* PingReq is fixed length packet, therefore there is no need to calculate the size,
* just makes sure static buffer can accommodate ping request */
configASSERT( MQTT_PACKET_PINGREQ_SIZE <= mqttexampleSHARED_BUFFER_SIZE );
xResult = IotMqtt_SerializePingreq( ucSharedBuffer, MQTT_PACKET_PINGREQ_SIZE );
configASSERT( xResult == IOT_MQTT_SUCCESS );
if( FreeRTOS_send( xMQTTSocket, ( void * ) ucSharedBuffer, MQTT_PACKET_PINGREQ_SIZE, 0 ) == ( BaseType_t ) MQTT_PACKET_PINGREQ_SIZE )
{
memset( ( void * ) &xIncomingPacket, 0x00, sizeof( IotMqttPacketInfo_t ) );
xResult = IotMqtt_GetIncomingMQTTPacketTypeAndLength( &xIncomingPacket, getNextByte, ( void * ) xMQTTSocket );
configASSERT( xResult == IOT_MQTT_SUCCESS );
configASSERT( xIncomingPacket.type == MQTT_PACKET_TYPE_PINGRESP );
configASSERT( xIncomingPacket.remainingLength <= sizeof( ucSharedBuffer ) );
/* Receive the remaining bytes. */
if( FreeRTOS_recv( xMQTTSocket, ( void * ) ucSharedBuffer, xIncomingPacket.remainingLength, 0 )
== ( BaseType_t ) xIncomingPacket.remainingLength )
{
xIncomingPacket.pRemainingData = ucSharedBuffer;
if( IotMqtt_DeserializeResponse( &xIncomingPacket ) != IOT_MQTT_SUCCESS )
{
xResult = IOT_MQTT_BAD_RESPONSE;
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
return xResult;
}
/*-----------------------------------------------------------*/
static IotMqttError_t prvMQTTDisconnect( Socket_t xMQTTSocket )
{
IotMqttError_t xResult;
/* Disconnect is fixed length packet, therefore there is no need to calculate the size,
* just makes sure static buffer can accommodate disconnect request */
configASSERT( MQTT_PACKET_DISCONNECT_SIZE <= mqttexampleSHARED_BUFFER_SIZE );
xResult = IotMqtt_SerializeDisconnect( ucSharedBuffer, MQTT_PACKET_DISCONNECT_SIZE );
configASSERT( xResult == IOT_MQTT_SUCCESS );
if( FreeRTOS_send( xMQTTSocket, ( void * ) ucSharedBuffer, MQTT_PACKET_DISCONNECT_SIZE, 0 ) == ( BaseType_t ) MQTT_PACKET_DISCONNECT_SIZE )
{
xResult = IOT_MQTT_SUCCESS;
}
else
{
xResult = IOT_MQTT_NETWORK_ERROR;
}
return xResult;
}

659
FreeRTOS-Labs/Demo/FreeRTOS_IoT_Libraries/mqtt/common/DemoTasks/SimpleMQTTExamples.c Normal file
View file

@ -0,0 +1,659 @@
/*
* FreeRTOS Kernel V10.2.1
* Copyright (C) 2019 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://www.FreeRTOS.org
* http://aws.amazon.com/freertos
*
* 1 tab == 4 spaces!
*/
/*
* This file contains the common functions for plain text, basic TLS, and mutual
* authentication MQTT demos. Aside from the difference in security level during
* connect, the three demos perform the same interaction with a MQTT broker. The
* demos create a single application task that connects to a MQTT broker,
* subscribes to a topic, publishes a topic with a message, and disconnect from a
* MQTT broker. The task subscribes to the same topic it publishes to, receiving
* the message it sends to the broker. Note that this demo does not terminate, the
* connect-subscribe-publish-disconnect cycle is repeated for unlimited number of
* times.
*
* The plain text MQTT demo does not authenticate the server nor the client. The
* basic TLS MQTT demo builds on top of the plain text demo, adding broker
* authentication and encryption. The mutual authentication MQTT demo builds on top
* of the basic TLS demo, enabling both server and client authentication.
*
* For more information regarding the MQTT library and the demo, please refer to:
* https://freertos.org/mqtt/index.html
*/
/* Standard includes. */
#include <string.h>
#include <stdio.h>
/* Kernel includes. */
#include "FreeRTOS.h"
#include "task.h"
/* FreeRTOS+TCP includes. */
#include "FreeRTOS_IP.h"
/* IoT SDK includes. */
#include "iot_mqtt.h"
#include "iot_taskpool_freertos.h"
#include "platform/iot_network_freertos.h"
/* MQTT Demo Select */
#include "demo_config.h"
/* Select MQTT profile based on the setting in demo_config.h */
#if ( democonfigPROFILE_USE_AWS_IOT == 1 )
#include "aws_iot_demo_profile.h"
#else
#include "mqtt_demo_profile.h"
#endif
/* Preprocessor check for configuration */
#include "aws_iot_setup_check.h"
/*
* Set connection profile based on the setting in demo_config.h. For more
* information on each variable, please refer to the respective *_profile.h
* file in FreeRTOS-Labs\Demo\FreeRTOS_IoT_Libraries\include.
*
* Note that if you are running mqtt_tls_mutual_auth demo please make sure to
* visit the following link for setup:
* https://www.freertos.org/mqtt/preconfiguredexamplesMA.html
*/
#if ( democonfigPROFILE_USE_AWS_IOT == 1 )
#define mqttexampleBROKER_ENDPOINT awsiotdemoprofileAWS_ENDPOINT
#define mqttexampleBROKER_PORT awsiotdemoprofileAWS_MQTT_PORT
#define mqttexampleBROKER_CERTIFICATE_PEM awsiotdemoprofileAWS_CERTIFICATE_PEM
#define mqttexampleCLIENT_IDENTIFIER awsiotdemoprofileCLIENT_IDENTIFIER
#define mqttexampleCLIENT_CERTIFICATE_PEM awsiotdemoprofileCLIENT_CERTIFICATE_PEM
#define mqttexampleCLIENT_PRIVATE_KEY_PEM awsiotdemoprofileCLIENT_PRIVATE_KEY_PEM
#else
#define mqttexampleBROKER_ENDPOINT mqttdemoprofileBROKER_ENDPOINT
#define mqttexampleBROKER_PORT mqttdemoprofileBROKER_PORT
#define mqttexampleBROKER_CERTIFICATE_PEM mqttdemoprofileBROKER_CERTIFICATE_PEM
#define mqttexampleCLIENT_IDENTIFIER mqttdemoprofileCLIENT_IDENTIFIER
#endif /* if ( democonfigPROFILE_USE_AWS_IOT == pdTRUE ) */
/**
* @brief The keep-alive interval used for this example.
*
* An MQTT ping request will be sent periodically at this interval.
*
* @note: This value is set to zero to disable MQTT
* keep alive for the Windows simulator project.
* The FreeRTOS kernel does not accurately calculate time for the Windows
* Simulator. Therefore, MQTT PING Request messages may be sent
* at an incorrect time interval to the broker. If the broker does
* not receive a ping request within 1.5x the time sent in a
* connection request, the broker may close the connection.
* To enable the keep alive feature, set this value
* to the desired interval in seconds.
*/
#define mqttexampleKEEP_ALIVE_SECONDS ( 0 )
/**
* @brief The timeout for MQTT operations in this example.
*/
#define mqttexampleMQTT_TIMEOUT_MS ( 5000 )
/**
* @brief The topic to subscribe and publish to in the example.
*
* The topic starts with the client identifier to ensure that each demo interacts
* with a unique topic.
*/
#define mqttexampleTOPIC mqttexampleCLIENT_IDENTIFIER "/example/topic"
/**
* @brief The MQTT message published in this example.
*/
#define mqttexampleMESSAGE "Hello World!"
/**
* @brief Parameters to control the retry behavior in case a QoS1 publish
* message gets lost.
*
* Retry every minutes up to a maximum of 5 retries.
*/
#define mqttexamplePUBLISH_RETRY_MS ( 1000 )
#define mqttexamplePUBLISH_RETRY_LIMIT ( 5 )
/**
* @brief The bit which is set in the demo task's notification value from the
* disconnect callback to inform the demo task about the MQTT disconnect.
*/
#define mqttexampleDISCONNECTED_BIT ( 1UL << 0UL )
/**
* @brief The bit which is set in the demo task's notification value from the
* publish callback to inform the demo task about the message received from the
* MQTT broker.
*/
#define mqttexampleMESSAGE_RECEIVED_BIT ( 1UL << 1UL )
/*-----------------------------------------------------------*/
/**
* @brief The task used to demonstrate the MQTT API.
*
* @param[in] pvParameters Parameters as passed at the time of task creation. Not
* used in this example.
*/
static void prvMQTTDemoTask( void * pvParameters );
/**
* @brief The callback invoked by the MQTT library when the MQTT connection gets
* disconnected.
*
* @param[in] pvCallbackContext Callback context as provided at the time of
* connect.
* @param[in] pxCallbackParams Contains the reason why the MQTT connection was
* disconnected.
*/
static void prvExample_OnDisconnect( void * pvCallbackContext,
IotMqttCallbackParam_t * pxCallbackParams );
/**
* @brief The callback invoked by the MQTT library when a message is received on
* a subscribed topic from the MQTT broker.
*
* @param[in] pvCallbackContext Callback context as provided at the time of
* subscribe.
* @param[in] pxCallbackParams Contain the details about the received message -
* topic on which the message was received, the received message.
*/
static void prvExample_OnMessageReceived( void * pvCallbackContext,
IotMqttCallbackParam_t * pxCallbackParams );
/**
* @brief Connects to the MQTT broker as specified in mqttexampleBROKER_ENDPOINT
* and mqttexampleBROKER_PORT.
*/
static void prvMQTTConnect( void );
/**
* @brief Subscribes to the topic as specified in mqttexampleTOPIC.
*/
static void prvMQTTSubscribe( void );
/**
* @brief Publishes a messages mqttexampleMESSAGE on mqttexampleTOPIC topic.
*/
static void prvMQTTPublish( void );
/**
* @brief Unsubscribes from the mqttexampleTOPIC topic.
*/
static void prvMQTTUnsubscribe( void );
/**
* @brief Disconnects from the MQTT broker gracefully by sending an MQTT
* DISCONNECT message.
*/
static void prvMQTTDisconnect( void );
/**
* @brief Initializes the IoT libraries used by this demo.
*/
static void prvInitialiseLibraries( void );
/**
* @brief The MQTT connection handle used in this example.
*/
static IotMqttConnection_t xMQTTConnection = IOT_MQTT_CONNECTION_INITIALIZER;
/*-----------------------------------------------------------*/
/**
* @brief Parameters used to create the system task pool.
*/
static const IotTaskPoolInfo_t xTaskPoolParameters =
{
/* Minimum number of threads in a task pool.
* Note the slimmed down version of the task
* pool used by this library does not auto-scale
* the number of tasks in the pool so in this
* case this sets the number of tasks in the
* pool. */
1,
/* Maximum number of threads in a task pool.
* Note the slimmed down version of the task
* pool used by this library does not auto-scale
* the number of tasks in the pool so in this
* case this parameter is just ignored. */
1,
/* Stack size for every task pool thread - in
* bytes, hence multiplying by the number of bytes
* in a word as configMINIMAL_STACK_SIZE is
* specified in words. */
configMINIMAL_STACK_SIZE * sizeof( portSTACK_TYPE ),
/* Priority for every task pool thread. */
tskIDLE_PRIORITY,
};
/*-----------------------------------------------------------*/
static const struct IotNetworkServerInfo xMQTTBrokerInfo =
{
.pHostName = mqttexampleBROKER_ENDPOINT,
.port = mqttexampleBROKER_PORT
};
#if ( democonfigENABLE_TLS )
static struct IotNetworkCredentials xNetworkSecurityCredentials =
{
/* Optional TLS extensions. For this demo, they are disabled. */
.pAlpnProtos = NULL,
.maxFragmentLength = 0,
/* SNI is enabled by default. */
.disableSni = false,
/* Provide the certificate for validating the server. Only required for
* demos using TLS. */
.pRootCa = mqttexampleBROKER_CERTIFICATE_PEM,
.rootCaSize = sizeof( mqttexampleBROKER_CERTIFICATE_PEM ),
/* Strong mutual authentication to authenticate both the broker and
* the client. */
#if ( democonfigENABLE_MUTUAL_AUTH )
.pClientCert = mqttexampleCLIENT_CERTIFICATE_PEM,
.clientCertSize = sizeof( mqttexampleCLIENT_CERTIFICATE_PEM ),
.pPrivateKey = mqttexampleCLIENT_PRIVATE_KEY_PEM,
.privateKeySize = sizeof( mqttexampleCLIENT_PRIVATE_KEY_PEM )
#else
.pClientCert = NULL,
.clientCertSize = 0,
.pPrivateKey = NULL,
.privateKeySize = 0
#endif /* if ( democonfigENABLE_MUTUAL_AUTH ) */
};
#endif /* if ( democonfigENABLE_TLS ) */
static IotMqttNetworkInfo_t xNetworkInfo =
{
/* No connection to the MQTT broker has been established yet and we want to
* establish a new connection. */
.createNetworkConnection = true,
.u.setup.pNetworkServerInfo = &( xMQTTBrokerInfo ),
/* Set the TLS credentials for the new MQTT connection. This member is NULL
* for the plain text MQTT demo. */
#if ( democonfigENABLE_TLS )
.u.setup.pNetworkCredentialInfo = &xNetworkSecurityCredentials,
#else
.u.setup.pNetworkCredentialInfo = NULL, /* Not using TLS so no credentials. */
#endif
/* Use FreeRTOS+TCP network interface. */
.pNetworkInterface = IOT_NETWORK_INTERFACE_FREERTOS,
/* Setup the callback which is called when the MQTT connection is
* disconnected. The task handle is passed as the callback context which
* is used by the callback to send a task notification to this task.*/
.disconnectCallback.function = prvExample_OnDisconnect
};
static const IotMqttConnectInfo_t xConnectInfo =
{
/* Set this flag to true if connecting to the AWS IoT MQTT broker. */
#if ( democonfigPROFILE_USE_AWS_IOT == 1 )
.awsIotMqttMode = true,
#else
.awsIotMqttMode = false,
#endif
/* 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. */
.cleanSession = true,
/* Since we are starting with a clean session, there are no previous
* subscriptions to be restored. */
.pPreviousSubscriptions = NULL,
.previousSubscriptionCount = 0,
/* We do not want to publish Last Will and Testament (LWT) message if the
* client gets disconnected. */
.pWillInfo = NULL,
/* Send an MQTT PING request every minute to keep the connection open if
there is no other MQTT traffic. */
.keepAliveSeconds = mqttexampleKEEP_ALIVE_SECONDS,
/* 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. */
.pClientIdentifier = mqttexampleCLIENT_IDENTIFIER,
.clientIdentifierLength = ( uint16_t ) sizeof( mqttexampleCLIENT_IDENTIFIER ) - 1,
/* This example does not authenticate the client and therefore username and
* password fields are not used. */
.pUserName = NULL,
.userNameLength = 0,
.pPassword = NULL,
.passwordLength = 0
};
/*-----------------------------------------------------------*/
void vStartSimpleMQTTDemo( void )
{
TickType_t xShortDelay = ( TickType_t ) pdMS_TO_TICKS( ( TickType_t ) 500 );
/* Wait a short time to allow receipt of the ARP replies. */
vTaskDelay( xShortDelay );
/* This example uses a single application task, which in turn is used to
* connect, subscribe, publish, unsubscribe and disconnect from the MQTT
* broker. */
xTaskCreate( prvMQTTDemoTask, /* Function that implements the task. */
"MQTTDemo", /* Text name for the task - only used for debugging. */
democonfigDEMO_STACKSIZE, /* Size of stack (in words, not bytes) to allocate for the task. */
NULL, /* Task parameter - not used in this case. */
tskIDLE_PRIORITY, /* Task priority, must be between 0 and configMAX_PRIORITIES - 1. */
NULL ); /* Used to pass out a handle to the created task - not used in this case. */
}
/*-----------------------------------------------------------*/
static void prvMQTTDemoTask( void * pvParameters )
{
uint32_t ulNotificationValue = 0, ulPublishCount;
const uint32_t ulMaxPublishCount = 5UL;
const TickType_t xNoDelay = ( TickType_t ) 0;
/* Remove compiler warnings about unused parameters. */
( void ) pvParameters;
/* One time initialization of the libraries used by this demo. */
prvInitialiseLibraries();
for( ; ; )
{
/* Notifications are used to send events from the callback functions to this
* task. Don't expect any notifications to be pending at the beginning of the
* loop. */
configASSERT( ulTaskNotifyTake( pdTRUE, xNoDelay ) == 0 );
/****************************** Connect. ******************************/
/* Establish a connection to the MQTT broker. This example connects to
* the MQTT broker as specified by the compile time constants
* mqttexampleBROKER_ENDPOINT and mqttexampleBROKER_PORT.
* Please change it to the MQTT broker you want to connect to. */
configPRINTF( ( "Attempt to connect to %s:%d\r\n", mqttexampleBROKER_ENDPOINT, mqttexampleBROKER_PORT ) );
prvMQTTConnect();
configPRINTF( ( "Connected to %s:%d\r\n", mqttexampleBROKER_ENDPOINT, mqttexampleBROKER_PORT ) );
/**************************** Subscribe. ******************************/
/* The client is now connected to the broker. Subscribe to the topic
* as specified by the mqttexampleTOPIC compile time constant. This
* client will then publish to the same topic it subscribed to, so will
* expect all the messages it sends to the broker to be sent back to it
* from the broker. */
configPRINTF( ( "Attempt to subscribed to the topic %s\r\n", mqttexampleTOPIC ) );
prvMQTTSubscribe();
configPRINTF( ( "Subscribed to the topic %s\r\n", mqttexampleTOPIC ) );
/*********************** Publish ulMaxPublishCount messages. **********/
/* Publish a few messages while connected. */
for( ulPublishCount = 0; ulPublishCount < ulMaxPublishCount; ulPublishCount++ )
{
/* Publish a message on the topic specified by the mqttexampleTOPIC
* compile time constant. */
configPRINTF( ( "Publish %s on the topic %s\r\n", mqttexampleMESSAGE, mqttexampleTOPIC ) );
prvMQTTPublish();
configPRINTF( ( "Published %s on the topic %s\r\n", mqttexampleMESSAGE, mqttexampleTOPIC ) );
/* Since we are subscribed to the same topic as we published on, we
* will get the same message back from the MQTT broker. Wait for the
* message to be received which is signaled to us by the publish
* callback (prvExample_OnMessageReceived) setting the
* mqttexampleMESSAGE_RECEIVED_BIT bit in this task's notification
* value. Note the bit is then cleared in the task's notification value
* to ensure the bit being set can be detected on the next iteration. */
xTaskNotifyWait( 0UL, /* Don't clear any bits on entry. */
mqttexampleMESSAGE_RECEIVED_BIT, /* Clear bit on exit. */
&( ulNotificationValue ), /* Obtain the notification value. */
pdMS_TO_TICKS( mqttexampleMQTT_TIMEOUT_MS ) );
configASSERT( ( ulNotificationValue & mqttexampleMESSAGE_RECEIVED_BIT ) == mqttexampleMESSAGE_RECEIVED_BIT );
}
/******************* Unsubscribe and Disconnect. **********************/
/* Unsubscribe from the topic mqttexampleTOPIC and disconnect
* gracefully. */
prvMQTTUnsubscribe();
prvMQTTDisconnect();
configPRINTF( ( "Disconnected from %s:%d\r\n\r\n", mqttexampleBROKER_ENDPOINT, mqttexampleBROKER_PORT ) );
/* Wait for the disconnect operation to complete which is signaled to us
* by the disconnect callback (prvExample_OnDisconnect)by setting
* the mqttexampleDISCONNECTED_BIT bit in this task's notification value.
* Note the bit is cleared in the task's notification value again to ensure
* it being set can be detected again on the next iteration. */
xTaskNotifyWait( 0UL, /* Don't clear any bits on entry. */
mqttexampleDISCONNECTED_BIT, /* Clear bit on exit. */
&( ulNotificationValue ), /* Obtain the notification value. */
pdMS_TO_TICKS( mqttexampleMQTT_TIMEOUT_MS ) );
configASSERT( ( ulNotificationValue & mqttexampleDISCONNECTED_BIT ) == mqttexampleDISCONNECTED_BIT );
/* Delay between iterations to avoid broker throttling. */
configPRINTF( ( "prvMQTTDemoTask() completed an iteration successfully. Total free heap is %u\r\n", xPortGetFreeHeapSize() ) );
configPRINTF( ( "Demo completed successfully.\r\n" ) );
configPRINTF( ( "Short delay before starting the next iteration.... \r\n\r\n" ) );
vTaskDelay( pdMS_TO_TICKS( mqttexampleMQTT_TIMEOUT_MS ) );
}
}
/*-----------------------------------------------------------*/
static void prvExample_OnDisconnect( void * pvCallbackContext,
IotMqttCallbackParam_t * pxCallbackParams )
{
TaskHandle_t xDemoTaskHandle = ( TaskHandle_t ) pvCallbackContext;
/* Ensure that we initiated the disconnect. */
configASSERT( pxCallbackParams->u.disconnectReason == IOT_MQTT_DISCONNECT_CALLED );
/* Inform the demo task about the disconnect. */
xTaskNotify( xDemoTaskHandle,
mqttexampleDISCONNECTED_BIT,
eSetBits /* Set the mqttexampleDISCONNECTED_BIT in the demo task's notification value. */
);
}
/*-----------------------------------------------------------*/
static void prvExample_OnMessageReceived( void * pvCallbackContext,
IotMqttCallbackParam_t * pxCallbackParams )
{
TaskHandle_t xDemoTaskHandle = ( TaskHandle_t ) pvCallbackContext;
/* Ensure the message is received on the expected topic. */
configASSERT( pxCallbackParams->u.message.info.topicNameLength == strlen( mqttexampleTOPIC ) );
configASSERT( strncmp( pxCallbackParams->u.message.info.pTopicName,
mqttexampleTOPIC,
strlen( mqttexampleTOPIC ) ) == 0 );
/* Ensure the message itself is as expected. */
configASSERT( pxCallbackParams->u.message.info.payloadLength == strlen( mqttexampleMESSAGE ) );
configASSERT( strncmp( pxCallbackParams->u.message.info.pPayload,
mqttexampleMESSAGE,
strlen( mqttexampleMESSAGE ) ) == 0 );
/* Ensure the message Quality of Service (QoS) is as expected. */
configASSERT( pxCallbackParams->u.message.info.qos == IOT_MQTT_QOS_1 );
/* So as not to worry about string lengths the print message uses the
* consts rather than the data from the message, but the asserts above have
* already checked the two are equal. */
configPRINTF( ( "Received %s on the topic %s\r\n", mqttexampleMESSAGE, mqttexampleTOPIC ) );
/* Inform the demo task about the message received from the MQTT broker by
* setting the mqttexampleMESSAGE_RECEIVED_BIT bit in the task's notification
* value. */
xTaskNotify( xDemoTaskHandle,
mqttexampleMESSAGE_RECEIVED_BIT,
eSetBits /* Set the mqttexampleMESSAGE_RECEIVED_BIT in the demo task's notification value. */
);
}
/*-----------------------------------------------------------*/
static void prvMQTTConnect( void )
{
IotMqttError_t xResult;
/* Set the context to pass into the disconnect callback function. */
xNetworkInfo.disconnectCallback.pCallbackContext = ( void * ) xTaskGetCurrentTaskHandle();
/* Establish the connection to the MQTT broker - It is a blocking call and
* will return only when connection is complete or a timeout occurs. The
* network and connection structures are declared and initialized at the top
* of this file. */
xResult = IotMqtt_Connect( &( xNetworkInfo ),
&( xConnectInfo ),
mqttexampleMQTT_TIMEOUT_MS,
&( xMQTTConnection ) );
configASSERT( xResult == IOT_MQTT_SUCCESS );
}
/*-----------------------------------------------------------*/
static void prvMQTTDisconnect( void )
{
/* Send a MQTT DISCONNECT packet to the MQTT broker to do a graceful
* disconnect. */
IotMqtt_Disconnect( xMQTTConnection,
0 /* flags - 0 means a graceful disconnect by sending MQTT DISCONNECT. */
);
}
/*-----------------------------------------------------------*/
static void prvMQTTSubscribe( void )
{
IotMqttError_t xResult;
IotMqttSubscription_t xMQTTSubscription;
/* Subscribe to the mqttexampleTOPIC topic filter. The task handle is passed
* as the callback context, which is then used by the callback to send a task
* notification to this task.*/
xMQTTSubscription.qos = IOT_MQTT_QOS_1;
xMQTTSubscription.pTopicFilter = mqttexampleTOPIC;
xMQTTSubscription.topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
xMQTTSubscription.callback.pCallbackContext = ( void * ) xTaskGetCurrentTaskHandle();
xMQTTSubscription.callback.function = prvExample_OnMessageReceived;
/* Use the synchronous API to subscribe - It is a blocking call and only
* returns when the subscribe operation is complete or a timeout occurs. */
xResult = IotMqtt_SubscribeSync( xMQTTConnection,
&( xMQTTSubscription ),
1, /* We are subscribing to one topic filter. */
0, /* flags - currently ignored. */
mqttexampleMQTT_TIMEOUT_MS );
configASSERT( xResult == IOT_MQTT_SUCCESS );
}
/*-----------------------------------------------------------*/
static void prvMQTTPublish( void )
{
IotMqttError_t xResult;
IotMqttPublishInfo_t xMQTTPublishInfo;
/* Publish a message with QoS1 on the mqttexampleTOPIC topic. Since we are
* subscribed to the same topic, the MQTT broker will send the same message
* back to us. It is verified in the publish callback. */
xMQTTPublishInfo.qos = IOT_MQTT_QOS_1;
xMQTTPublishInfo.retain = false;
xMQTTPublishInfo.pTopicName = mqttexampleTOPIC;
xMQTTPublishInfo.topicNameLength = ( uint16_t ) strlen( mqttexampleTOPIC );
xMQTTPublishInfo.pPayload = mqttexampleMESSAGE;
xMQTTPublishInfo.payloadLength = strlen( mqttexampleMESSAGE );
xMQTTPublishInfo.retryMs = mqttexamplePUBLISH_RETRY_MS;
xMQTTPublishInfo.retryLimit = mqttexamplePUBLISH_RETRY_LIMIT;
/* Use the synchronous API to publish - It is a blocking call and only
* returns when the publish operation is complete or a timeout occurs. */
xResult = IotMqtt_PublishSync( xMQTTConnection,
&( xMQTTPublishInfo ),
0, /* flags - currently ignored. */
mqttexampleMQTT_TIMEOUT_MS );
configASSERT( xResult == IOT_MQTT_SUCCESS );
}
/*-----------------------------------------------------------*/
static void prvMQTTUnsubscribe( void )
{
IotMqttError_t xResult;
IotMqttSubscription_t xMQTTSubscription;
/* Unsubscribe from the mqttexampleTOPIC topic filter. */
xMQTTSubscription.pTopicFilter = mqttexampleTOPIC;
xMQTTSubscription.topicFilterLength = ( uint16_t ) strlen( mqttexampleTOPIC );
/* The following members of the IotMqttSubscription_t are ignored by the
* unsubscribe operation. Just initialize them to avoid "use of uninitialized
* variable" warnings. */
xMQTTSubscription.qos = IOT_MQTT_QOS_1;
xMQTTSubscription.callback.pCallbackContext = NULL;
xMQTTSubscription.callback.function = NULL;
/* Use the synchronous API to unsubscribe - It is a blocking call and only
* returns when the unsubscribe operation is complete or a timeout occurs. */
xResult = IotMqtt_UnsubscribeSync( xMQTTConnection,
&( xMQTTSubscription ),
1, /* We are unsubscribing from one topic filter. */
0, /* flags - currently ignored. */
mqttexampleMQTT_TIMEOUT_MS );
configASSERT( xResult == IOT_MQTT_SUCCESS );
}
/*-----------------------------------------------------------*/
static void prvInitialiseLibraries( void )
{
IotTaskPoolError_t xTaskPoolResult;
IotMqttError_t xResult;
IotNetworkError_t xNetworkResult;
/* The MQTT library needs a task pool, so create the system task pool. */
xTaskPoolResult = IotTaskPool_CreateSystemTaskPool( &( xTaskPoolParameters ) );
configASSERT( xTaskPoolResult == IOT_TASKPOOL_SUCCESS );
/* Initialize the network stack abstraction for FreeRTOS. */
xNetworkResult = IotNetworkFreeRTOS_Init();
configASSERT( xNetworkResult == IOT_NETWORK_SUCCESS );
/* MQTT library must be initialized before it can be used. This is just one
* time initialization. */
xResult = IotMqtt_Init();
configASSERT( xResult == IOT_MQTT_SUCCESS );
}
/*-----------------------------------------------------------*/