Fix bugs in receiveSingleIteration and optimize sendMessageVector

docs/doxygen/include/size_table.md

@@ -10,7 +10,7 @@
     <tr>
         <td>core_mqtt.c</td>
         <td><center>3.9K</center></td>
-        <td><center>3.3K</center></td>
+        <td><center>3.4K</center></td>
     </tr>
     <tr>
         <td>core_mqtt_state.c</td>
@@ -25,6 +25,6 @@
     <tr>
         <td><b>Total estimates</b></td>
         <td><b><center>8.4K</center></b></td>
-        <td><b><center>6.8K</center></b></td>
+        <td><b><center>6.9K</center></b></td>
     </tr>
 </table>

lexicon.txt

@@ -17,6 +17,7 @@ bytesreceived
 bytesrecvd
 bytesremaining
 bytessent
+bytessentthisvector
 bytestoread
 bytestoreceive
 bytestorecv

source/core_mqtt.c

@@ -758,8 +758,7 @@ static int32_t sendMessageVector( MQTTContext_t * pContext,
     /* Reset the iterator to point to the first entry in the array. */
     pIoVectIterator = pIoVec;
 
-    while( ( pContext->getTime() < timeoutTime ) &&
-           ( bytesSentOrError < ( int32_t ) bytesToSend ) )
+    while( bytesSentOrError < ( int32_t ) bytesToSend )
     {
         int32_t sendResult;
         uint32_t bytesSentThisVector = 0U;
@@ -785,7 +784,10 @@ static int32_t sendMessageVector( MQTTContext_t * pContext,
         else
         {
             bytesSentOrError = sendResult;
-            break;
+
+            /* We do not need to break here as this condition is checked in
+             * the loop. The following code will not execute as bytesSentThisVector
+             * is not updated and is still 0. */
         }
 
         while( ( pIoVectIterator <= &( pIoVec[ ioVecCount - 1U ] ) ) &&
@@ -806,6 +808,12 @@ static int32_t sendMessageVector( MQTTContext_t * pContext,
             pIoVectIterator->iov_base = ( const void * ) &( ( ( const uint8_t * ) pIoVectIterator->iov_base )[ bytesSentThisVector ] );
             pIoVectIterator->iov_len -= bytesSentThisVector;
         }
+
+        /* Check for timeout. */
+        if( pContext->getTime() > timeoutTime )
+        {
+            break;
+        }
     }
 
     return bytesSentOrError;
@@ -816,7 +824,7 @@ static int32_t sendBuffer( MQTTContext_t * pContext,
                            size_t bytesToSend )
 {
     const uint8_t * pIndex = pBufferToSend;
-    size_t bytesRemaining = bytesToSend;
+    size_t bytesRemaining;
     int32_t totalBytesSent = 0, bytesSent;
     uint32_t lastSendTimeMs = 0U, timeSinceLastSendMs = 0U;
     bool sendError = false;
@@ -1650,11 +1658,15 @@ static MQTTStatus_t receiveSingleIteration( MQTTContext_t * pContext,
         /* The receive function has failed. Bubble up the error up to the user. */
         status = MQTTRecvFailed;
     }
-    else if( recvBytes == 0 )
+    else if( ( recvBytes == 0 ) && ( pContext->index == 0 ) )
     {
-        /* No more bytes available since the last read. */
+        /* No more bytes available since the last read and neither is anything in
+         * the buffer. */
         status = MQTTNoDataAvailable;
     }
+
+    /* Either something was received, or there is still data to be processed in the
+     * buffer, or both. */
     else
     {
         /* Update the number of bytes in the MQTT fixed buffer. */
@@ -2321,6 +2333,10 @@ static MQTTStatus_t handleSessionResumption( MQTTContext_t * pContext,
 
     assert( pContext != NULL );
 
+    /* Reset the index and clear the buffer when a new session is established. */
+    pContext->index = 0;
+    memset( pContext->networkBuffer.pBuffer, 0, pContext->networkBuffer.size );
+
     if( sessionPresent == true )
     {
         /* Get the next packet ID for which a PUBREL need to be resent. */
@@ -2969,6 +2985,10 @@ MQTTStatus_t MQTT_Disconnect( MQTTContext_t * pContext )
     {
         LogInfo( ( "Disconnected from the broker." ) );
         pContext->connectStatus = MQTTNotConnected;
+
+        /* Reset the index and clean the buffer on a successful disconnect. */
+        pContext->index = 0;
+        ( void ) memset( pContext->networkBuffer.pBuffer, 0, pContext->networkBuffer.size );
     }
 
     return status;

source/core_mqtt_serializer.c

@@ -868,6 +868,8 @@ static MQTTStatus_t processRemainingLength( const uint8_t * pBuffer,
         {
             remainingLength = MQTT_REMAINING_LENGTH_INVALID;
 
+            LogError( ( "Invalid remaining length in the packet.\n" ) );
+
             status = MQTTBadResponse;
         }
         else
@@ -904,6 +906,7 @@ static MQTTStatus_t processRemainingLength( const uint8_t * pBuffer,
 
         if( bytesDecoded != expectedSize )
         {
+            LogError( ( "Expected and actual length of decoded bytes do not match.\n" ) );
             status = MQTTBadResponse;
         }
         else

test/cbmc/proofs/MQTT_Disconnect/MQTT_Disconnect_harness.c

@@ -33,6 +33,8 @@ void harness()
 
     pContext = allocateMqttContext( NULL );
     __CPROVER_assume( isValidMqttContext( pContext ) );
+    __CPROVER_assume( pContext != NULL ); 
+    __CPROVER_assume( pContext->networkBuffer.pBuffer != NULL );
 
     MQTT_Disconnect( pContext );
 }

test/unit-test/core_mqtt_utest.c

@@ -2545,6 +2545,9 @@ void test_MQTT_Disconnect4( void )
     MQTTFixedBuffer_t networkBuffer = { 0 };
     size_t disconnectSize = 2;
 
+    /* Fill the buffer with garbage data. */
+    memset( mqttBuffer, 0xAB, MQTT_TEST_BUFFER_LENGTH );
+
     setupTransportInterface( &transport );
     setupNetworkBuffer( &networkBuffer );
     networkContext.buffer = &bufPtr;
@@ -2570,7 +2573,8 @@ void test_MQTT_Disconnect4( void )
 
     TEST_ASSERT_EQUAL( MQTTSuccess, status );
     TEST_ASSERT_EQUAL( MQTTNotConnected, mqttContext.connectStatus );
-    TEST_ASSERT_EQUAL_MEMORY( mqttBuffer, buffer, 1 );
+    /* At disconnect, the buffer is cleared of any pending packets. */
+    TEST_ASSERT_EACH_EQUAL_UINT8( 0, mqttBuffer, MQTT_TEST_BUFFER_LENGTH );
 }
 /* ========================================================================== */
 
@@ -3886,11 +3890,21 @@ void test_MQTT_ReceiveLoop( void )
     TEST_ASSERT_EQUAL( MQTTBadParameter, mqttStatus );
     setupNetworkBuffer( &( context.networkBuffer ) );
 
-    MQTT_ProcessIncomingPacketTypeAndLength_ExpectAnyArgsAndReturn( MQTTRecvFailed );
+    MQTT_ProcessIncomingPacketTypeAndLength_ExpectAnyArgsAndReturn( MQTTBadResponse );
     /* Error case, for branch coverage. */
     mqttStatus = MQTT_ReceiveLoop( &context );
-    TEST_ASSERT_EQUAL( MQTTRecvFailed, mqttStatus );
+    TEST_ASSERT_EQUAL( MQTTBadResponse, mqttStatus );
+
+    /* This will cover the case when there is data available in the buffer to be processed but
+     * no additional bytes have been received by the transport interface. */
+    context.transportInterface.recv = transportRecvNoData;
+    MQTT_ProcessIncomingPacketTypeAndLength_ExpectAnyArgsAndReturn( MQTTBadResponse );
+    /* Error case, for branch coverage. */
+    mqttStatus = MQTT_ReceiveLoop( &context );
+    TEST_ASSERT_EQUAL( MQTTBadResponse, mqttStatus );
 
+    /* Reset the index to clear the buffer of any remaining data. */
+    context.index = 0;
     /* Keep Alive should not trigger.*/
     context.keepAliveIntervalSec = 1;
     mqttStatus = MQTT_ReceiveLoop( &context );