Update demos for coreMQTT 2.1.0 and update submodule pointers to LTS 2.0 (#1823)

* Update for coreMQTT 2.1.0.

* coreMQTT demo fixes.

* Fix build files.

* Update submodule pointers for LTS 2.0.

* Fix transport.writev not being set to NULL for coreMQTT 2.1.0.

* Fleet provisioning clean up.

* Minor fix.

* Minor fix.

* Minor fix.

* Swap ProcessLoop with ProcessLoopWithTimeout.

* Remove unused variable from ProcessLoopWithTimeout.

* Minor fixes.

* Defender demo cleanup.

* Minor fix to defender demo.

* Minor clean up.

* Cleanup mqtt_demo_basic_tls.

* Minor cleanup.

* Fix conflicting type.

* Minor fixes.

* Cleanup of mqtt_demo_mutual_auth.

* Minor fixes.

* Minor fixes.

* Cleanup mqtt_demo_plaintext.

* Add QoS arrays to MQTT plaintext demo.

* Updated handle resubscribe for plaintext and mutual auth demo.

* Cleanup mqtt_demo_subscription_manager.c

* Minor fix to subscription manager demo.

* Fix formatting.

* Update OTA over HTTP for coreMQTT v2.1.0

* Minor fix to mqtt_demo_subscription_manager.c

* Fixes to OTA demos.

* Fixes to OTA demos.

* Fixes to OTA demos.

* Cmakelist fixes for OTA library.

* Cmake list fixes.

* Cmake list fixes.

* Cmake list fixes.

* Cmake lists fixes.

* Cmakelists fixes.

* Cmakelists fixes.

* Cmakelists fixes.

* Cmakelists fixes.

* CMakeList fixes.

* Cmake list fixes for ota over HTTP.

* Cmake list fixes for OTA over HTTP demo.

* OTA demo fixes.

* Update coreMQTT integration tests for coreMQTT v2.1.0.

* Fix ota_pal_posix_utest.c for OTA v3.4.0.

* Update shadow_system_test.c for coreMQTT v2.1.0.

* Fix shadow integration tests.

* Fix mqtt_system_test.c.

* Update lexicon.

* Update manifest.yml

* Update integration test lexicon.

* Update integration test lexicon.

* Clean up shadow demo.

* Remove trailing whitespace form install.cmake.

* Fix shadow demo.

* Fix writev not being initialized to NULL.

* Fix a test assert to check for receive failure instead of send failure since, coreMQTT process loop handles receiving first over handling keep alive.

* Remove wait for packet function from MQTT system test.

* Fixes to mqtt_system_test.c.

* Function naming cleanup.

Co-authored-by: Jason Carroll <[email protected]>

Author: jasonpcarroll

demos/defender/defender_demo_json/defender_demo.c

@@ -755,7 +755,7 @@ int main( int argc,
         {
             for( i = 0; i < DEFENDER_RESPONSE_WAIT_SECONDS; i++ )
             {
-                ( void ) ProcessLoop( 1000 );
+                ( void ) ProcessLoopWithTimeout( 1000 );
 
                 /* reportStatus is updated in the publishCallback. */
                 if( reportStatus != ReportStatusNotReceived )

demos/defender/defender_demo_json/mqtt_operations.c

@@ -178,6 +178,18 @@
  * @brief The length of the MQTT metrics string.
  */
 #define METRICS_STRING_LENGTH                    ( ( uint16_t ) ( sizeof( METRICS_STRING ) - 1 ) )
+
+/**
+ * @brief The length of the outgoing publish records array used by the coreMQTT
+ * library to track QoS > 0 packet ACKS for outgoing publishes.
+ */
+#define OUTGOING_PUBLISH_RECORD_LEN              ( 10U )
+
+/**
+ * @brief The length of the incoming publish records array used by the coreMQTT
+ * library to track QoS > 0 packet ACKS for incoming publishes.
+ */
+#define INCOMING_PUBLISH_RECORD_LEN              ( 10U )
 /*-----------------------------------------------------------*/
 
 /**
@@ -204,6 +216,13 @@ struct NetworkContext
 };
 /*-----------------------------------------------------------*/
 
+/**
+ * @brief Packet Identifier updated when an ACK packet is received.
+ *
+ * It is used to match an expected ACK for a transmitted packet.
+ */
+static uint16_t globalAckPacketIdentifier = 0U;
+
 /**
  * @brief Packet Identifier generated when Subscribe request was sent to the broker.
  *
@@ -258,6 +277,24 @@ static bool mqttSessionEstablished = false;
  * publish messages.
  */
 static MQTTPublishCallback_t appPublishCallback = NULL;
+
+/**
+ * @brief Array to track the outgoing publish records for outgoing publishes
+ * with QoS > 0.
+ *
+ * This is passed into #MQTT_InitStatefulQoS to allow for QoS > 0.
+ *
+ */
+static MQTTPubAckInfo_t pOutgoingPublishRecords[ OUTGOING_PUBLISH_RECORD_LEN ];
+
+/**
+ * @brief Array to track the incoming publish records for incoming publishes
+ * with QoS > 0.
+ *
+ * This is passed into #MQTT_InitStatefulQoS to allow for QoS > 0.
+ *
+ */
+static MQTTPubAckInfo_t pIncomingPublishRecords[ INCOMING_PUBLISH_RECORD_LEN ];
 /*-----------------------------------------------------------*/
 
 /**
@@ -336,13 +373,29 @@ static void mqttCallback( MQTTContext_t * pMqttContext,
  * false otherwise.
  */
 static bool handlePublishResend( MQTTContext_t * pMqttContext );
+
+/**
+ * @brief Wait for an expected ACK packet to be received.
+ *
+ * This function handles waiting for an expected ACK packet by calling
+ * #MQTT_ProcessLoop and waiting for #mqttCallback to set the global ACK
+ * packet identifier to the expected ACK packet identifier.
+ *
+ * @param[in] pMqttContext MQTT context pointer.
+ * @param[in] usPacketIdentifier Packet identifier for expected ACK packet.
+ * @param[in] ulTimeout Maximum duration to wait for expected ACK packet.
+ *
+ * @return true if the expected ACK packet was received, false otherwise.
+ */
+static bool waitForPacketAck( MQTTContext_t * pMqttContext,
+                              uint16_t usPacketIdentifier,
+                              uint32_t ulTimeout );
 /*-----------------------------------------------------------*/
 
 static uint32_t generateRandomNumber()
 {
     return( rand() );
 }
-
 /*-----------------------------------------------------------*/
 
 static bool connectToBrokerWithBackoffRetries( NetworkContext_t * pNetworkContext )
@@ -556,6 +609,9 @@ static void mqttCallback( MQTTContext_t * pMqttContext,
                 /* Make sure the ACK packet identifier matches with the request
                  * packet identifier. */
                 assert( globalSubscribePacketIdentifier == packetIdentifier );
+
+                /* Update the global ACK packet identifier. */
+                globalAckPacketIdentifier = packetIdentifier;
                 break;
 
             case MQTT_PACKET_TYPE_UNSUBACK:
@@ -564,6 +620,9 @@ static void mqttCallback( MQTTContext_t * pMqttContext,
                 /* Make sure the ACK packet identifier matches with the request
                  * packet identifier. */
                 assert( globalUnsubscribePacketIdentifier == packetIdentifier );
+
+                /* Update the global ACK packet identifier. */
+                globalAckPacketIdentifier = packetIdentifier;
                 break;
 
             case MQTT_PACKET_TYPE_PINGRESP:
@@ -578,6 +637,9 @@ static void mqttCallback( MQTTContext_t * pMqttContext,
                 LogDebug( ( "PUBACK received for packet id %u.",
                             packetIdentifier ) );
 
+                /* Update the global ACK packet identifier. */
+                globalAckPacketIdentifier = packetIdentifier;
+
                 /* Cleanup the publish packet from the #outgoingPublishPackets
                  * array when a PUBACK is received. */
                 cleanupOutgoingPublishWithPacketID( packetIdentifier );
@@ -641,6 +703,53 @@ static bool handlePublishResend( MQTTContext_t * pMqttContext )
 }
 /*-----------------------------------------------------------*/
 
+static bool waitForPacketAck( MQTTContext_t * pMqttContext,
+                              uint16_t usPacketIdentifier,
+                              uint32_t ulTimeout )
+{
+    uint32_t ulMqttProcessLoopEntryTime;
+    uint32_t ulMqttProcessLoopTimeoutTime;
+    uint32_t ulCurrentTime;
+
+    MQTTStatus_t eMqttStatus = MQTTSuccess;
+    bool xStatus = false;
+
+    /* Reset the ACK packet identifier being received. */
+    globalAckPacketIdentifier = 0U;
+
+    ulCurrentTime = pMqttContext->getTime();
+    ulMqttProcessLoopEntryTime = ulCurrentTime;
+    ulMqttProcessLoopTimeoutTime = ulCurrentTime + ulTimeout;
+
+    /* Call MQTT_ProcessLoop multiple times until the expected packet ACK
+     * is received, a timeout happens, or MQTT_ProcessLoop fails. */
+    while( ( globalAckPacketIdentifier != usPacketIdentifier ) &&
+           ( ulCurrentTime < ulMqttProcessLoopTimeoutTime ) &&
+           ( eMqttStatus == MQTTSuccess || eMqttStatus == MQTTNeedMoreBytes ) )
+    {
+        /* Event callback will set #globalAckPacketIdentifier when receiving
+         * appropriate packet. */
+        eMqttStatus = MQTT_ProcessLoop( pMqttContext );
+        ulCurrentTime = pMqttContext->getTime();
+    }
+
+    if( ( ( eMqttStatus != MQTTSuccess ) && ( eMqttStatus != MQTTNeedMoreBytes ) ) ||
+        ( globalAckPacketIdentifier != usPacketIdentifier ) )
+    {
+        LogError( ( "MQTT_ProcessLoop failed to receive ACK packet: Expected ACK Packet ID=%02X, LoopDuration=%u, Status=%s",
+                    usPacketIdentifier,
+                    ( ulCurrentTime - ulMqttProcessLoopEntryTime ),
+                    MQTT_Status_strerror( eMqttStatus ) ) );
+    }
+    else
+    {
+        xStatus = true;
+    }
+
+    return xStatus;
+}
+/*-----------------------------------------------------------*/
+
 bool EstablishMqttSession( MQTTPublishCallback_t publishCallback )
 {
     bool returnStatus = false;
@@ -678,6 +787,7 @@ bool EstablishMqttSession( MQTTPublishCallback_t publishCallback )
         transport.pNetworkContext = pNetworkContext;
         transport.send = Openssl_Send;
         transport.recv = Openssl_Recv;
+        transport.writev = NULL;
 
         /* Fill the values for network buffer. */
         networkBuffer.pBuffer = buffer;
@@ -696,55 +806,70 @@ bool EstablishMqttSession( MQTTPublishCallback_t publishCallback )
         if( mqttStatus != MQTTSuccess )
         {
             returnStatus = false;
-            LogError( ( "MQTT init failed with status %s.",
+            LogError( ( "MQTT_Init failed with status %s.",
                         MQTT_Status_strerror( mqttStatus ) ) );
         }
         else
         {
-            /* Establish an MQTT session by sending a CONNECT packet. */
-
-            /* If #createCleanSession is true, start with a clean session
-             * i.e. direct the MQTT broker to discard any previous session data.
-             * If #createCleanSession is false, direct the broker to attempt to
-             * reestablish a session which was already present. */
-            connectInfo.cleanSession = createCleanSession;
-
-            /* 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. */
-            connectInfo.pClientIdentifier = CLIENT_IDENTIFIER;
-            connectInfo.clientIdentifierLength = CLIENT_IDENTIFIER_LENGTH;
-
-            /* The maximum time interval in seconds which is allowed to elapse
-             * between two Control Packets.
-             * It is the responsibility of the client to ensure that the interval between
-             * control packets being sent does not exceed the this keep-alive value. In the
-             * absence of sending any other control packets, the client MUST send a
-             * PINGREQ packet. */
-            connectInfo.keepAliveSeconds = MQTT_KEEP_ALIVE_INTERVAL_SECONDS;
-
-            /* Username and password for authentication. Not used in this demo. */
-            connectInfo.pUserName = METRICS_STRING;
-            connectInfo.userNameLength = METRICS_STRING_LENGTH;
-            connectInfo.pPassword = NULL;
-            connectInfo.passwordLength = 0U;
-
-            /* Send an MQTT CONNECT packet to the broker. */
-            mqttStatus = MQTT_Connect( pMqttContext,
-                                       &connectInfo,
-                                       NULL,
-                                       CONNACK_RECV_TIMEOUT_MS,
-                                       &sessionPresent );
+            mqttStatus = MQTT_InitStatefulQoS( pMqttContext,
+                                               pOutgoingPublishRecords,
+                                               OUTGOING_PUBLISH_RECORD_LEN,
+                                               pIncomingPublishRecords,
+                                               INCOMING_PUBLISH_RECORD_LEN );
 
             if( mqttStatus != MQTTSuccess )
             {
                 returnStatus = false;
-                LogError( ( "Connection with MQTT broker failed with status %s.",
+                LogError( ( "MQTT_InitStatefulQoS failed with status %s.",
                             MQTT_Status_strerror( mqttStatus ) ) );
             }
             else
             {
-                LogDebug( ( "MQTT connection successfully established with broker." ) );
+                /* Establish an MQTT session by sending a CONNECT packet. */
+
+                /* If #createCleanSession is true, start with a clean session
+                 * i.e. direct the MQTT broker to discard any previous session data.
+                 * If #createCleanSession is false, direct the broker to attempt to
+                 * reestablish a session which was already present. */
+                connectInfo.cleanSession = createCleanSession;
+
+                /* 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. */
+                connectInfo.pClientIdentifier = CLIENT_IDENTIFIER;
+                connectInfo.clientIdentifierLength = CLIENT_IDENTIFIER_LENGTH;
+
+                /* The maximum time interval in seconds which is allowed to elapse
+                 * between two Control Packets.
+                 * It is the responsibility of the client to ensure that the interval between
+                 * control packets being sent does not exceed the this keep-alive value. In the
+                 * absence of sending any other control packets, the client MUST send a
+                 * PINGREQ packet. */
+                connectInfo.keepAliveSeconds = MQTT_KEEP_ALIVE_INTERVAL_SECONDS;
+
+                /* Username and password for authentication. Not used in this demo. */
+                connectInfo.pUserName = METRICS_STRING;
+                connectInfo.userNameLength = METRICS_STRING_LENGTH;
+                connectInfo.pPassword = NULL;
+                connectInfo.passwordLength = 0U;
+
+                /* Send an MQTT CONNECT packet to the broker. */
+                mqttStatus = MQTT_Connect( pMqttContext,
+                                           &connectInfo,
+                                           NULL,
+                                           CONNACK_RECV_TIMEOUT_MS,
+                                           &sessionPresent );
+
+                if( mqttStatus != MQTTSuccess )
+                {
+                    returnStatus = false;
+                    LogError( ( "Connection with MQTT broker failed with status %s.",
+                                MQTT_Status_strerror( mqttStatus ) ) );
+                }
+                else
+                {
+                    LogDebug( ( "MQTT connection successfully established with broker." ) );
+                }
             }
         }
 
@@ -867,17 +992,9 @@ bool SubscribeToTopic( const char * pTopicFilter,
          * of receiving publish message before subscribe ack is zero; but application
          * must be ready to receive any packet. This demo uses MQTT_ProcessLoop to
          * receive packet from network. */
-        mqttStatus = MQTT_ProcessLoop( pMqttContext, MQTT_PROCESS_LOOP_TIMEOUT_MS );
-
-        if( mqttStatus != MQTTSuccess )
-        {
-            LogError( ( "MQTT_ProcessLoop returned with status = %s.",
-                        MQTT_Status_strerror( mqttStatus ) ) );
-        }
-        else
-        {
-            returnStatus = true;
-        }
+        returnStatus = waitForPacketAck( pMqttContext,
+                                         globalSubscribePacketIdentifier,
+                                         MQTT_PROCESS_LOOP_TIMEOUT_MS );
     }
 
     return returnStatus;
@@ -927,17 +1044,9 @@ bool UnsubscribeFromTopic( const char * pTopicFilter,
         /* Process incoming packet from the broker. Acknowledgment for unsubscribe
          * operation ( UNSUBACK ) will be received here. This demo uses
          * MQTT_ProcessLoop to receive packet from network. */
-        mqttStatus = MQTT_ProcessLoop( pMqttContext, MQTT_PROCESS_LOOP_TIMEOUT_MS );
-
-        if( mqttStatus != MQTTSuccess )
-        {
-            LogError( ( "MQTT_ProcessLoop returned with status = %s.",
-                        MQTT_Status_strerror( mqttStatus ) ) );
-        }
-        else
-        {
-            returnStatus = true;
-        }
+        returnStatus = waitForPacketAck( pMqttContext,
+                                         globalUnsubscribePacketIdentifier,
+                                         MQTT_PROCESS_LOOP_TIMEOUT_MS );
     }
 
     return returnStatus;
@@ -1009,17 +1118,30 @@ bool PublishToTopic( const char * pTopicFilter,
 }
 /*-----------------------------------------------------------*/
 
-bool ProcessLoop( uint32_t timeoutMs )
+bool ProcessLoopWithTimeout( uint32_t ulTimeoutMs )
 {
+    uint32_t ulMqttProcessLoopTimeoutTime;
+    uint32_t ulCurrentTime;
+
+    MQTTStatus_t eMqttStatus = MQTTSuccess;
     bool returnStatus = false;
-    MQTTStatus_t mqttStatus = MQTTSuccess;
 
-    mqttStatus = MQTT_ProcessLoop( &mqttContext, timeoutMs );
+    ulCurrentTime = mqttContext.getTime();
+    ulMqttProcessLoopTimeoutTime = ulCurrentTime + ulTimeoutMs;
 
-    if( mqttStatus != MQTTSuccess )
+    /* Call MQTT_ProcessLoop multiple times until the timeout expires or
+     * #MQTT_ProcessLoop fails. */
+    while( ( ulCurrentTime < ulMqttProcessLoopTimeoutTime ) &&
+           ( eMqttStatus == MQTTSuccess || eMqttStatus == MQTTNeedMoreBytes ) )
+    {
+        eMqttStatus = MQTT_ProcessLoop( &mqttContext );
+        ulCurrentTime = mqttContext.getTime();
+    }
+
+    if( ( eMqttStatus != MQTTSuccess ) && ( eMqttStatus != MQTTNeedMoreBytes ) )
     {
         LogError( ( "MQTT_ProcessLoop returned with status = %s.",
-                    MQTT_Status_strerror( mqttStatus ) ) );
+                    MQTT_Status_strerror( eMqttStatus ) ) );
     }
     else
     {

demos/defender/defender_demo_json/mqtt_operations.h

@@ -100,6 +100,6 @@ bool PublishToTopic( const char * pTopic,
  * @return true if process loop was successful;
  * false otherwise.
  */
-bool ProcessLoop( uint32_t timemoutMs );
+bool ProcessLoopWithTimeout( uint32_t ulTimeoutMs );
 
 #endif /* ifndef MQTT_OPERATIONS_H_ */

demos/fleet_provisioning/fleet_provisioning_with_csr/fleet_provisioning_with_csr_demo.c

@@ -318,7 +318,7 @@ static bool waitForResponse( void )
     responseStatus = ResponseNotReceived;
 
     /* responseStatus is updated from the MQTT publish callback. */
-    ( void ) ProcessLoop();
+    ( void ) ProcessLoopWithTimeout();
 
     if( responseStatus == ResponseNotReceived )
     {