Fix gui semaphore and dynamic mqtt msg queue #174

src/hasp_gui.cpp

@@ -230,15 +230,6 @@ static inline void gui_init_filesystems()
 
 void guiSetup()
 {
-#if ESP32
-    g_lvgl_task_handle = xTaskGetCurrentTaskHandle();
-
-    xGuiSemaphore = xSemaphoreCreateMutex();
-    if(!xGuiSemaphore) {
-        LOG_FATAL(TAG_GUI, "Create mutex for LVGL failed");
-    }
-#endif
-
     // Initialize hardware drivers
     gui_init_tft();
     haspDevice.show_info(); // debug info + preload app flash size
@@ -366,19 +357,23 @@ void guiSetup()
     lv_obj_set_style_local_bg_color(lv_layer_sys(), LV_OBJ_PART_MAIN, LV_STATE_DEFAULT, LV_COLOR_BLACK);
     lv_obj_set_style_local_bg_opa(lv_layer_sys(), LV_OBJ_PART_MAIN, LV_STATE_DEFAULT, LV_OPA_0);
 
+#if defined(ESP32) && defined(HASP_USE_ESP_MQTT)
+    xGuiSemaphore = xSemaphoreCreateMutex();
+    if(!xGuiSemaphore) {
+        LOG_FATAL(TAG_GUI, "Create mutex for LVGL failed");
+    }
+    gui_acquire(); // Block LVGL until plate is fully booted
+#if HASP_USE_LVGL_TASK
+    gui_setup_lvgl_task();
+#endif // HASP_USE_LVGL_TASK
+#endif // ESP32 && HASP_USE_ESP_MQTT
+
     LOG_INFO(TAG_LVGL, F(D_SERVICE_STARTED));
 }
 
 IRAM_ATTR void guiLoop(void)
 {
-#if ESP32
-    if(pdTRUE == xSemaphoreTake(xGuiSemaphore, portMAX_DELAY)) {
-        lv_task_handler();
-        xSemaphoreGive(xGuiSemaphore);
-    }
-#else
     lv_task_handler(); // process animations
-#endif
 
 #if defined(STM32F4xx)
     //  tick.update();
@@ -394,22 +389,64 @@ void guiEverySecond(void)
     // nothing
 }
 
-void gui_acquire(void)
+#if defined(ESP32) && defined(HASP_USE_ESP_MQTT)
+
+#if HASP_USE_LVGL_TASK == 1
+static void gui_task(void* args)
 {
+    LOG_TRACE(TAG_GUI, "Start to run LVGL");
+    while(1) {
+        vTaskDelay(pdMS_TO_TICKS(10));
+
+        /* Try to take the semaphore, call lvgl related function on success */
+        if(pdTRUE == xSemaphoreTake(xGuiSemaphore, pdMS_TO_TICKS(10))) {
+            lv_task_handler();
+            xSemaphoreGive(xGuiSemaphore);
+        }
+    }
+}
+
+esp_err_t gui_setup_lvgl_task()
+{
+#if CONFIG_FREERTOS_UNICORE == 0
+    int err = xTaskCreatePinnedToCore(gui_task, "lvglTask", 1024 * 8, NULL, 5, &g_lvgl_task_handle, 1);
+#else
+    int err = xTaskCreatePinnedToCore(gui_task, "lvglTask", 1024 * 8, NULL, 5, &g_lvgl_task_handle, 0);
+#endif
+    if(!err) {
+        LOG_FATAL(TAG_GUI, "Create task for LVGL failed");
+        return ESP_FAIL;
+    }
+    return ESP_OK;
+}
+#endif // HASP_USE_LVGL_TASK
+
+bool gui_acquire(void)
+{
+#if ESP32
     TaskHandle_t task = xTaskGetCurrentTaskHandle();
     if(g_lvgl_task_handle != task) {
-        xSemaphoreTake(xGuiSemaphore, portMAX_DELAY);
+        if(xSemaphoreTake(xGuiSemaphore, pdMS_TO_TICKS(30)) != pdTRUE) {
+            return false;
+        }
     }
+#endif
+    return true;
 }
 
 void gui_release(void)
 {
+#if ESP32
     TaskHandle_t task = xTaskGetCurrentTaskHandle();
     if(g_lvgl_task_handle != task) {
         xSemaphoreGive(xGuiSemaphore);
+        // LOG_VERBOSE(TAG_TFT, F("GIVE"));
     }
+#endif
 }
 
+#endif // ESP32 && HASP_USE_ESP_MQTT
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 #if HASP_USE_CONFIG > 0
 bool guiGetConfig(const JsonObject& settings)

src/hasp_gui.h

@@ -61,8 +61,9 @@ void gui_flush_cb(lv_disp_drv_t* disp, const lv_area_t* area, lv_color_t* color_
 void gui_antiburn_cb(lv_disp_drv_t* disp, const lv_area_t* area, lv_color_t* color_p);
 
 /* ===== Locks ===== */
-void gui_acquire(void);
+bool gui_acquire(void);
 void gui_release(void);
+esp_err_t gui_setup_lvgl_task();
 
 /* ===== Read/Write Configuration ===== */
 #if HASP_USE_CONFIG > 0

src/main_arduino.cpp

@@ -3,6 +3,16 @@
 
 #if !(defined(WINDOWS) || defined(POSIX))
 
+/*
+#ifdef CORE_DEBUG_LEVEL
+#undef CORE_DEBUG_LEVEL
+#endif
+#define CORE_DEBUG_LEVEL 3
+#define LOG_LOCAL_LEVEL ESP_LOG_DEBUG
+
+#include "esp_log.h"
+*/
+
 #include "hasplib.h"
 #include "hasp_oobe.h"
 #include "sys/net/hasp_network.h"
@@ -31,6 +41,11 @@ void setup()
 {
     //   hal_setup();
 
+    esp_log_level_set("*", ESP_LOG_NONE); // set all components to ERROR level
+    // esp_log_level_set("wifi", ESP_LOG_NONE);              // enable WARN logs from WiFi stack
+    // esp_log_level_set("dhcpc", ESP_LOG_INFO);             // enable INFO logs from DHCP client
+    // esp_log_level_set("esp_crt_bundle", ESP_LOG_VERBOSE); // enable WARN logs from WiFi stack
+    // esp_log_level_set("esp_tls", ESP_LOG_VERBOSE);        // enable WARN logs from WiFi stack
     haspDevice.init();
 
     /****************************
@@ -75,16 +90,14 @@ void setup()
      * Apply User Configuration
      ***************************/
 
-#if HASP_USE_MQTT > 0
+    // #if HASP_USE_MQTT > 0
-    mqttSetup(); // Load Hostname before starting WiFi
+    //     mqttSetup(); // Load Hostname before starting WiFi
-#endif
+    // #endif
 
 #if HASP_USE_GPIO > 0
     gpioSetup();
 #endif
 
-
-
 #if HASP_USE_MDNS > 0
     mdnsSetup();
 #endif
@@ -120,21 +133,38 @@ void setup()
     // guiStart();
 
     delay(20);
+
     if(!oobe) {
-        dispatch_exec(NULL, "L:/boot.cmd", TAG_HASP);
+        dispatch_run_script(NULL, "L:/boot.cmd", TAG_HASP);
 #if HASP_USE_WIFI > 0 || HASP_USE_ETHERNET > 0
         network_run_scripts();
 #endif
     }
+
+#if HASP_USE_MQTT > 0
+    mqttSetup();
+#endif
+
     mainLastLoopTime = -1000; // reset loop counter
+    gui_release();
 }
 
 IRAM_ATTR void loop()
 {
+#if defined(ESP32) && defined(HASP_USE_ESP_MQTT)
+    if(!gui_acquire()) {
+        // LOG_ERROR(TAG_MAIN, F("TAKE Mutex"));
+        delay(10); // ms
+        return;
+    }
+#endif
+
+#if HASP_USE_LVGL_TASK == 0
     guiLoop();
+#endif
 
 #if HASP_USE_WIFI > 0 || HASP_USE_ETHERNET > 0
-   networkLoop();
+    networkLoop();
 #endif
 
 #if HASP_USE_GPIO > 0
@@ -222,11 +252,19 @@ IRAM_ATTR void loop()
         }
     }
 
+#if defined(ESP32) && defined(HASP_USE_ESP_MQTT)
+    gui_release();
+#endif
+
 // allow the cpu to switch to other tasks
+#if HASP_USE_LVGL_TASK == 0
 #ifdef ARDUINO_ARCH_ESP8266
     delay(2); // ms
 #else
-    delay(3); // ms
+    delay(5); // ms
+#endif
+#else // HASP_USE_LVGL_TASK != 0
+    delay(10); // ms
 #endif
 }
 

src/mqtt/hasp_mqtt_esp.cpp

@@ -19,14 +19,14 @@
 #include "hasp_gui.h"
 
 #include "../hasp/hasp_dispatch.h"
-/* #include "freertos/queue.h"
+#include "freertos/queue.h"
 
 QueueHandle_t queue;
 typedef struct
 {
-    char topic[64];
+    char* topic;   //[64];
-    char payload[512];
+    char* payload; //[512];
-} mqtt_message_t; */
+} mqtt_message_t;
 
 char mqttLwtTopic[28];
 char mqttNodeTopic[24];
@@ -74,6 +74,12 @@ void mqtt_run_scripts()
         //     attempt++;
         // };
 
+        if(current_mqtt_state) {
+            dispatch_run_script(NULL, "L:/mqtt_on.cmd", TAG_HASP);
+        } else {
+            dispatch_run_script(NULL, "L:/mqtt_off.cmd", TAG_HASP);
+        }
+
         last_mqtt_state = current_mqtt_state;
     }
 }
@@ -101,6 +107,10 @@ int mqttPublish(const char* topic, const char* payload, size_t len, bool retain)
 
     // Write directly to the client, don't use the buffer
     if(current_mqtt_state && esp_mqtt_client_publish(mqttClient, topic, payload, len, 0, retain) != ESP_FAIL) {
+
+        // Enqueue a message to the outbox, to be sent later
+        // if(current_mqtt_state && esp_mqtt_client_enqueue(mqttClient, topic, payload, len, 0, retain, true) !=
+        // ESP_FAIL) {
         mqttPublishCount++;
         return MQTT_ERR_OK;
     }
@@ -158,12 +168,52 @@ int mqtt_send_discovery(const char* payload, size_t len)
     return mqttPublish(tmp_topic, payload, len, false);
 }
 
+static inline size_t mqtt_msg_length(size_t len)
+{
+    return (len / 64) * 64 + 64;
+}
+
+void mqtt_process_topic_payload(const char* topic, const char* payload, unsigned int length)
+{
+    if(gui_acquire()) {
+        mqttLoop(); // First empty the MQTT queue
+        LOG_TRACE(TAG_MQTT_RCV, F("%s = %s"), topic, payload);
+        dispatch_topic_payload(topic, payload, length > 0, TAG_MQTT);
+        gui_release();
+    } else {
+        // Add new message to the queue
+        mqtt_message_t data;
+
+        size_t topic_len   = strlen(topic);
+        size_t payload_len = length;
+        data.topic         = (char*)hasp_calloc(sizeof(char), mqtt_msg_length(topic_len + 1));
+        data.payload       = (char*)hasp_calloc(sizeof(char), mqtt_msg_length(payload_len + 1));
+
+        if(!data.topic || !data.payload) {
+            LOG_ERROR(TAG_MQTT_RCV, D_ERROR_OUT_OF_MEMORY);
+            hasp_free(data.topic);
+            hasp_free(data.payload);
+            return;
+        }
+        memcpy(data.topic, topic, topic_len);
+        memcpy(data.payload, payload, payload_len);
+
+        {
+            size_t attempt = 0;
+            while(xQueueSend(queue, &data, (TickType_t)0) == errQUEUE_FULL && attempt < 100) {
+                delay(5);
+                attempt++;
+            };
+        }
+    }
+}
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 // Receive incoming messages
 static void mqtt_message_cb(const char* topic, byte* payload, unsigned int length)
 { // Handle incoming commands from MQTT
     mqttReceiveCount++;
-    LOG_TRACE(TAG_MQTT_RCV, F("[%s] %s = %s"), pcTaskGetTaskName(NULL), topic, (char*)payload);
+    // LOG_TRACE(TAG_MQTT_RCV, F("%s = %s"), topic, (char*)payload);
 
     if(topic == strstr(topic, mqttNodeTopic)) { // startsWith mqttNodeTopic
 
@@ -174,16 +224,12 @@ static void mqtt_message_cb(const char* topic, byte* payload, unsigned int lengt
 
         // Group topic
         topic += strlen(mqttGroupTopic); // shorten topic
-                                         // dispatch_topic_payload(topic, (const char*)payload, length > 0, TAG_MQTT);
-                                         // return;
 
 #ifdef HASP_USE_BROADCAST
     } else if(topic == strstr_P(topic, PSTR(MQTT_PREFIX "/" MQTT_TOPIC_BROADCAST "/"))) { // broadcast  topic
 
         // Broadcast topic
         topic += strlen_P(PSTR(MQTT_PREFIX "/" MQTT_TOPIC_BROADCAST "/")); // shorten topic
-        // dispatch_topic_payload(topic, (const char*)payload, length > 0, TAG_MQTT);
-        // return;
 #endif
 
 #ifdef HASP_USE_HA
@@ -223,9 +269,7 @@ static void mqtt_message_cb(const char* topic, byte* payload, unsigned int lengt
                 else */
 
     {
-        gui_acquire();
+        mqtt_process_topic_payload(topic, (const char*)payload, length);
-        dispatch_topic_payload(topic, (const char*)payload, length > 0, TAG_MQTT);
-        gui_release();
     }
 
     /*    {
@@ -326,8 +370,8 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
     int msg_id;
     switch(event->event_id) {
         case MQTT_EVENT_DISCONNECTED:
-            mqtt_disconnected();
             LOG_WARNING(TAG_MQTT, F(D_MQTT_DISCONNECTED));
+            mqtt_disconnected();
             break;
         case MQTT_EVENT_BEFORE_CONNECT:
             // LOG_INFO(TAG_MQTT, F(D_MQTT_CONNECTING));
@@ -338,7 +382,7 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
             onMqttConnect(event->client);
             break;
         case MQTT_EVENT_SUBSCRIBED:
-            // onMqttSubscribed(event);
+            onMqttSubscribed(event);
             break;
         case MQTT_EVENT_DATA:
             onMqttData(event);
@@ -385,8 +429,7 @@ static esp_err_t mqtt_event_handler(esp_mqtt_event_handle_t event)
 
 void mqttSetup()
 {
-    /*queue = xQueueCreate(20, sizeof(mqtt_message_t)); */
+    queue = xQueueCreate(64, sizeof(mqtt_message_t));
-
     esp_crt_bundle_set(rootca_crt_bundle_start);
     mqttStart();
 }
@@ -395,15 +438,17 @@ IRAM_ATTR void mqttLoop(void)
 {
     // mqttClient.loop();
 
-    /*
+    if(!uxQueueMessagesWaiting(queue)) return;
+
     mqtt_message_t data;
     while(xQueueReceive(queue, &data, (TickType_t)0)) {
-        LOG_DEBUG(TAG_MQTT, F("[%s] Received data from queue == %s\n"), pcTaskGetTaskName(NULL), data.topic);
+        LOG_WARNING(TAG_MQTT, F("[%d] QUE %s => %s"), uxQueueMessagesWaiting(queue), data.topic, data.payload);
         size_t length = strlen(data.payload);
         dispatch_topic_payload(data.topic, data.payload, length > 0, TAG_MQTT);
-        delay(1);
+        hasp_free(data.topic);
+        hasp_free(data.payload);
+        // delay(1);
     }
-    */
 }
 
 void mqttEvery5Seconds(bool networkIsConnected)
@@ -461,6 +506,8 @@ void mqttStart()
     mqtt_cfg.lwt_topic  = mqttLwtTopic;
     mqtt_cfg.lwt_qos    = 1;
 
+    mqtt_cfg.task_prio = 1;
+
     // mqtt_cfg.crt_bundle_attach = esp_crt_bundle_attach;
 
     // test Mosquitto doesn't need a user/pwd