aws_iot_mqtt_client_common_internal.cpp

00001 /*
00002 * Copyright 2015-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
00003 *
00004 * Licensed under the Apache License, Version 2.0 (the "License").
00005 * You may not use this file except in compliance with the License.
00006 * A copy of the License is located at
00007 *
00008 * http://aws.amazon.com/apache2.0
00009 *
00010 * or in the "license" file accompanying this file. This file is distributed
00011 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
00012 * express or implied. See the License for the specific language governing
00013 * permissions and limitations under the License.
00014 */
00015 
00016 // Based on Eclipse Paho.
00017 /*******************************************************************************
00018  * Copyright (c) 2014 IBM Corp.
00019  *
00020  * All rights reserved. This program and the accompanying materials
00021  * are made available under the terms of the Eclipse Public License v1.0
00022  * and Eclipse Distribution License v1.0 which accompany this distribution.
00023  *
00024  * The Eclipse Public License is available at
00025  *    http://www.eclipse.org/legal/epl-v10.html
00026  * and the Eclipse Distribution License is available at
00027  *   http://www.eclipse.org/org/documents/edl-v10.php.
00028  *
00029  * Contributors:
00030  *    Ian Craggs - initial API and implementation and/or initial documentation
00031  *    Sergio R. Caprile - non-blocking packet read functions for stream transport
00032  *******************************************************************************/
00033 
00034 /**
00035  * @file aws_iot_mqtt_client_common_internal.c
00036  * @brief MQTT client internal API definitions
00037  */
00038 
00039 #ifdef __cplusplus
00040 extern "C" {
00041 #endif
00042 
00043 #include <aws_iot_mqtt_client.h>
00044 #include <unistd.h>
00045 #include "aws_iot_mqtt_client_common_internal.h"
00046 
00047 /* Max length of packet header */
00048 #define MAX_NO_OF_REMAINING_LENGTH_BYTES 4
00049 
00050 /**
00051  * Encodes the message length according to the MQTT algorithm
00052  * @param buf the buffer into which the encoded data is written
00053  * @param length the length to be encoded
00054  * @return the number of bytes written to buffer
00055  */
00056 size_t aws_iot_mqtt_internal_write_len_to_buffer(unsigned char *buf, uint32_t length) {
00057     size_t outLen = 0;
00058     unsigned char encodedByte;
00059 
00060     FUNC_ENTRY;
00061     do {
00062         encodedByte = (unsigned char) (length % 128);
00063         length /= 128;
00064         /* if there are more digits to encode, set the top bit of this digit */
00065         if(length > 0) {
00066             encodedByte |= 0x80;
00067         }
00068         buf[outLen++] = encodedByte;
00069     } while(length > 0);
00070 
00071     FUNC_EXIT_RC(outLen);
00072 }
00073 
00074 /**
00075  * Decodes the message length according to the MQTT algorithm
00076  * @param the buffer containing the message
00077  * @param value the decoded length returned
00078  * @return the number of bytes read from the socket
00079  */
00080 IoT_Error_t aws_iot_mqtt_internal_decode_remaining_length_from_buffer(unsigned char *buf, uint32_t *decodedLen,
00081                                                                       uint32_t *readBytesLen) {
00082     unsigned char encodedByte;
00083     uint32_t multiplier, len;
00084     FUNC_ENTRY;
00085 
00086     multiplier = 1;
00087     len = 0;
00088     *decodedLen = 0;
00089 
00090     do {
00091         if(++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) {
00092             /* bad data */
00093             FUNC_EXIT_RC(MQTT_DECODE_REMAINING_LENGTH_ERROR);
00094         }
00095         encodedByte = *buf;
00096         buf++;
00097         *decodedLen += (encodedByte & 127) * multiplier;
00098         multiplier *= 128;
00099     } while((encodedByte & 128) != 0);
00100 
00101     *readBytesLen = len;
00102 
00103     FUNC_EXIT_RC(IOT_SUCCESS);
00104 }
00105 
00106 uint32_t aws_iot_mqtt_internal_get_final_packet_length_from_remaining_length(uint32_t rem_len) {
00107     rem_len += 1; /* header byte */
00108     /* now remaining_length field (MQTT 3.1.1 - 2.2.3)*/
00109     if(rem_len < 128) {
00110         rem_len += 1;
00111     } else if(rem_len < 16384) {
00112         rem_len += 2;
00113     } else if(rem_len < 2097152) {
00114         rem_len += 3;
00115     } else {
00116         rem_len += 4;
00117     }
00118     return rem_len;
00119 }
00120 
00121 /**
00122  * Calculates uint16 packet id from two bytes read from the input buffer
00123  * Checks Endianness at runtime
00124  *
00125  * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned
00126  * @return the value calculated
00127  */
00128 uint16_t aws_iot_mqtt_internal_read_uint16_t(unsigned char **pptr) {
00129     unsigned char *ptr = *pptr;
00130     uint16_t len = 0;
00131     uint8_t firstByte = (uint8_t) (*ptr);
00132     uint8_t secondByte = (uint8_t) (*(ptr + 1));
00133     len = (uint16_t) (secondByte + (256 * firstByte));
00134 
00135     *pptr += 2;
00136     return len;
00137 }
00138 
00139 /**
00140  * Writes an integer as 2 bytes to an output buffer.
00141  * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
00142  * @param anInt the integer to write
00143  */
00144 void aws_iot_mqtt_internal_write_uint_16(unsigned char **pptr, uint16_t anInt) {
00145     **pptr = (unsigned char) (anInt / 256);
00146     (*pptr)++;
00147     **pptr = (unsigned char) (anInt % 256);
00148     (*pptr)++;
00149 }
00150 
00151 /**
00152  * Reads one character from the input buffer.
00153  * @param pptr pointer to the input buffer - incremented by the number of bytes used & returned
00154  * @return the character read
00155  */
00156 unsigned char aws_iot_mqtt_internal_read_char(unsigned char **pptr) {
00157     unsigned char c = **pptr;
00158     (*pptr)++;
00159     return c;
00160 }
00161 
00162 /**
00163  * Writes one character to an output buffer.
00164  * @param pptr pointer to the output buffer - incremented by the number of bytes used & returned
00165  * @param c the character to write
00166  */
00167 void aws_iot_mqtt_internal_write_char(unsigned char **pptr, unsigned char c) {
00168     **pptr = c;
00169     (*pptr)++;
00170 }
00171 
00172 void aws_iot_mqtt_internal_write_utf8_string(unsigned char **pptr, const char *string, uint16_t stringLen) {
00173     /* Nothing that calls this function will have a stringLen with a size larger than 2 bytes (MQTT 3.1.1 - 1.5.3) */
00174     aws_iot_mqtt_internal_write_uint_16(pptr, stringLen);
00175     if(stringLen > 0) {
00176         memcpy(*pptr, string, stringLen);
00177         *pptr += stringLen;
00178     }
00179 }
00180 
00181 /**
00182  * Initialize the MQTTHeader structure. Used to ensure that Header bits are
00183  * always initialized using the proper mappings. No Endianness issues here since
00184  * the individual fields are all less than a byte. Also generates no warnings since
00185  * all fields are initialized using hex constants
00186  */
00187 IoT_Error_t aws_iot_mqtt_internal_init_header(MQTTHeader *pHeader, MessageTypes message_type,
00188                                               QoS qos, uint8_t dup, uint8_t retained) {
00189     FUNC_ENTRY;
00190 
00191     if(NULL == pHeader) {
00192         FUNC_EXIT_RC(NULL_VALUE_ERROR);
00193     }
00194 
00195     /* Set all bits to zero */
00196     pHeader->byte = 0;
00197     switch(message_type) {
00198         case UNKNOWN:
00199             /* Should never happen */
00200             return IOT_FAILURE;
00201         case CONNECT:
00202             pHeader->bits.type = 0x01;
00203             break;
00204         case CONNACK:
00205             pHeader->bits.type = 0x02;
00206             break;
00207         case PUBLISH:
00208             pHeader->bits.type = 0x03;
00209             break;
00210         case PUBACK:
00211             pHeader->bits.type = 0x04;
00212             break;
00213         case PUBREC:
00214             pHeader->bits.type = 0x05;
00215             break;
00216         case PUBREL:
00217             pHeader->bits.type = 0x06;
00218             break;
00219         case PUBCOMP:
00220             pHeader->bits.type = 0x07;
00221             break;
00222         case SUBSCRIBE:
00223             pHeader->bits.type = 0x08;
00224             break;
00225         case SUBACK:
00226             pHeader->bits.type = 0x09;
00227             break;
00228         case UNSUBSCRIBE:
00229             pHeader->bits.type = 0x0A;
00230             break;
00231         case UNSUBACK:
00232             pHeader->bits.type = 0x0B;
00233             break;
00234         case PINGREQ:
00235             pHeader->bits.type = 0x0C;
00236             break;
00237         case PINGRESP:
00238             pHeader->bits.type = 0x0D;
00239             break;
00240         case DISCONNECT:
00241             pHeader->bits.type = 0x0E;
00242             break;
00243         default:
00244             /* Should never happen */
00245         FUNC_EXIT_RC(IOT_FAILURE);
00246     }
00247 
00248     pHeader->bits.dup = (1 == dup) ? 0x01 : 0x00;
00249     switch(qos) {
00250         case QOS0:
00251             pHeader->bits.qos = 0x00;
00252             break;
00253         case QOS1:
00254             pHeader->bits.qos = 0x01;
00255             break;
00256         default:
00257             /* Using QOS0 as default */
00258             pHeader->bits.qos = 0x00;
00259             break;
00260     }
00261 
00262     pHeader->bits.retain = (1 == retained) ? 0x01 : 0x00;
00263 
00264     FUNC_EXIT_RC(IOT_SUCCESS);
00265 }
00266 
00267 IoT_Error_t aws_iot_mqtt_internal_send_packet(AWS_IoT_Client *pClient, size_t length, TimerAWS *pTimer) {
00268 
00269     size_t sentLen, sent;
00270     IoT_Error_t rc;
00271 
00272     FUNC_ENTRY;
00273 
00274     if(NULL == pClient || NULL == pTimer) {
00275         FUNC_EXIT_RC(NULL_VALUE_ERROR);
00276     }
00277 
00278     if(length >= pClient->clientData.writeBufSize) {
00279         FUNC_EXIT_RC(MQTT_TX_BUFFER_TOO_SHORT_ERROR);
00280     }
00281 
00282 #ifdef _ENABLE_THREAD_SUPPORT_
00283     rc = aws_iot_mqtt_client_lock_mutex(pClient, &(pClient->clientData.tls_write_mutex));
00284     if(IOT_SUCCESS != rc) {
00285         FUNC_EXIT_RC(rc);
00286     }
00287 #endif
00288 
00289     sentLen = 0;
00290     sent = 0;
00291 
00292     while(sent < length && !has_timer_expired(pTimer)) {
00293         rc = pClient->networkStack.write(&(pClient->networkStack), &pClient->clientData.writeBuf[sent], length, pTimer,
00294                                          &sentLen);
00295         if(IOT_SUCCESS != rc) {
00296             /* there was an error writing the data */
00297             break;
00298         }
00299         sent += sentLen;
00300     }
00301 
00302 #ifdef _ENABLE_THREAD_SUPPORT_
00303     rc = aws_iot_mqtt_client_unlock_mutex(pClient, &(pClient->clientData.tls_write_mutex));
00304     if(IOT_SUCCESS != rc) {
00305         FUNC_EXIT_RC(rc);
00306     }
00307 #endif
00308 
00309     if(sent == length) {
00310         /* record the fact that we have successfully sent the packet */
00311         //countdown_sec(&c->pingTimer, c->clientData.keepAliveInterval);
00312         FUNC_EXIT_RC(IOT_SUCCESS);
00313     }
00314 
00315     FUNC_EXIT_RC(IOT_FAILURE);
00316 }
00317 
00318 static IoT_Error_t _aws_iot_mqtt_internal_decode_packet_remaining_len(AWS_IoT_Client *pClient,
00319                                                                       size_t *rem_len, TimerAWS *pTimer) {
00320     unsigned char encodedByte;
00321     size_t multiplier, len;
00322     IoT_Error_t rc;
00323 
00324     FUNC_ENTRY;
00325 
00326     multiplier = 1;
00327     len = 0;
00328     *rem_len = 0;
00329 
00330     do {
00331         if(++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) {
00332             /* bad data */
00333             FUNC_EXIT_RC(MQTT_DECODE_REMAINING_LENGTH_ERROR);
00334         }
00335 
00336         rc = pClient->networkStack.read(&(pClient->networkStack), &encodedByte, 1, pTimer, &len);
00337         if(IOT_SUCCESS != rc) {
00338             FUNC_EXIT_RC(rc);
00339         }
00340 
00341         *rem_len += ((encodedByte & 127) * multiplier);
00342         multiplier *= 128;
00343     } while((encodedByte & 128) != 0);
00344 
00345     FUNC_EXIT_RC(rc);
00346 }
00347 
00348 static IoT_Error_t _aws_iot_mqtt_internal_read_packet(AWS_IoT_Client *pClient, TimerAWS *pTimer, uint8_t *pPacketType) {
00349     size_t len, rem_len, total_bytes_read, bytes_to_be_read, read_len;
00350     IoT_Error_t rc;
00351     MQTTHeader header = {0};
00352     TimerAWS packetTimer;
00353     init_timer(&packetTimer);
00354     countdown_ms(&packetTimer, pClient->clientData.packetTimeoutMs);
00355 
00356     len = 0;
00357     rem_len = 0;
00358     total_bytes_read = 0;
00359     bytes_to_be_read = 0;
00360     read_len = 0;
00361 
00362     rc = pClient->networkStack.read(&(pClient->networkStack), pClient->clientData.readBuf, 1, pTimer, &read_len);
00363     /* 1. read the header byte.  This has the packet type in it */
00364     if(NETWORK_SSL_NOTHING_TO_READ == rc) {
00365         return MQTT_NOTHING_TO_READ;
00366     } else if(IOT_SUCCESS != rc) {
00367         return rc;
00368     }
00369 
00370     len = 1;
00371 
00372     /* Use the constant packet receive timeout, instead of the variable (remaining) pTimer time, to
00373      * determine packet receiving timeout. This is done so we don't prematurely time out packet receiving
00374      * if the remaining time in pTimer is too short.
00375      */
00376     pTimer = &packetTimer;
00377 
00378     /* 2. read the remaining length.  This is variable in itself */
00379     rc = _aws_iot_mqtt_internal_decode_packet_remaining_len(pClient, &rem_len, pTimer);
00380     if(IOT_SUCCESS != rc) {
00381         return rc;
00382     }
00383 
00384     /* if the buffer is too short then the message will be dropped silently */
00385     if(rem_len >= pClient->clientData.readBufSize) {
00386         bytes_to_be_read = pClient->clientData.readBufSize;
00387         do {
00388             rc = pClient->networkStack.read(&(pClient->networkStack), pClient->clientData.readBuf, bytes_to_be_read,
00389                                             pTimer, &read_len);
00390             if(IOT_SUCCESS == rc) {
00391                 total_bytes_read += read_len;
00392                 if((rem_len - total_bytes_read) >= pClient->clientData.readBufSize) {
00393                     bytes_to_be_read = pClient->clientData.readBufSize;
00394                 } else {
00395                     bytes_to_be_read = rem_len - total_bytes_read;
00396                 }
00397             }
00398         } while(total_bytes_read < rem_len && IOT_SUCCESS == rc);
00399         return MQTT_RX_BUFFER_TOO_SHORT_ERROR;
00400     }
00401 
00402     /* put the original remaining length into the read buffer */
00403     len += aws_iot_mqtt_internal_write_len_to_buffer(pClient->clientData.readBuf + 1, (uint32_t) rem_len);
00404 
00405     /* 3. read the rest of the buffer using a callback to supply the rest of the data */
00406     if(rem_len > 0) {
00407         rc = pClient->networkStack.read(&(pClient->networkStack), pClient->clientData.readBuf + len, rem_len, pTimer,
00408                                         &read_len);
00409         if(IOT_SUCCESS != rc || read_len != rem_len) {
00410             return IOT_FAILURE;
00411         }
00412     }
00413 
00414     header.byte = pClient->clientData.readBuf[0];
00415     *pPacketType = header.bits.type;
00416 
00417     FUNC_EXIT_RC(rc);
00418 }
00419 
00420 // assume topic filter and name is in correct format
00421 // # can only be at end
00422 // + and # can only be next to separator
00423 static char _aws_iot_mqtt_internal_is_topic_matched(char *pTopicFilter, char *pTopicName, uint16_t topicNameLen) {
00424 
00425     char *curf, *curn, *curn_end;
00426 
00427     if(NULL == pTopicFilter || NULL == pTopicName) {
00428         return NULL_VALUE_ERROR;
00429     }
00430 
00431     curf = pTopicFilter;
00432     curn = pTopicName;
00433     curn_end = curn + topicNameLen;
00434 
00435     while(*curf && (curn < curn_end)) {
00436         if(*curn == '/' && *curf != '/') {
00437             break;
00438         }
00439         if(*curf != '+' && *curf != '#' && *curf != *curn) {
00440             break;
00441         }
00442         if(*curf == '+') {
00443             /* skip until we meet the next separator, or end of string */
00444             char *nextpos = curn + 1;
00445             while(nextpos < curn_end && *nextpos != '/')
00446                 nextpos = ++curn + 1;
00447         } else if(*curf == '#') {
00448             /* skip until end of string */
00449             curn = curn_end - 1;
00450         }
00451 
00452         curf++;
00453         curn++;
00454     };
00455 
00456     return (curn == curn_end) && (*curf == '\0');
00457 }
00458 
00459 static IoT_Error_t _aws_iot_mqtt_internal_deliver_message(AWS_IoT_Client *pClient, char *pTopicName,
00460                                                           uint16_t topicNameLen,
00461                                                           IoT_Publish_Message_Params *pMessageParams) {
00462     uint32_t itr;
00463     IoT_Error_t rc;
00464     ClientState clientState;
00465 
00466     FUNC_ENTRY;
00467 
00468     if(NULL == pTopicName) {
00469         FUNC_EXIT_RC(NULL_VALUE_ERROR);
00470     }
00471 
00472     /* This function can be called from all MQTT APIs
00473      * But while callback return is in progress, Yield should not be called.
00474      * The state for CB_RETURN accomplishes that, as yield cannot be called while in that state */
00475     clientState = aws_iot_mqtt_get_client_state(pClient);
00476     rc = aws_iot_mqtt_set_client_state(pClient, clientState, CLIENT_STATE_CONNECTED_WAIT_FOR_CB_RETURN);
00477 
00478     /* Find the right message handler - indexed by topic */
00479     for(itr = 0; itr < AWS_IOT_MQTT_NUM_SUBSCRIBE_HANDLERS; ++itr) {
00480         if(NULL != pClient->clientData.messageHandlers[itr].topicName) {
00481             if(((topicNameLen == pClient->clientData.messageHandlers[itr].topicNameLen)
00482                 &&
00483                 (strncmp(pTopicName, (char *) pClient->clientData.messageHandlers[itr].topicName, topicNameLen) == 0))
00484                || _aws_iot_mqtt_internal_is_topic_matched((char *) pClient->clientData.messageHandlers[itr].topicName,
00485                                                           pTopicName, topicNameLen)) {
00486                 if(NULL != pClient->clientData.messageHandlers[itr].pApplicationHandler) {
00487                     pClient->clientData.messageHandlers[itr].pApplicationHandler(pClient, pTopicName, topicNameLen,
00488                                                                                  pMessageParams,
00489                                                                                  pClient->clientData.messageHandlers[itr].pApplicationHandlerData);
00490                 }
00491             }
00492         }
00493     }
00494     rc = aws_iot_mqtt_set_client_state(pClient, CLIENT_STATE_CONNECTED_WAIT_FOR_CB_RETURN, clientState);
00495 
00496     FUNC_EXIT_RC(rc);
00497 }
00498 
00499 static IoT_Error_t _aws_iot_mqtt_internal_handle_publish(AWS_IoT_Client *pClient, TimerAWS *pTimer) {
00500     char *topicName;
00501     uint16_t topicNameLen;
00502     uint32_t len;
00503     IoT_Error_t rc;
00504     IoT_Publish_Message_Params msg;
00505 
00506     FUNC_ENTRY;
00507 
00508     topicName = NULL;
00509     topicNameLen = 0;
00510     len = 0;
00511 
00512     rc = aws_iot_mqtt_internal_deserialize_publish(&msg.isDup, &msg.qos, &msg.isRetained,
00513                                                    &msg.id, &topicName, &topicNameLen,
00514                                                    (unsigned char **) &msg.payload, &msg.payloadLen,
00515                                                    pClient->clientData.readBuf,
00516                                                    pClient->clientData.readBufSize);
00517 
00518     if(IOT_SUCCESS != rc) {
00519         FUNC_EXIT_RC(rc);
00520     }
00521 
00522     rc = _aws_iot_mqtt_internal_deliver_message(pClient, topicName, topicNameLen, &msg);
00523     if(IOT_SUCCESS != rc) {
00524         FUNC_EXIT_RC(rc);
00525     }
00526 
00527     if(QOS0 == msg.qos) {
00528         /* No further processing required for QoS0 */
00529         FUNC_EXIT_RC(IOT_SUCCESS);
00530     }
00531 
00532     /* Message assumed to be QoS1 since we do not support QoS2 at this time */
00533     rc = aws_iot_mqtt_internal_serialize_ack(pClient->clientData.writeBuf, pClient->clientData.writeBufSize,
00534                                              PUBACK, 0, msg.id, &len);
00535 
00536     if(IOT_SUCCESS != rc) {
00537         FUNC_EXIT_RC(rc);
00538     }
00539 
00540     rc = aws_iot_mqtt_internal_send_packet(pClient, len, pTimer);
00541     if(IOT_SUCCESS != rc) {
00542         FUNC_EXIT_RC(rc);
00543     }
00544 
00545     FUNC_EXIT_RC(IOT_SUCCESS);
00546 }
00547 
00548 IoT_Error_t aws_iot_mqtt_internal_cycle_read(AWS_IoT_Client *pClient, TimerAWS *pTimer, uint8_t *pPacketType) {
00549     IoT_Error_t rc;
00550 
00551 #ifdef _ENABLE_THREAD_SUPPORT_
00552     IoT_Error_t threadRc;
00553 #endif
00554 
00555     if(NULL == pClient || NULL == pTimer) {
00556         return NULL_VALUE_ERROR;
00557     }
00558 
00559 #ifdef _ENABLE_THREAD_SUPPORT_
00560     threadRc = aws_iot_mqtt_client_lock_mutex(pClient, &(pClient->clientData.tls_read_mutex));
00561     if(IOT_SUCCESS != threadRc) {
00562         FUNC_EXIT_RC(threadRc);
00563     }
00564 #endif
00565 
00566     /* read the socket, see what work is due */
00567     rc = _aws_iot_mqtt_internal_read_packet(pClient, pTimer, pPacketType);
00568 
00569 #ifdef _ENABLE_THREAD_SUPPORT_
00570     threadRc = aws_iot_mqtt_client_unlock_mutex(pClient, &(pClient->clientData.tls_read_mutex));
00571     if(IOT_SUCCESS != threadRc && (MQTT_NOTHING_TO_READ == rc || IOT_SUCCESS == rc)) {
00572         return threadRc;
00573     }
00574 #endif
00575 
00576     if(MQTT_NOTHING_TO_READ == rc) {
00577         /* Nothing to read, not a cycle failure */
00578         return IOT_SUCCESS;
00579     } else if(IOT_SUCCESS != rc) {
00580         return rc;
00581     }
00582 
00583     switch(*pPacketType) {
00584         case CONNACK:
00585         case PUBACK:
00586         case SUBACK:
00587         case UNSUBACK:
00588             /* SDK is blocking, these responses will be forwarded to calling function to process */
00589             break;
00590         case PUBLISH: {
00591             rc = _aws_iot_mqtt_internal_handle_publish(pClient, pTimer);
00592             break;
00593         }
00594         case PUBREC:
00595         case PUBCOMP:
00596             /* QoS2 not supported at this time */
00597             break;
00598         case PINGRESP: {
00599             pClient->clientStatus.isPingOutstanding = 0;
00600             countdown_sec(&pClient->pingTimer, pClient->clientData.keepAliveInterval);
00601             break;
00602         }
00603         default: {
00604             /* Either unknown packet type or Failure occurred
00605              * Should not happen */
00606             rc = MQTT_RX_MESSAGE_PACKET_TYPE_INVALID_ERROR;
00607             break;
00608         }
00609     }
00610 
00611     return rc;
00612 }
00613 
00614 /* only used in single-threaded mode where one command at a time is in process */
00615 IoT_Error_t aws_iot_mqtt_internal_wait_for_read(AWS_IoT_Client *pClient, uint8_t packetType, TimerAWS *pTimer) {
00616     IoT_Error_t rc;
00617     uint8_t read_packet_type;
00618 
00619     FUNC_ENTRY;
00620     if(NULL == pClient || NULL == pTimer) {
00621         FUNC_EXIT_RC(NULL_VALUE_ERROR);
00622     }
00623 
00624     read_packet_type = 0;
00625     do {
00626         if(has_timer_expired(pTimer)) {
00627             /* we timed out */
00628             rc = MQTT_REQUEST_TIMEOUT_ERROR;
00629             break;
00630         }
00631         rc = aws_iot_mqtt_internal_cycle_read(pClient, pTimer, &read_packet_type);
00632     } while(NETWORK_DISCONNECTED_ERROR != rc && read_packet_type != packetType);
00633 
00634     if(MQTT_REQUEST_TIMEOUT_ERROR != rc && NETWORK_DISCONNECTED_ERROR != rc && read_packet_type != packetType) {
00635         FUNC_EXIT_RC(IOT_FAILURE);
00636     }
00637 
00638     /* Something failed or we didn't receive the expected packet, return error code */
00639     FUNC_EXIT_RC(rc);
00640 }
00641 
00642 /**
00643   * Serializes a 0-length packet into the supplied buffer, ready for writing to a socket
00644   * @param buf the buffer into which the packet will be serialized
00645   * @param buflen the length in bytes of the supplied buffer, to avoid overruns
00646   * @param packettype the message type
00647   * @param serialized length
00648   * @return IoT_Error_t indicating function execution status
00649   */
00650 IoT_Error_t aws_iot_mqtt_internal_serialize_zero(unsigned char *pTxBuf, size_t txBufLen, MessageTypes packetType,
00651                                                  size_t *pSerializedLength) {
00652     unsigned char *ptr;
00653     IoT_Error_t rc;
00654     MQTTHeader header = {0};
00655 
00656     FUNC_ENTRY;
00657     if(NULL == pTxBuf || NULL == pSerializedLength) {
00658         FUNC_EXIT_RC(NULL_VALUE_ERROR);
00659     }
00660 
00661     /* Buffer should have at least 2 bytes for the header */
00662     if(4 > txBufLen) {
00663         FUNC_EXIT_RC(MQTT_TX_BUFFER_TOO_SHORT_ERROR);
00664     }
00665 
00666     ptr = pTxBuf;
00667 
00668     rc = aws_iot_mqtt_internal_init_header(&header, packetType, QOS0, 0, 0);
00669     if(IOT_SUCCESS != rc) {
00670         FUNC_EXIT_RC(rc);
00671     }
00672 
00673     /* write header */
00674     aws_iot_mqtt_internal_write_char(&ptr, header.byte);
00675 
00676     /* write remaining length */
00677     ptr += aws_iot_mqtt_internal_write_len_to_buffer(ptr, 0);
00678     *pSerializedLength = (uint32_t) (ptr - pTxBuf);
00679 
00680     FUNC_EXIT_RC(IOT_SUCCESS);
00681 }
00682 
00683 #ifdef __cplusplus
00684 }
00685 #endif