From 166741fd6d68fa7b2b8ebadc4cdb352761e5ee9c Mon Sep 17 00:00:00 2001 From: Euripedes Rocha Date: Mon, 6 Jun 2022 12:52:22 -0300 Subject: [PATCH] fix: MQTT tick not starting on connect and switch to esp_timer - Tick should be updated on connect - Dependency of system timer removed to avoid issues when updating via SNTP Closes https://github.com/espressif/esp-idf/issues/9492 --- lib/include/platform_esp32_idf.h | 4 +- lib/platform_esp32_idf.c | 15 ++------ mqtt_client.c | 63 ++++++++++++++++++++++---------- 3 files changed, 49 insertions(+), 33 deletions(-) diff --git a/lib/include/platform_esp32_idf.h b/lib/include/platform_esp32_idf.h index 673e856c..4569041e 100644 --- a/lib/include/platform_esp32_idf.h +++ b/lib/include/platform_esp32_idf.h @@ -9,12 +9,12 @@ #include "freertos/FreeRTOS.h" #include "freertos/event_groups.h" +#include #include char *platform_create_id_string(void); int platform_random(int max); -long long platform_tick_get_ms(void); -void ms_to_timeval(int timeout_ms, struct timeval *tv); +uint64_t platform_tick_get_ms(void); #define ESP_MEM_CHECK(TAG, a, action) if (!(a)) { \ ESP_LOGE(TAG,"%s(%d): %s", __FUNCTION__, __LINE__, "Memory exhausted"); \ diff --git a/lib/platform_esp32_idf.c b/lib/platform_esp32_idf.c index cf834509..a370e601 100644 --- a/lib/platform_esp32_idf.c +++ b/lib/platform_esp32_idf.c @@ -3,7 +3,7 @@ #ifdef ESP_PLATFORM #include "esp_log.h" #include "esp_system.h" -#include +#include "esp_timer.h" #include #include @@ -26,18 +26,9 @@ int platform_random(int max) return esp_random() % max; } -long long platform_tick_get_ms(void) +uint64_t platform_tick_get_ms(void) { - struct timeval te; - gettimeofday(&te, NULL); // get current time - long long milliseconds = te.tv_sec * 1000LL + te.tv_usec / 1000; // calculate milliseconds - // printf("milliseconds: %lld\n", milliseconds); - return milliseconds; + return esp_timer_get_time()/(int64_t)1000; } -void ms_to_timeval(int timeout_ms, struct timeval *tv) -{ - tv->tv_sec = timeout_ms / 1000; - tv->tv_usec = (timeout_ms - (tv->tv_sec * 1000)) * 1000; -} #endif diff --git a/mqtt_client.c b/mqtt_client.c index a589f626..ccc38b9f 100644 --- a/mqtt_client.c +++ b/mqtt_client.c @@ -601,6 +601,39 @@ static void esp_mqtt_destroy_config(esp_mqtt_client_handle_t client) client->config = NULL; } +static inline bool has_timed_out(uint64_t last_tick, uint64_t timeout) { + uint64_t next = last_tick + timeout; + return (int64_t)(next - platform_tick_get_ms()) <= 0; +} + +static esp_err_t process_keepalive(esp_mqtt_client_handle_t client) +{ + if (client->connect_info.keepalive > 0) { + const uint64_t keepalive_ms = client->connect_info.keepalive * 1000; + + if (client->wait_for_ping_resp == true ) { + if (has_timed_out(client->keepalive_tick, keepalive_ms)) { + ESP_LOGE(TAG, "No PING_RESP, disconnected"); + esp_mqtt_abort_connection(client); + client->wait_for_ping_resp = false; + return ESP_FAIL; + } + return ESP_OK; + } + + if (has_timed_out(client->keepalive_tick, keepalive_ms/2)) { + if (esp_mqtt_client_ping(client) == ESP_FAIL) { + ESP_LOGE(TAG, "Can't send ping, disconnected"); + esp_mqtt_abort_connection(client); + return ESP_FAIL; + } + client->wait_for_ping_resp = true; + return ESP_OK; + } + } + return ESP_OK; +} + static inline esp_err_t esp_mqtt_write(esp_mqtt_client_handle_t client) { int wlen = 0, widx = 0, len = client->mqtt_state.outbound_message->length; @@ -1332,6 +1365,12 @@ static esp_err_t mqtt_process_receive(esp_mqtt_client_handle_t client) case MQTT_MSG_TYPE_PINGRESP: ESP_LOGD(TAG, "MQTT_MSG_TYPE_PINGRESP"); client->wait_for_ping_resp = false; + /* It is the responsibility of the Client to ensure that the interval between Control Packets + * being sent does not exceed the Keep Alive value. In the absence of sending any other Control + * Packets, the Client MUST send a PINGREQ Packet [MQTT-3.1.2-23]. + * [MQTT-3.1.2-23] + */ + client->keepalive_tick = platform_tick_get_ms(); break; } @@ -1450,6 +1489,7 @@ static void esp_mqtt_task(void *pv) client->state = MQTT_STATE_CONNECTED; esp_mqtt_dispatch_event_with_msgid(client); client->refresh_connection_tick = platform_tick_get_ms(); + client->keepalive_tick = platform_tick_get_ms(); break; case MQTT_STATE_CONNECTED: @@ -1475,7 +1515,7 @@ static void esp_mqtt_task(void *pv) outbox_set_pending(client->outbox, client->mqtt_state.pending_msg_id, TRANSMITTED); } // resend other "transmitted" messages after 1s - } else if (platform_tick_get_ms() - last_retransmit > client->config->message_retransmit_timeout) { + } else if (has_timed_out(last_retransmit, client->config->message_retransmit_timeout)) { last_retransmit = platform_tick_get_ms(); item = outbox_dequeue(client->outbox, TRANSMITTED, &msg_tick); if (item && (last_retransmit - msg_tick > client->config->message_retransmit_timeout)) { @@ -1483,27 +1523,12 @@ static void esp_mqtt_task(void *pv) } } - if (client->connect_info.keepalive && // connect_info.keepalive=0 means that the keepslive is disabled - platform_tick_get_ms() - client->keepalive_tick > client->connect_info.keepalive * 1000 / 2) { - //No ping resp from last ping => Disconnected - if (client->wait_for_ping_resp) { - ESP_LOGE(TAG, "No PING_RESP, disconnected"); - esp_mqtt_abort_connection(client); - client->wait_for_ping_resp = false; - break; - } - if (esp_mqtt_client_ping(client) == ESP_FAIL) { - ESP_LOGE(TAG, "Can't send ping, disconnected"); - esp_mqtt_abort_connection(client); - break; - } else { - client->wait_for_ping_resp = true; - } - ESP_LOGD(TAG, "PING sent"); + if (process_keepalive(client) != ESP_OK) { + break; } if (client->config->refresh_connection_after_ms && - platform_tick_get_ms() - client->refresh_connection_tick > client->config->refresh_connection_after_ms) { + has_timed_out(client->refresh_connection_tick, client->config->refresh_connection_after_ms)) { ESP_LOGD(TAG, "Refreshing the connection..."); esp_mqtt_abort_connection(client); client->state = MQTT_STATE_INIT;