Merge branch 'bugfix/mqtt_fragmented_packed_msg_3.2' into 'release/v3.2'

MQTT support for fragmented and packed messages (Backport 3.2) See merge request idf/esp-idf!4593
espressif · Mar 25, 2019 · 286202c · 286202c
2 parents f49e2af + 9ebaf36
commit 286202c
Show file tree

Hide file tree

Showing 7 changed files with 34 additions and 2 deletions.
diff --git a/components/mqtt/esp-mqtt b/components/mqtt/esp-mqtt
diff --git a/components/tcp_transport/transport_ws.c b/components/tcp_transport/transport_ws.c
@@ -191,7 +191,10 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_
     if ((poll_read = esp_transport_poll_read(ws->parent, timeout_ms)) <= 0) {
         return poll_read;
     }
-    if ((rlen = esp_transport_read(ws->parent, buffer, len, timeout_ms)) <= 0) {
+
+    // Receive and process header first (based on header size)
+    int header = 2;
+    if ((rlen = esp_transport_read(ws->parent, buffer, header, timeout_ms)) <= 0) {
         ESP_LOGE(TAG, "Error read data");
         return rlen;
     }
@@ -203,11 +206,20 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_
     ESP_LOGD(TAG, "Opcode: %d, mask: %d, len: %d\r\n", opcode, mask, payload_len);
     if (payload_len == 126) {
         // headerLen += 2;
+        if ((rlen = esp_transport_read(ws->parent, data_ptr, header, timeout_ms)) <= 0) {
+            ESP_LOGE(TAG, "Error read data");
+            return rlen;
+        }
         payload_len = data_ptr[0] << 8 | data_ptr[1];
         payload_len_buff = len - 4;
         data_ptr += 2;
     } else if (payload_len == 127) {
         // headerLen += 8;
+        header = 8;
+        if ((rlen = esp_transport_read(ws->parent, data_ptr, header, timeout_ms)) <= 0) {
+            ESP_LOGE(TAG, "Error read data");
+            return rlen;
+        }
 
         if (data_ptr[0] != 0 || data_ptr[1] != 0 || data_ptr[2] != 0 || data_ptr[3] != 0) {
             // really too big!
@@ -218,6 +230,11 @@ static int ws_read(esp_transport_handle_t t, char *buffer, int len, int timeout_
         data_ptr += 8;
         payload_len_buff = len - 10;
     }
+    // Then receive and process payload
+    if ((rlen = esp_transport_read(ws->parent, data_ptr, payload_len, timeout_ms)) <= 0) {
+        ESP_LOGE(TAG, "Error read data");
+        return rlen;
+    }
     if (payload_len > payload_len_buff) {
         ESP_LOGD(TAG, "Actual data received (%d) are longer than mqtt buffer (%d)", payload_len, payload_len_buff);
         payload_len = payload_len_buff;

diff --git a/examples/protocols/mqtt/ssl/main/app_main.c b/examples/protocols/mqtt/ssl/main/app_main.c
@@ -116,6 +116,9 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
         case MQTT_EVENT_ERROR:
             ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
             break;
+        default:
+            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
+            break;
     }
     return ESP_OK;
 }

diff --git a/examples/protocols/mqtt/ssl_mutual_auth/main/app_main.c b/examples/protocols/mqtt/ssl_mutual_auth/main/app_main.c
@@ -114,6 +114,9 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
         case MQTT_EVENT_ERROR:
             ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
             break;
+        default:
+            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
+            break;
     }
     return ESP_OK;
 }

diff --git a/examples/protocols/mqtt/tcp/main/app_main.c b/examples/protocols/mqtt/tcp/main/app_main.c
@@ -69,6 +69,9 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
         case MQTT_EVENT_ERROR:
             ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
             break;
+        default:
+            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
+            break;
     }
     return ESP_OK;
 }

diff --git a/examples/protocols/mqtt/ws/main/app_main.c b/examples/protocols/mqtt/ws/main/app_main.c
@@ -66,6 +66,9 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
         case MQTT_EVENT_ERROR:
             ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
             break;
+        default:
+            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
+            break;
     }
     return ESP_OK;
 }

diff --git a/examples/protocols/mqtt/wss/main/app_main.c b/examples/protocols/mqtt/wss/main/app_main.c
@@ -116,6 +116,9 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
         case MQTT_EVENT_ERROR:
             ESP_LOGI(TAG, "MQTT_EVENT_ERROR");
             break;
+        default:
+            ESP_LOGI(TAG, "Other event id:%d", event->event_id);
+            break;
     }
     return ESP_OK;
 }
+3 −0		include/mqtt_client.h
+2 −1		lib/include/mqtt_msg.h
+11 −1		lib/include/mqtt_outbox.h
+46 −20		lib/mqtt_msg.c
+10 −7		lib/mqtt_outbox.c
+218 −81		mqtt_client.c