git.sur5r.net Git - freertos/blob - FreeRTOS-Labs/Source/FreeRTOS-IoT-Libraries/c_sdk/standard/mqtt/src/iot_mqtt_network.c

   1 /*\r
   2  * IoT MQTT V2.1.0\r
   3  * Copyright (C) 2019 Amazon.com, Inc. or its affiliates.  All Rights Reserved.\r
   4  *\r
   5  * Permission is hereby granted, free of charge, to any person obtaining a copy of\r
   6  * this software and associated documentation files (the "Software"), to deal in\r
   7  * the Software without restriction, including without limitation the rights to\r
   8  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of\r
   9  * the Software, and to permit persons to whom the Software is furnished to do so,\r
  10  * subject to the following conditions:\r
  11  *\r
  12  * The above copyright notice and this permission notice shall be included in all\r
  13  * copies or substantial portions of the Software.\r
  14  *\r
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\r
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS\r
  17  * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR\r
  18  * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER\r
  19  * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN\r
  20  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.\r
  21  */\r
  22 \r
  23 /**\r
  24  * @file iot_mqtt_network.c\r
  25  * @brief Implements functions involving transport layer connections.\r
  26  */\r
  27 \r
  28 /* The config header is always included first. */\r
  29 #include "iot_config.h"\r
  30 \r
  31 /* Standard includes. */\r
  32 #include <string.h>\r
  33 \r
  34 /* Error handling include. */\r
  35 #include "iot_error.h"\r
  36 \r
  37 /* MQTT internal include. */\r
  38 #include "private/iot_mqtt_internal.h"\r
  39 \r
  40 /* Platform layer includes. */\r
  41 #include "platform/iot_threads.h"\r
  42 \r
  43 /* Atomics include. */\r
  44 #include "iot_atomic.h"\r
  45 \r
  46 /*-----------------------------------------------------------*/\r
  47 \r
  48 /**\r
  49  * @brief Check if an incoming packet type is valid.\r
  50  *\r
  51  * @param[in] packetType The packet type to check.\r
  52  *\r
  53  * @return `true` if the packet type is valid; `false` otherwise.\r
  54  */\r
  55 static bool _incomingPacketValid( uint8_t packetType );\r
  56 \r
  57 /**\r
  58  * @brief Get an incoming MQTT packet from the network.\r
  59  *\r
  60  * @param[in] pNetworkConnection Network connection to use for receive, which\r
  61  * may be different from the network connection associated with the MQTT connection.\r
  62  * @param[in] pMqttConnection The associated MQTT connection.\r
  63  * @param[out] pIncomingPacket Output parameter for the incoming packet.\r
  64  *\r
  65  * @return #IOT_MQTT_SUCCESS, #IOT_MQTT_NO_MEMORY or #IOT_MQTT_BAD_RESPONSE.\r
  66  */\r
  67 static IotMqttError_t _getIncomingPacket( void * pNetworkConnection,\r
  68                                           const _mqttConnection_t * pMqttConnection,\r
  69                                           _mqttPacket_t * pIncomingPacket );\r
  70 \r
  71 /**\r
  72  * @brief Deserialize a packet received from the network.\r
  73  *\r
  74  * @param[in] pMqttConnection The associated MQTT connection.\r
  75  * @param[in] pIncomingPacket The packet received from the network.\r
  76  *\r
  77  * @return #IOT_MQTT_SUCCESS, #IOT_MQTT_NO_MEMORY, #IOT_MQTT_NETWORK_ERROR,\r
  78  * #IOT_MQTT_SCHEDULING_ERROR, #IOT_MQTT_BAD_RESPONSE, or #IOT_MQTT_SERVER_REFUSED.\r
  79  */\r
  80 static IotMqttError_t _deserializeIncomingPacket( _mqttConnection_t * pMqttConnection,\r
  81                                                   _mqttPacket_t * pIncomingPacket );\r
  82 \r
  83 /**\r
  84  * @brief Send a PUBACK for a received QoS 1 PUBLISH packet.\r
  85  *\r
  86  * @param[in] pMqttConnection Which connection the PUBACK should be sent over.\r
  87  * @param[in] packetIdentifier Which packet identifier to include in PUBACK.\r
  88  */\r
  89 static void _sendPuback( _mqttConnection_t * pMqttConnection,\r
  90                          uint16_t packetIdentifier );\r
  91 \r
  92 /**\r
  93  * @brief Flush a packet from the stream of incoming data.\r
  94  *\r
  95  * This function is called when memory for a packet cannot be allocated. The\r
  96  * packet is flushed from the stream of incoming data so that the next packet\r
  97  * may be read.\r
  98  *\r
  99  * @param[in] pNetworkConnection Network connection to use for receive, which\r
 100  * may be different from the network connection associated with the MQTT connection.\r
 101  * @param[in] pMqttConnection The associated MQTT connection.\r
 102  * @param[in] length The length of the packet to flush.\r
 103  */\r
 104 static void _flushPacket( void * pNetworkConnection,\r
 105                           const _mqttConnection_t * pMqttConnection,\r
 106                           size_t length );\r
 107 \r
 108 /**\r
 109  * @cond DOXYGEN_IGNORE\r
 110  * Doxygen should ignore this section.\r
 111  *\r
 112  * Declaration of local MQTT serializer override selectors\r
 113  */\r
 114 #if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1\r
 115     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttGetPacketType_t,\r
 116                                    _getPacketTypeFunc,\r
 117                                    _IotMqtt_GetPacketType,\r
 118                                    getPacketType )\r
 119     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttGetRemainingLength_t,\r
 120                                    _getRemainingLengthFunc,\r
 121                                    _IotMqtt_GetRemainingLength,\r
 122                                    getRemainingLength )\r
 123     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttDeserialize_t,\r
 124                                    _getConnackDeserializer,\r
 125                                    _IotMqtt_DeserializeConnack,\r
 126                                    deserialize.connack )\r
 127     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttDeserialize_t,\r
 128                                    _getPublishDeserializer,\r
 129                                    _IotMqtt_DeserializePublish,\r
 130                                    deserialize.publish )\r
 131     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttDeserialize_t,\r
 132                                    _getPubackDeserializer,\r
 133                                    _IotMqtt_DeserializePuback,\r
 134                                    deserialize.puback )\r
 135     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttDeserialize_t,\r
 136                                    _getSubackDeserializer,\r
 137                                    _IotMqtt_DeserializeSuback,\r
 138                                    deserialize.suback )\r
 139     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttDeserialize_t,\r
 140                                    _getUnsubackDeserializer,\r
 141                                    _IotMqtt_DeserializeUnsuback,\r
 142                                    deserialize.unsuback )\r
 143     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttDeserialize_t,\r
 144                                    _getPingrespDeserializer,\r
 145                                    _IotMqtt_DeserializePingresp,\r
 146                                    deserialize.pingresp )\r
 147     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttSerializePuback_t,\r
 148                                    _getMqttPubackSerializer,\r
 149                                    _IotMqtt_SerializePuback,\r
 150                                    serialize.puback )\r
 151     _SERIALIZER_OVERRIDE_SELECTOR( IotMqttFreePacket_t,\r
 152                                    _getMqttFreePacketFunc,\r
 153                                    _IotMqtt_FreePacket,\r
 154                                    freePacket )\r
 155 #else /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */\r
 156     #define _getPacketTypeFunc( pSerializer )          _IotMqtt_GetPacketType\r
 157     #define _getRemainingLengthFunc( pSerializer )     _IotMqtt_GetRemainingLength\r
 158     #define _getConnackDeserializer( pSerializer )     _IotMqtt_DeserializeConnack\r
 159     #define _getPublishDeserializer( pSerializer )     _IotMqtt_DeserializePublish\r
 160     #define _getPubackDeserializer( pSerializer )      _IotMqtt_DeserializePuback\r
 161     #define _getSubackDeserializer( pSerializer )      _IotMqtt_DeserializeSuback\r
 162     #define _getUnsubackDeserializer( pSerializer )    _IotMqtt_DeserializeUnsuback\r
 163     #define _getPingrespDeserializer( pSerializer )    _IotMqtt_DeserializePingresp\r
 164     #define _getMqttPubackSerializer( pSerializer )    _IotMqtt_SerializePuback\r
 165     #define _getMqttFreePacketFunc( pSerializer )      _IotMqtt_FreePacket\r
 166 #endif /* if IOT_MQTT_ENABLE_SERIALIZER_OVERRIDES == 1 */\r
 167 /** @endcond */\r
 168 \r
 169 /*-----------------------------------------------------------*/\r
 170 \r
 171 static bool _incomingPacketValid( uint8_t packetType )\r
 172 {\r
 173     bool status = true;\r
 174 \r
 175     /* Check packet type. Mask out lower bits to ignore flags. */\r
 176     switch( packetType & 0xf0 )\r
 177     {\r
 178         /* Valid incoming packet types. */\r
 179         case MQTT_PACKET_TYPE_CONNACK:\r
 180         case MQTT_PACKET_TYPE_PUBLISH:\r
 181         case MQTT_PACKET_TYPE_PUBACK:\r
 182         case MQTT_PACKET_TYPE_SUBACK:\r
 183         case MQTT_PACKET_TYPE_UNSUBACK:\r
 184         case MQTT_PACKET_TYPE_PINGRESP:\r
 185             break;\r
 186 \r
 187         /* Any other packet type is invalid. */\r
 188         default:\r
 189             status = false;\r
 190             break;\r
 191     }\r
 192 \r
 193     return status;\r
 194 }\r
 195 \r
 196 /*-----------------------------------------------------------*/\r
 197 \r
 198 static IotMqttError_t _getIncomingPacket( void * pNetworkConnection,\r
 199                                           const _mqttConnection_t * pMqttConnection,\r
 200                                           _mqttPacket_t * pIncomingPacket )\r
 201 {\r
 202     IOT_FUNCTION_ENTRY( IotMqttError_t, IOT_MQTT_SUCCESS );\r
 203     size_t dataBytesRead = 0;\r
 204 \r
 205     /* No buffer for remaining data should be allocated. */\r
 206     IotMqtt_Assert( pIncomingPacket->pRemainingData == NULL );\r
 207     IotMqtt_Assert( pIncomingPacket->remainingLength == 0 );\r
 208 \r
 209     /* Read the packet type, which is the first byte available. */\r
 210     pIncomingPacket->type = _getPacketTypeFunc( pMqttConnection->pSerializer )( pNetworkConnection,\r
 211                                                                                 pMqttConnection->pNetworkInterface );\r
 212 \r
 213     /* Check that the incoming packet type is valid. */\r
 214     if( _incomingPacketValid( pIncomingPacket->type ) == false )\r
 215     {\r
 216         IotLogError( "(MQTT connection %p) Unknown packet type %02x received.",\r
 217                      pMqttConnection,\r
 218                      pIncomingPacket->type );\r
 219 \r
 220         IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_RESPONSE );\r
 221     }\r
 222     else\r
 223     {\r
 224         EMPTY_ELSE_MARKER;\r
 225     }\r
 226 \r
 227     /* Read the remaining length. */\r
 228     pIncomingPacket->remainingLength = _getRemainingLengthFunc( pMqttConnection->pSerializer )( pNetworkConnection,\r
 229                                                                                                 pMqttConnection->pNetworkInterface );\r
 230 \r
 231     if( pIncomingPacket->remainingLength == MQTT_REMAINING_LENGTH_INVALID )\r
 232     {\r
 233         IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_RESPONSE );\r
 234     }\r
 235     else\r
 236     {\r
 237         EMPTY_ELSE_MARKER;\r
 238     }\r
 239 \r
 240     /* Allocate a buffer for the remaining data and read the data. */\r
 241     if( pIncomingPacket->remainingLength > 0 )\r
 242     {\r
 243         pIncomingPacket->pRemainingData = IotMqtt_MallocMessage( pIncomingPacket->remainingLength );\r
 244 \r
 245         if( pIncomingPacket->pRemainingData == NULL )\r
 246         {\r
 247             IotLogError( "(MQTT connection %p) Failed to allocate buffer of length "\r
 248                          "%lu for incoming packet type %lu.",\r
 249                          pMqttConnection,\r
 250                          ( unsigned long ) pIncomingPacket->remainingLength,\r
 251                          ( unsigned long ) pIncomingPacket->type );\r
 252 \r
 253             _flushPacket( pNetworkConnection, pMqttConnection, pIncomingPacket->remainingLength );\r
 254 \r
 255             IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_NO_MEMORY );\r
 256         }\r
 257         else\r
 258         {\r
 259             EMPTY_ELSE_MARKER;\r
 260         }\r
 261 \r
 262         dataBytesRead = pMqttConnection->pNetworkInterface->receive( pNetworkConnection,\r
 263                                                                      pIncomingPacket->pRemainingData,\r
 264                                                                      pIncomingPacket->remainingLength );\r
 265 \r
 266         if( dataBytesRead != pIncomingPacket->remainingLength )\r
 267         {\r
 268             IOT_SET_AND_GOTO_CLEANUP( IOT_MQTT_BAD_RESPONSE );\r
 269         }\r
 270         else\r
 271         {\r
 272             EMPTY_ELSE_MARKER;\r
 273         }\r
 274     }\r
 275     else\r
 276     {\r
 277         EMPTY_ELSE_MARKER;\r
 278     }\r
 279 \r
 280     /* Clean up on error. */\r
 281     IOT_FUNCTION_CLEANUP_BEGIN();\r
 282 \r
 283     if( status != IOT_MQTT_SUCCESS )\r
 284     {\r
 285         if( pIncomingPacket->pRemainingData != NULL )\r
 286         {\r
 287             IotMqtt_FreeMessage( pIncomingPacket->pRemainingData );\r
 288         }\r
 289         else\r
 290         {\r
 291             EMPTY_ELSE_MARKER;\r
 292         }\r
 293     }\r
 294     else\r
 295     {\r
 296         EMPTY_ELSE_MARKER;\r
 297     }\r
 298 \r
 299     IOT_FUNCTION_CLEANUP_END();\r
 300 }\r
 301 \r
 302 /*-----------------------------------------------------------*/\r
 303 \r
 304 static IotMqttError_t _deserializeIncomingPacket( _mqttConnection_t * pMqttConnection,\r
 305                                                   _mqttPacket_t * pIncomingPacket )\r
 306 {\r
 307     IotMqttError_t status = IOT_MQTT_STATUS_PENDING;\r
 308     _mqttOperation_t * pOperation = NULL;\r
 309 \r
 310     /* A buffer for remaining data must be allocated if remaining length is not 0. */\r
 311     IotMqtt_Assert( ( pIncomingPacket->remainingLength > 0 ) ==\r
 312                     ( pIncomingPacket->pRemainingData != NULL ) );\r
 313 \r
 314     /* Only valid packets should be given to this function. */\r
 315     IotMqtt_Assert( _incomingPacketValid( pIncomingPacket->type ) == true );\r
 316 \r
 317     /* Mask out the low bits of packet type to ignore flags. */\r
 318     switch( ( pIncomingPacket->type & 0xf0 ) )\r
 319     {\r
 320         case MQTT_PACKET_TYPE_CONNACK:\r
 321             IotLogDebug( "(MQTT connection %p) CONNACK in data stream.", pMqttConnection );\r
 322 \r
 323             /* Deserialize CONNACK and notify of result. */\r
 324             status = _getConnackDeserializer( pMqttConnection->pSerializer )( pIncomingPacket );\r
 325 \r
 326             pOperation = _IotMqtt_FindOperation( pMqttConnection,\r
 327                                                  IOT_MQTT_CONNECT,\r
 328                                                  NULL );\r
 329 \r
 330             if( pOperation != NULL )\r
 331             {\r
 332                 pOperation->u.operation.status = status;\r
 333                 _IotMqtt_Notify( pOperation );\r
 334             }\r
 335             else\r
 336             {\r
 337                 EMPTY_ELSE_MARKER;\r
 338             }\r
 339 \r
 340             break;\r
 341 \r
 342         case MQTT_PACKET_TYPE_PUBLISH:\r
 343             IotLogDebug( "(MQTT connection %p) PUBLISH in data stream.", pMqttConnection );\r
 344 \r
 345             /* Allocate memory to handle the incoming PUBLISH. */\r
 346             pOperation = IotMqtt_MallocOperation( sizeof( _mqttOperation_t ) );\r
 347 \r
 348             if( pOperation == NULL )\r
 349             {\r
 350                 IotLogWarn( "Failed to allocate memory for incoming PUBLISH." );\r
 351                 status = IOT_MQTT_NO_MEMORY;\r
 352 \r
 353                 break;\r
 354             }\r
 355             else\r
 356             {\r
 357                 /* Set the members of the incoming PUBLISH operation. */\r
 358                 ( void ) memset( pOperation, 0x00, sizeof( _mqttOperation_t ) );\r
 359                 pOperation->incomingPublish = true;\r
 360                 pOperation->pMqttConnection = pMqttConnection;\r
 361                 pIncomingPacket->u.pIncomingPublish = pOperation;\r
 362             }\r
 363 \r
 364             /* Deserialize incoming PUBLISH. */\r
 365             status = _getPublishDeserializer( pMqttConnection->pSerializer )( pIncomingPacket );\r
 366 \r
 367             if( status == IOT_MQTT_SUCCESS )\r
 368             {\r
 369                 /* Send a PUBACK for QoS 1 PUBLISH. */\r
 370                 if( pOperation->u.publish.publishInfo.qos == IOT_MQTT_QOS_1 )\r
 371                 {\r
 372                     _sendPuback( pMqttConnection, pIncomingPacket->packetIdentifier );\r
 373                 }\r
 374                 else\r
 375                 {\r
 376                     EMPTY_ELSE_MARKER;\r
 377                 }\r
 378 \r
 379                 /* Transfer ownership of the received MQTT packet to the PUBLISH operation. */\r
 380                 pOperation->u.publish.pReceivedData = pIncomingPacket->pRemainingData;\r
 381                 pIncomingPacket->pRemainingData = NULL;\r
 382 \r
 383                 /* Add the PUBLISH to the list of operations pending processing. */\r
 384                 IotMutex_Lock( &( pMqttConnection->referencesMutex ) );\r
 385                 IotListDouble_InsertHead( &( pMqttConnection->pendingProcessing ),\r
 386                                           &( pOperation->link ) );\r
 387                 IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );\r
 388 \r
 389                 /* Increment the MQTT connection reference count before scheduling an\r
 390                  * incoming PUBLISH. */\r
 391                 if( _IotMqtt_IncrementConnectionReferences( pMqttConnection ) == true )\r
 392                 {\r
 393                     /* Schedule PUBLISH for callback invocation. */\r
 394                     status = _IotMqtt_ScheduleOperation( pOperation, _IotMqtt_ProcessIncomingPublish, 0 );\r
 395                 }\r
 396                 else\r
 397                 {\r
 398                     status = IOT_MQTT_NETWORK_ERROR;\r
 399                 }\r
 400             }\r
 401             else\r
 402             {\r
 403                 EMPTY_ELSE_MARKER;\r
 404             }\r
 405 \r
 406             /* Free PUBLISH operation on error. */\r
 407             if( status != IOT_MQTT_SUCCESS )\r
 408             {\r
 409                 /* Check ownership of the received MQTT packet. */\r
 410                 if( pOperation->u.publish.pReceivedData != NULL )\r
 411                 {\r
 412                     /* Retrieve the pointer MQTT packet pointer so it may be freed later. */\r
 413                     IotMqtt_Assert( pIncomingPacket->pRemainingData == NULL );\r
 414                     pIncomingPacket->pRemainingData = ( uint8_t * ) pOperation->u.publish.pReceivedData;\r
 415                 }\r
 416                 else\r
 417                 {\r
 418                     EMPTY_ELSE_MARKER;\r
 419                 }\r
 420 \r
 421                 /* Remove operation from pending processing list. */\r
 422                 IotMutex_Lock( &( pMqttConnection->referencesMutex ) );\r
 423 \r
 424                 if( IotLink_IsLinked( &( pOperation->link ) ) == true )\r
 425                 {\r
 426                     IotListDouble_Remove( &( pOperation->link ) );\r
 427                 }\r
 428                 else\r
 429                 {\r
 430                     EMPTY_ELSE_MARKER;\r
 431                 }\r
 432 \r
 433                 IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );\r
 434 \r
 435                 IotMqtt_Assert( pOperation != NULL );\r
 436                 IotMqtt_FreeOperation( pOperation );\r
 437             }\r
 438             else\r
 439             {\r
 440                 EMPTY_ELSE_MARKER;\r
 441             }\r
 442 \r
 443             break;\r
 444 \r
 445         case MQTT_PACKET_TYPE_PUBACK:\r
 446             IotLogDebug( "(MQTT connection %p) PUBACK in data stream.", pMqttConnection );\r
 447 \r
 448             /* Deserialize PUBACK and notify of result. */\r
 449             status = _getPubackDeserializer( pMqttConnection->pSerializer )( pIncomingPacket );\r
 450 \r
 451             pOperation = _IotMqtt_FindOperation( pMqttConnection,\r
 452                                                  IOT_MQTT_PUBLISH_TO_SERVER,\r
 453                                                  &( pIncomingPacket->packetIdentifier ) );\r
 454 \r
 455             if( pOperation != NULL )\r
 456             {\r
 457                 pOperation->u.operation.status = status;\r
 458                 _IotMqtt_Notify( pOperation );\r
 459             }\r
 460             else\r
 461             {\r
 462                 EMPTY_ELSE_MARKER;\r
 463             }\r
 464 \r
 465             break;\r
 466 \r
 467         case MQTT_PACKET_TYPE_SUBACK:\r
 468             IotLogDebug( "(MQTT connection %p) SUBACK in data stream.", pMqttConnection );\r
 469 \r
 470             /* Deserialize SUBACK and notify of result. */\r
 471             pIncomingPacket->u.pMqttConnection = pMqttConnection;\r
 472 \r
 473             status = _getSubackDeserializer( pMqttConnection->pSerializer )( pIncomingPacket );\r
 474 \r
 475             pOperation = _IotMqtt_FindOperation( pMqttConnection,\r
 476                                                  IOT_MQTT_SUBSCRIBE,\r
 477                                                  &( pIncomingPacket->packetIdentifier ) );\r
 478 \r
 479             if( pOperation != NULL )\r
 480             {\r
 481                 pOperation->u.operation.status = status;\r
 482                 _IotMqtt_Notify( pOperation );\r
 483             }\r
 484             else\r
 485             {\r
 486                 EMPTY_ELSE_MARKER;\r
 487             }\r
 488 \r
 489             break;\r
 490 \r
 491         case MQTT_PACKET_TYPE_UNSUBACK:\r
 492             IotLogDebug( "(MQTT connection %p) UNSUBACK in data stream.", pMqttConnection );\r
 493 \r
 494             /* Deserialize UNSUBACK and notify of result. */\r
 495             status = _getUnsubackDeserializer( pMqttConnection->pSerializer )( pIncomingPacket );\r
 496 \r
 497             pOperation = _IotMqtt_FindOperation( pMqttConnection,\r
 498                                                  IOT_MQTT_UNSUBSCRIBE,\r
 499                                                  &( pIncomingPacket->packetIdentifier ) );\r
 500 \r
 501             if( pOperation != NULL )\r
 502             {\r
 503                 pOperation->u.operation.status = status;\r
 504                 _IotMqtt_Notify( pOperation );\r
 505             }\r
 506             else\r
 507             {\r
 508                 EMPTY_ELSE_MARKER;\r
 509             }\r
 510 \r
 511             break;\r
 512 \r
 513         default:\r
 514             /* The only remaining valid type is PINGRESP. */\r
 515             IotMqtt_Assert( ( pIncomingPacket->type & 0xf0 ) == MQTT_PACKET_TYPE_PINGRESP );\r
 516 \r
 517             IotLogDebug( "(MQTT connection %p) PINGRESP in data stream.", pMqttConnection );\r
 518 \r
 519             /* Deserialize PINGRESP. */\r
 520             status = _getPingrespDeserializer( pMqttConnection->pSerializer )( pIncomingPacket );\r
 521 \r
 522             if( status == IOT_MQTT_SUCCESS )\r
 523             {\r
 524                 if( Atomic_CompareAndSwap_u32( &( pMqttConnection->pingreq.u.operation.periodic.ping.failure ),\r
 525                                                0,\r
 526                                                1 ) == 1 )\r
 527                 {\r
 528                     IotLogDebug( "(MQTT connection %p) PINGRESP successfully parsed.",\r
 529                                  pMqttConnection );\r
 530                 }\r
 531                 else\r
 532                 {\r
 533                     IotLogWarn( "(MQTT connection %p) Unexpected PINGRESP received.",\r
 534                                 pMqttConnection );\r
 535                 }\r
 536             }\r
 537             else\r
 538             {\r
 539                 EMPTY_ELSE_MARKER;\r
 540             }\r
 541 \r
 542             break;\r
 543     }\r
 544 \r
 545     if( status != IOT_MQTT_SUCCESS )\r
 546     {\r
 547         IotLogError( "(MQTT connection %p) Packet parser status %s.",\r
 548                      pMqttConnection,\r
 549                      IotMqtt_strerror( status ) );\r
 550     }\r
 551     else\r
 552     {\r
 553         EMPTY_ELSE_MARKER;\r
 554     }\r
 555 \r
 556     return status;\r
 557 }\r
 558 \r
 559 /*-----------------------------------------------------------*/\r
 560 \r
 561 static void _sendPuback( _mqttConnection_t * pMqttConnection,\r
 562                          uint16_t packetIdentifier )\r
 563 {\r
 564     IotMqttError_t status = IOT_MQTT_STATUS_PENDING;\r
 565     _mqttOperation_t * pPubackOperation = NULL;\r
 566 \r
 567     IotLogDebug( "(MQTT connection %p) Sending PUBACK for received PUBLISH %hu.",\r
 568                  pMqttConnection,\r
 569                  packetIdentifier );\r
 570 \r
 571     /* Create a PUBACK operation. */\r
 572     status = _IotMqtt_CreateOperation( pMqttConnection,\r
 573                                        0,\r
 574                                        NULL,\r
 575                                        &pPubackOperation );\r
 576 \r
 577     if( status != IOT_MQTT_SUCCESS )\r
 578     {\r
 579         IOT_GOTO_CLEANUP();\r
 580     }\r
 581 \r
 582     /* Set the operation type. */\r
 583     pPubackOperation->u.operation.type = IOT_MQTT_PUBACK;\r
 584 \r
 585     /* Generate a PUBACK packet from the packet identifier. */\r
 586     status = _getMqttPubackSerializer( pMqttConnection->pSerializer )( packetIdentifier,\r
 587                                                                        &( pPubackOperation->u.operation.pMqttPacket ),\r
 588                                                                        &( pPubackOperation->u.operation.packetSize ) );\r
 589 \r
 590     if( status != IOT_MQTT_SUCCESS )\r
 591     {\r
 592         IOT_GOTO_CLEANUP();\r
 593     }\r
 594 \r
 595     /* Add the PUBACK operation to the send queue for network transmission. */\r
 596     status = _IotMqtt_ScheduleOperation( pPubackOperation,\r
 597                                          _IotMqtt_ProcessSend,\r
 598                                          0 );\r
 599 \r
 600     if( status != IOT_MQTT_SUCCESS )\r
 601     {\r
 602         IotLogError( "(MQTT connection %p) Failed to enqueue PUBACK for sending.",\r
 603                      pMqttConnection );\r
 604 \r
 605         IOT_GOTO_CLEANUP();\r
 606     }\r
 607     else\r
 608     {\r
 609         EMPTY_ELSE_MARKER;\r
 610     }\r
 611 \r
 612     /* Clean up on error. */\r
 613     IOT_FUNCTION_CLEANUP_BEGIN();\r
 614 \r
 615     if( status != IOT_MQTT_SUCCESS )\r
 616     {\r
 617         if( pPubackOperation != NULL )\r
 618         {\r
 619             _IotMqtt_DestroyOperation( pPubackOperation );\r
 620         }\r
 621         else\r
 622         {\r
 623             EMPTY_ELSE_MARKER;\r
 624         }\r
 625     }\r
 626     else\r
 627     {\r
 628         EMPTY_ELSE_MARKER;\r
 629     }\r
 630 }\r
 631 \r
 632 /*-----------------------------------------------------------*/\r
 633 \r
 634 static void _flushPacket( void * pNetworkConnection,\r
 635                           const _mqttConnection_t * pMqttConnection,\r
 636                           size_t length )\r
 637 {\r
 638     size_t bytesFlushed = 0;\r
 639     uint8_t receivedByte = 0;\r
 640 \r
 641     for( bytesFlushed = 0; bytesFlushed < length; bytesFlushed++ )\r
 642     {\r
 643         ( void ) _IotMqtt_GetNextByte( pNetworkConnection,\r
 644                                        pMqttConnection->pNetworkInterface,\r
 645                                        &receivedByte );\r
 646     }\r
 647 }\r
 648 \r
 649 /*-----------------------------------------------------------*/\r
 650 \r
 651 bool _IotMqtt_GetNextByte( void * pNetworkConnection,\r
 652                            const IotNetworkInterface_t * pNetworkInterface,\r
 653                            uint8_t * pIncomingByte )\r
 654 {\r
 655     bool status = false;\r
 656     uint8_t incomingByte = 0;\r
 657     size_t bytesReceived = 0;\r
 658 \r
 659     /* Attempt to read 1 byte. */\r
 660     bytesReceived = pNetworkInterface->receive( pNetworkConnection,\r
 661                                                 &incomingByte,\r
 662                                                 1 );\r
 663 \r
 664     /* Set the output parameter and return success if 1 byte was read. */\r
 665     if( bytesReceived == 1 )\r
 666     {\r
 667         *pIncomingByte = incomingByte;\r
 668         status = true;\r
 669     }\r
 670     else\r
 671     {\r
 672         /* Network receive must return 0 on failure. */\r
 673         IotMqtt_Assert( bytesReceived == 0 );\r
 674     }\r
 675 \r
 676     return status;\r
 677 }\r
 678 \r
 679 /*-----------------------------------------------------------*/\r
 680 \r
 681 void _IotMqtt_CloseNetworkConnection( IotMqttDisconnectReason_t disconnectReason,\r
 682                                       _mqttConnection_t * pMqttConnection )\r
 683 {\r
 684     IotTaskPoolError_t taskPoolStatus = IOT_TASKPOOL_SUCCESS;\r
 685     IotNetworkError_t closeStatus = IOT_NETWORK_SUCCESS;\r
 686     IotMqttCallbackParam_t callbackParam = { .u.message = { 0 } };\r
 687     void * pNetworkConnection = NULL, * pDisconnectCallbackContext = NULL;\r
 688 \r
 689     /* Disconnect callback function. */\r
 690     void ( * disconnectCallback )( void *,\r
 691                                    IotMqttCallbackParam_t * ) = NULL;\r
 692 \r
 693     /* Network close function. */\r
 694     IotNetworkError_t ( * closeConnection) ( IotNetworkConnection_t ) = NULL;\r
 695 \r
 696     /* Mark the MQTT connection as disconnected and the keep-alive as failed. */\r
 697     IotMutex_Lock( &( pMqttConnection->referencesMutex ) );\r
 698     pMqttConnection->disconnected = true;\r
 699 \r
 700     if( pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs != 0 )\r
 701     {\r
 702         /* Keep-alive must have a PINGREQ allocated. */\r
 703         IotMqtt_Assert( pMqttConnection->pingreq.u.operation.pMqttPacket != NULL );\r
 704         IotMqtt_Assert( pMqttConnection->pingreq.u.operation.packetSize != 0 );\r
 705 \r
 706         /* PINGREQ provides a reference to the connection, so reference count must\r
 707          * be nonzero. */\r
 708         IotMqtt_Assert( pMqttConnection->references > 0 );\r
 709 \r
 710         /* Attempt to cancel the keep-alive job. */\r
 711         taskPoolStatus = IotTaskPool_TryCancel( IOT_SYSTEM_TASKPOOL,\r
 712                                                 pMqttConnection->pingreq.job,\r
 713                                                 NULL );\r
 714 \r
 715         /* Clean up keep-alive if its job was successfully canceled. Otherwise,\r
 716          * the executing keep-alive job will clean up itself. */\r
 717         if( taskPoolStatus == IOT_TASKPOOL_SUCCESS )\r
 718         {\r
 719             /* Free the packet */\r
 720             _getMqttFreePacketFunc( pMqttConnection->pSerializer )( pMqttConnection->pingreq.u.operation.pMqttPacket );\r
 721 \r
 722             /* Clear data about the keep-alive. */\r
 723             pMqttConnection->pingreq.u.operation.periodic.ping.keepAliveMs = 0;\r
 724             pMqttConnection->pingreq.u.operation.pMqttPacket = NULL;\r
 725             pMqttConnection->pingreq.u.operation.packetSize = 0;\r
 726 \r
 727             /* Keep-alive is cleaned up; decrement reference count. Since this\r
 728              * function must be followed with a call to DISCONNECT, a check to\r
 729              * destroy the connection is not done here. */\r
 730             pMqttConnection->references--;\r
 731 \r
 732             IotLogDebug( "(MQTT connection %p) Keep-alive job canceled and cleaned up.",\r
 733                          pMqttConnection );\r
 734         }\r
 735         else\r
 736         {\r
 737             EMPTY_ELSE_MARKER;\r
 738         }\r
 739     }\r
 740     else\r
 741     {\r
 742         EMPTY_ELSE_MARKER;\r
 743     }\r
 744 \r
 745     /* Copy the function pointers and contexts, as the MQTT connection may be\r
 746      * modified after the mutex is released. */\r
 747     disconnectCallback = pMqttConnection->disconnectCallback.function;\r
 748     pDisconnectCallbackContext = pMqttConnection->disconnectCallback.pCallbackContext;\r
 749 \r
 750     closeConnection = pMqttConnection->pNetworkInterface->close;\r
 751     pNetworkConnection = pMqttConnection->pNetworkConnection;\r
 752 \r
 753     IotMutex_Unlock( &( pMqttConnection->referencesMutex ) );\r
 754 \r
 755     /* Close the network connection. */\r
 756     if( closeConnection != NULL )\r
 757     {\r
 758         closeStatus = closeConnection( pNetworkConnection );\r
 759 \r
 760         if( closeStatus == IOT_NETWORK_SUCCESS )\r
 761         {\r
 762             IotLogInfo( "(MQTT connection %p) Network connection closed.", pMqttConnection );\r
 763         }\r
 764         else\r
 765         {\r
 766             IotLogWarn( "(MQTT connection %p) Failed to close network connection, error %d.",\r
 767                         pMqttConnection,\r
 768                         closeStatus );\r
 769         }\r
 770     }\r
 771     else\r
 772     {\r
 773         IotLogWarn( "(MQTT connection %p) No network close function was set. Network connection"\r
 774                     " not closed.", pMqttConnection );\r
 775     }\r
 776 \r
 777     /* Invoke the disconnect callback. */\r
 778     if( disconnectCallback != NULL )\r
 779     {\r
 780         /* Set the members of the callback parameter. */\r
 781         callbackParam.mqttConnection = pMqttConnection;\r
 782         callbackParam.u.disconnectReason = disconnectReason;\r
 783 \r
 784         disconnectCallback( pDisconnectCallbackContext,\r
 785                             &callbackParam );\r
 786     }\r
 787     else\r
 788     {\r
 789         EMPTY_ELSE_MARKER;\r
 790     }\r
 791 }\r
 792 \r
 793 /*-----------------------------------------------------------*/\r
 794 \r
 795 void IotMqtt_ReceiveCallback( IotNetworkConnection_t pNetworkConnection,\r
 796                               void * pReceiveContext )\r
 797 {\r
 798     IotMqttError_t status = IOT_MQTT_SUCCESS;\r
 799     _mqttPacket_t incomingPacket = { .u.pMqttConnection = NULL };\r
 800 \r
 801     /* Cast context to correct type. */\r
 802     _mqttConnection_t * pMqttConnection = ( _mqttConnection_t * ) pReceiveContext;\r
 803 \r
 804     /* Read an MQTT packet from the network. */\r
 805     status = _getIncomingPacket( pNetworkConnection,\r
 806                                  pMqttConnection,\r
 807                                  &incomingPacket );\r
 808 \r
 809     if( status == IOT_MQTT_SUCCESS )\r
 810     {\r
 811         /* Deserialize the received packet. */\r
 812         status = _deserializeIncomingPacket( pMqttConnection,\r
 813                                              &incomingPacket );\r
 814 \r
 815         /* Free any buffers allocated for the MQTT packet. */\r
 816         if( incomingPacket.pRemainingData != NULL )\r
 817         {\r
 818             IotMqtt_FreeMessage( incomingPacket.pRemainingData );\r
 819         }\r
 820         else\r
 821         {\r
 822             EMPTY_ELSE_MARKER;\r
 823         }\r
 824     }\r
 825     else\r
 826     {\r
 827         EMPTY_ELSE_MARKER;\r
 828     }\r
 829 \r
 830     /* Close the network connection on a bad response. */\r
 831     if( status == IOT_MQTT_BAD_RESPONSE )\r
 832     {\r
 833         IotLogError( "(MQTT connection %p) Error processing incoming data. Closing connection.",\r
 834                      pMqttConnection );\r
 835 \r
 836         _IotMqtt_CloseNetworkConnection( IOT_MQTT_BAD_PACKET_RECEIVED,\r
 837                                          pMqttConnection );\r
 838     }\r
 839     else\r
 840     {\r
 841         EMPTY_ELSE_MARKER;\r
 842     }\r
 843 }\r
 844 \r
 845 /*-----------------------------------------------------------*/\r
 846 \r
 847 IotMqttError_t IotMqtt_GetIncomingMQTTPacketTypeAndLength( IotMqttPacketInfo_t * pIncomingPacket,\r
 848                                                            IotMqttGetNextByte_t getNextByte,\r
 849                                                            void * pNetworkConnection )\r
 850 {\r
 851     IotMqttError_t status = IOT_MQTT_SUCCESS;\r
 852 \r
 853     /* Read the packet type, which is the first byte available. */\r
 854     if( getNextByte( pNetworkConnection, &( pIncomingPacket->type ) ) == IOT_MQTT_SUCCESS )\r
 855     {\r
 856         /* Check that the incoming packet type is valid. */\r
 857         if( _incomingPacketValid( pIncomingPacket->type ) == false )\r
 858         {\r
 859             IotLogError( "(MQTT connection) Unknown packet type %02x received.",\r
 860                          pIncomingPacket->type );\r
 861 \r
 862             status = IOT_MQTT_BAD_RESPONSE;\r
 863         }\r
 864         else\r
 865         {\r
 866             /* Read the remaining length. */\r
 867             pIncomingPacket->remainingLength = _IotMqtt_GetRemainingLength_Generic( pNetworkConnection,\r
 868                                                                                     getNextByte );\r
 869 \r
 870             if( pIncomingPacket->remainingLength == MQTT_REMAINING_LENGTH_INVALID )\r
 871             {\r
 872                 status = IOT_MQTT_BAD_RESPONSE;\r
 873             }\r
 874         }\r
 875     }\r
 876     else\r
 877     {\r
 878         status = IOT_MQTT_NETWORK_ERROR;\r
 879     }\r
 880 \r
 881     return status;\r
 882 }\r
 883 \r
 884 /*-----------------------------------------------------------*/\r