Changeset 399

Timestamp:

Jun 5, 2019, 2:23:55 PM (5 years ago)

Author:

coas-nagasima

Message:

更新

Location:

azure_iot_hub/trunk

Files:

• app_iothub_client/src/client.c (modified) (4 diffs)
• azure_iothub/umqtt/inc/azure_umqtt_c/mqtt_codec.h (modified) (1 diff)
• azure_iothub/umqtt/src/mqtt_client.c (modified) (7 diffs)
• azure_iothub/umqtt/src/mqtt_codec.c (modified) (9 diffs)
• ntshell/src/fdtable.c (modified) (2 diffs)
• ntshell/src/socket_stub.c (modified) (7 diffs)

azure_iot_hub/trunk/app_iothub_client/src/client.c

@@ -218 +218 @@
                                 printf("failure to set option \"MinimumPollingTime\"\r\n");
                         }
+
+                        bool traceOn = 1;
+                        if (IoTHubClient_LL_SetOption(iotHubClientHandle, OPTION_LOG_TRACE, &traceOn) != IOTHUB_CLIENT_OK)
+                        {
+                                printf("failure to set option \"log trace on\"\r\n");
+                        }
 #endif
 #ifdef SET_TRUSTED_CERT_IN_SAMPLES
@@ -241 +247 @@
                                 do
                                 {
-                                        //(void)printf("iterator: [%d], callbackCounter: [%d]. \r\n", iterator, callbackCounter);
-
-                                        if (iterator < MESSAGE_COUNT && (iterator <= callbackCounter))
+                                        if ((iterator < MESSAGE_COUNT) && (iterator <= callbackCounter))
                                         {
                                                 temperature = minTemperature + (rand() % 10);
@@ -265 +269 @@
                                                         }
 
-                                                        (void)IoTHubMessage_SetContentTypeSystemProperty(messages[iterator].messageHandle, "application/json");
-                                                        (void)IoTHubMessage_SetContentEncodingSystemProperty(messages[iterator].messageHandle, "utf-8");
+                                                        if (proto == 0) {
+                                                                (void)IoTHubMessage_SetContentTypeSystemProperty(messages[iterator].messageHandle, "application/json");
+                                                                (void)IoTHubMessage_SetContentEncodingSystemProperty(messages[iterator].messageHandle, "utf-8");
+                                                        }
 
                                                         if (IoTHubClient_LL_SendEventAsync(iotHubClientHandle, messages[iterator].messageHandle, SendConfirmationCallback, &messages[iterator]) != IOTHUB_CLIENT_OK)
@@ -277 +283 @@
                                                         }
                                                 }
+                                                iterator++;
                                         }
 
                                         IoTHubClient_LL_DoWork(iotHubClientHandle);
                                         ThreadAPI_Sleep(1);
-
-                                        iterator++;
 
                                         if (callbackCounter >= MESSAGE_COUNT)

azure_iot_hub/trunk/azure_iothub/umqtt/inc/azure_umqtt_c/mqtt_codec.h

@@ -28 +28 @@
 MOCKABLE_FUNCTION(, void, mqtt_codec_destroy, MQTTCODEC_HANDLE, handle);
 
+MOCKABLE_FUNCTION(, void, mqtt_codec_reset, MQTTCODEC_HANDLE, handle);
 MOCKABLE_FUNCTION(, BUFFER_HANDLE, mqtt_codec_connect, const MQTT_CLIENT_OPTIONS*, mqttOptions, STRING_HANDLE, trace_log);
 MOCKABLE_FUNCTION(, BUFFER_HANDLE, mqtt_codec_disconnect);

azure_iot_hub/trunk/azure_iothub/umqtt/src/mqtt_client.c

@@ -113 +113 @@
 static STRING_HANDLE construct_trace_log_handle(MQTT_CLIENT* mqtt_client)
 {
+#ifndef NO_LOGGING
     STRING_HANDLE trace_log;
     if (mqtt_client->logTrace)
@@ -123 +124 @@
     }
     return trace_log;
+#else
+    return NULL;
+#endif
 }
 
@@ -445 +449 @@
         if (mqtt_codec_bytesReceived(mqtt_client->codec_handle, buffer, size) != 0)
         {
+            mqtt_codec_reset(mqtt_client->codec_handle);
             set_error_callback(mqtt_client, MQTT_CLIENT_PARSE_ERROR);
         }
@@ -1215 +1220 @@
                 else
                 {
+#ifndef NO_LOGGING
                     if (mqtt_client->logTrace)
                     {
@@ -1221 +1227 @@
                         STRING_delete(trace_log);
                     }
+#endif
                     result = 0;
                 }
@@ -1280 +1287 @@
                         BUFFER_delete(pingPacket);
                         (void)tickcounter_get_current_ms(mqtt_client->packetTickCntr, &mqtt_client->timeSincePing);
-
+#ifndef NO_LOGGING
                         if (mqtt_client->logTrace)
                         {
@@ -1287 +1294 @@
                             STRING_delete(trace_log);
                         }
+#endif
                     }
                 }

azure_iot_hub/trunk/azure_iothub/umqtt/src/mqtt_codec.c

@@ -38 +38 @@
 #define UNSUBSCRIBE_FIXED_HEADER_FLAG       0x2
 
-#define MAX_SEND_SIZE                       0xFFFFFF7F
+#define MAX_SEND_SIZE                       0xFFFFFF7F // 268435455
+
+// This captures the maximum packet size for 3 digits.
+// If it's above this value then we bail out of the loop
+#define MAX_3_DIGIT_PACKET_SIZE             2097152
 
 #define CODEC_STATE_VALUES      \
@@ -409 +413 @@
         size_t willTopicLen = 0;
         size_t spaceLen = 0;
+        size_t currLen = 0;
+        size_t totalLen = 0;
 
         if (mqttOptions->clientId != NULL)
@@ -436 +442 @@
         }
 
-        size_t currLen = BUFFER_length(ctrlPacket);
-        size_t totalLen = clientLen + usernameLen + passwordLen + willMessageLen + willTopicLen + spaceLen;
+        currLen = BUFFER_length(ctrlPacket);
+        totalLen = clientLen + usernameLen + passwordLen + willMessageLen + willTopicLen + spaceLen;
 
         // Validate the Username & Password
@@ -507 +513 @@
                     {
                         (void)STRING_sprintf(connect_payload_trace, " | PWD: XXXX");
+                        //(void)STRING_sprintf(connect_payload_trace, " | PWD: %s", mqttOptions->password);
                     }
                 }
@@ -554 +561 @@
                 multiplier *= NEXT_128_CHUNK;
 
-                if (multiplier > 128 * 128 * 128)
+                if (multiplier > MAX_3_DIGIT_PACKET_SIZE)
                 {
                     result = MU_FAILURE;
@@ -561 +568 @@
             } while ((encodeByte & NEXT_128_CHUNK) != 0);
 
-            codecData->codecState = CODEC_STATE_VAR_HEADER;
-
-            // Reset remainLen Index
-            codecData->remainLenIndex = 0;
-            memset(codecData->storeRemainLen, 0, 4 * sizeof(uint8_t));
-
-            if (totalLen > 0)
-            {
-                codecData->bufferOffset = 0;
-                codecData->headerData = BUFFER_new();
-                if (codecData->headerData == NULL)
-                {
-                    /* Codes_SRS_MQTT_CODEC_07_035: [ If any error is encountered then the packet state will be marked as error and mqtt_codec_bytesReceived shall return a non-zero value. ] */
-                    LogError("Failed BUFFER_new");
-                    result = MU_FAILURE;
-                }
-                else
-                {
-                    if (BUFFER_pre_build(codecData->headerData, totalLen) != 0)
+            if (result != 0 || totalLen > MAX_SEND_SIZE)
+            {
+                LogError("Receive buffer too large for MQTT packet");
+                result = MU_FAILURE;
+            }
+            else
+            {
+                codecData->codecState = CODEC_STATE_VAR_HEADER;
+
+                // Reset remainLen Index
+                codecData->remainLenIndex = 0;
+                memset(codecData->storeRemainLen, 0, 4 * sizeof(uint8_t));
+
+                if (totalLen > 0)
+                {
+                    codecData->bufferOffset = 0;
+                    codecData->headerData = BUFFER_new();
+                    if (codecData->headerData == NULL)
                     {
                         /* Codes_SRS_MQTT_CODEC_07_035: [ If any error is encountered then the packet state will be marked as error and mqtt_codec_bytesReceived shall return a non-zero value. ] */
-                        LogError("Failed BUFFER_pre_build");
+                        LogError("Failed BUFFER_new");
                         result = MU_FAILURE;
                     }
-
+                    else
+                    {
+                        if (BUFFER_pre_build(codecData->headerData, totalLen) != 0)
+                        {
+                            /* Codes_SRS_MQTT_CODEC_07_035: [ If any error is encountered then the packet state will be marked as error and mqtt_codec_bytesReceived shall return a non-zero value. ] */
+                            LogError("Failed BUFFER_pre_build");
+                            result = MU_FAILURE;
+                        }
+
+                    }
                 }
             }
@@ -611 +626 @@
 }
 
+static void clear_codec_data(MQTTCODEC_INSTANCE* codec_data)
+{
+    // Clear the code instance data
+    codec_data->currPacket = UNKNOWN_TYPE;
+    codec_data->codecState = CODEC_STATE_FIXED_HEADER;
+    codec_data->headerFlags = 0;
+    codec_data->bufferOffset = 0;
+    codec_data->headerData = NULL;
+    memset(codec_data->storeRemainLen, 0, 4 * sizeof(uint8_t));
+    codec_data->remainLenIndex = 0;
+}
+
+void mqtt_codec_reset(MQTTCODEC_HANDLE handle)
+{
+    if (handle != NULL)
+    {
+        clear_codec_data(handle);
+    }
+}
+
 MQTTCODEC_HANDLE mqtt_codec_create(ON_PACKET_COMPLETE_CALLBACK packetComplete, void* callbackCtx)
 {
@@ -619 +654 @@
     {
         /* Codes_SRS_MQTT_CODEC_07_002: [On success mqtt_codec_create shall return a MQTTCODEC_HANDLE value.] */
-        result->currPacket = UNKNOWN_TYPE;
-        result->codecState = CODEC_STATE_FIXED_HEADER;
-        result->headerFlags = 0;
-        result->bufferOffset = 0;
+        clear_codec_data(result);
         result->packetComplete = packetComplete;
         result->callContext = callbackCtx;
-        result->headerData = NULL;
-        memset(result->storeRemainLen, 0, 4 * sizeof(uint8_t));
-        result->remainLenIndex = 0;
     }
     return result;
@@ -1086 +1115 @@
                         result = MU_FAILURE;
                     }
-                    if (codec_Data->currPacket == PINGRESP_TYPE)
+                    else if (codec_Data->currPacket == PINGRESP_TYPE)
                     {
                         /* Codes_SRS_MQTT_CODEC_07_034: [Upon a constructing a complete MQTT packet mqtt_codec_bytesReceived shall call the ON_PACKET_COMPLETE_CALLBACK function.] */

azure_iot_hub/trunk/ntshell/src/fdtable.c

@@ -299 +299 @@
 ER shell_get_evts(struct fd_events *evts, TMO tmout)
 {
-        int count = 0;
+        int count;
         SYSTIM prev, now;
         FLGPTN flgptn;
@@ -315 +315 @@
                 waitptn = *((FLGPTN *)&evts->readfds) | *((FLGPTN *)&evts->errorfds);
 #endif
+                count = 0;
                 for (int fd = 0; fd < fd_table_count; fd++) {
                         fp = &fd_table[fd];

azure_iot_hub/trunk/ntshell/src/socket_stub.c

@@ -70 +70 @@
 
 #define SOCKET_TIMEOUT 2000000
+#define TCP_SOCKET_BUF_SIZE 512
 
 struct addrinfo {
@@ -363 +364 @@
                                 return -ENOMEM;
 
-                        socket->buf_size = 512 + 512;
+                        socket->buf_size = TCP_SOCKET_BUF_SIZE + TCP_SOCKET_BUF_SIZE;
                         socket->buf = malloc(socket->buf_size);
 #ifdef _DEBUG
                         memset(socket->buf, 0, socket->buf_size);
 #endif
-                        T_TCP_CCEP ccep = { 0, socket->buf, 512, &socket->buf[512], 512, (FP)socket_tcp_callback };
+                        T_TCP_CCEP ccep = { 0, socket->buf, TCP_SOCKET_BUF_SIZE, &socket->buf[TCP_SOCKET_BUF_SIZE], TCP_SOCKET_BUF_SIZE, (FP)socket_tcp_callback };
                         ret = tcp_cre_cep(cepid, &ccep);
                         if (ret != E_OK) {
@@ -410 +411 @@
                                 return -ENOMEM;
 
-                        socket->buf_size = 512 + 512;
+                        socket->buf_size = TCP_SOCKET_BUF_SIZE + TCP_SOCKET_BUF_SIZE;
                         socket->buf = malloc(socket->buf_size);
 #ifdef _DEBUG
                         memset(socket->buf, 0, socket->buf_size);
 #endif
-                        T_TCP6_CCEP ccep = { 0, socket->buf, 512, &socket->buf[512], 512, (FP)socket_tcp6_callback };
+                        T_TCP6_CCEP ccep = { 0, socket->buf, TCP_SOCKET_BUF_SIZE, &socket->buf[TCP_SOCKET_BUF_SIZE], TCP_SOCKET_BUF_SIZE, (FP)socket_tcp6_callback };
                         ret = tcp6_cre_cep(cepid, &ccep);
                         if (ret != E_OK) {
@@ -522 +523 @@
                                 return -ENOMEM;
 
-                        socket->buf_size = 512 + 512;
+                        socket->buf_size = TCP_SOCKET_BUF_SIZE + TCP_SOCKET_BUF_SIZE;
                         socket->buf = malloc(socket->buf_size);
 #ifdef _DEBUG
                         memset(socket->buf, 0, socket->buf_size);
 #endif
-                        T_TCP_CCEP ccep = { 0, socket->buf, 512, &socket->buf[512], 512, (FP)socket_tcp_callback };
+                        T_TCP_CCEP ccep = { 0, socket->buf, TCP_SOCKET_BUF_SIZE, &socket->buf[TCP_SOCKET_BUF_SIZE], TCP_SOCKET_BUF_SIZE, (FP)socket_tcp_callback };
                         ret = tcp_cre_cep(cepid, &ccep);
                         if (ret != E_OK) {
@@ -560 +561 @@
                                 return -ENOMEM;
 
-                        socket->buf_size = 512 + 512;
+                        socket->buf_size = TCP_SOCKET_BUF_SIZE + TCP_SOCKET_BUF_SIZE;
                         socket->buf = malloc(socket->buf_size);
 #ifdef _DEBUG
                         memset(socket->buf, 0, socket->buf_size);
 #endif
-                        T_TCP6_CCEP ccep = { 0, socket->buf, 512, &socket->buf[512], 512, (FP)socket_tcp6_callback };
+                        T_TCP6_CCEP ccep = { 0, socket->buf, TCP_SOCKET_BUF_SIZE, &socket->buf[TCP_SOCKET_BUF_SIZE], TCP_SOCKET_BUF_SIZE, (FP)socket_tcp6_callback };
                         ret = tcp6_cre_cep(cepid, &ccep);
                         if (ret != E_OK) {
@@ -629 +630 @@
                                 return -ENOMEM;
 
-                        socket->buf_size = 512 + 512;
+                        socket->buf_size = TCP_SOCKET_BUF_SIZE + TCP_SOCKET_BUF_SIZE;
                         socket->buf = malloc(socket->buf_size);
 #ifdef _DEBUG
                         memset(socket->buf, 0, socket->buf_size);
 #endif
-                        T_TCP_CCEP ccep = { 0, socket->buf, 512, &socket->buf[512], 512, (FP)socket_tcp_callback };
+                        T_TCP_CCEP ccep = { 0, socket->buf, TCP_SOCKET_BUF_SIZE, &socket->buf[TCP_SOCKET_BUF_SIZE], TCP_SOCKET_BUF_SIZE, (FP)socket_tcp_callback };
                         ret = tcp_cre_cep(cepid, &ccep);
                         if (ret != E_OK) {
@@ -684 +685 @@
                                 return -ENOMEM;
 
-                        socket->buf_size = 512 + 512;
+                        socket->buf_size = TCP_SOCKET_BUF_SIZE + TCP_SOCKET_BUF_SIZE;
                         socket->buf = malloc(socket->buf_size);
 #ifdef _DEBUG
                         memset(socket->buf, 0, socket->buf_size);
 #endif
-                        T_TCP6_CCEP ccep = { 0, socket->buf, 512, &socket->buf[512], 512, (FP)socket_tcp6_callback };
+                        T_TCP6_CCEP ccep = { 0, socket->buf, TCP_SOCKET_BUF_SIZE, &socket->buf[TCP_SOCKET_BUF_SIZE], TCP_SOCKET_BUF_SIZE, (FP)socket_tcp6_callback };
                         ret = tcp6_cre_cep(cepid, &ccep);
                         if (ret != E_OK) {