diff --git a/environments.ini b/environments.ini index 9cc0b5c..8f56b16 100644 --- a/environments.ini +++ b/environments.ini @@ -185,6 +185,7 @@ custom_description = Gateway using Somfy Remote library, need CC1101 [env:esp32dev-pilight-somfy-cc1101] platform = ${com.esp32_platform} board = esp32dev +board_build.partitions = min_spiffs.csv lib_deps = ${com-esp32.lib_deps} ${libraries.esppilight} diff --git a/main/ZgatewayBT.ino b/main/ZgatewayBT.ino index 2612e9c..722f4e2 100644 --- a/main/ZgatewayBT.ino +++ b/main/ZgatewayBT.ino @@ -662,8 +662,8 @@ void procBLETask(void* pvParameters) { */ void BLEscan() { // Don't start the next scan until processing of previous results is complete. - while (uxQueueMessagesWaiting(BLEQueue) || jsonQueue.size() != 0) { - yield(); + while (uxQueueMessagesWaiting(BLEQueue) || queueLength != 0) { // the criteria on queueLength could be adjusted to parallelize the scan and the queue processing + delay(1); // Wait for queue to empty, a yield here instead of the delay cause the WDT to trigger } Log.notice(F("Scan begin" CR)); BLEScan* pBLEScan = BLEDevice::getScan(); @@ -806,9 +806,8 @@ void coreTask(void* pvParameters) { delay(waitms = interval > 100 ? 100 : interval); // 100ms } } - } else { - Log.trace(F("BLE core task canceled by processLock" CR)); } + delay(1); } } @@ -872,7 +871,7 @@ void setupBTTasksAndBLE() { xTaskCreatePinnedToCore( procBLETask, /* Function to implement the task */ "procBLETask", /* Name of the task */ - 8000, /* Stack size in bytes */ + 8500, /* Stack size in bytes */ NULL, /* Task input parameter */ 2, /* Priority of the task (set higher than core task) */ &xProcBLETaskHandle, /* Task handle. */ @@ -1031,7 +1030,7 @@ void launchBTDiscovery(bool overrideDiscovery) { 0, "", "", false, "", model.c_str(), brand.c_str(), model_id.c_str(), macWOdots.c_str(), false, stateClassMeasurement, nullptr, nullptr, "[\"lb\",\"kg\",\"jin\"]"); - } else if (strcmp(prop.key().c_str(), "device") != 0 || strcmp(prop.key().c_str(), "mac") != 0) { // Exception on device and mac as these ones are not sensors + } else if (strcmp(prop.key().c_str(), "device") != 0 && strcmp(prop.key().c_str(), "mac") != 0) { // Exception on device and mac as these ones are not sensors createDiscovery("sensor", discovery_topic.c_str(), entity_name.c_str(), unique_id.c_str(), will_Topic, prop.value()["name"], value_template.c_str(), @@ -1138,7 +1137,7 @@ void PublishDeviceData(JsonObject& BLEdata) { } void process_bledata(JsonObject& BLEdata) { - yield(); + yield(); // Necessary to let the loop run in case of connectivity issues if (!BLEdata.containsKey("id")) { Log.error(F("No mac address in the payload" CR)); return; diff --git a/main/main.ino b/main/main.ino index 68ad8f7..0509c9b 100644 --- a/main/main.ino +++ b/main/main.ino @@ -62,8 +62,12 @@ unsigned long timer_sys_checks = 0; # define ARDUINOJSON_ENABLE_STD_STRING 1 # include +int queueLength = 0; +int queueLengthSum = 0; +int blockedMessages = 0; +int maxQueueLength = 0; # ifndef QueueSize -# define QueueSize 32 +# define QueueSize 18 # endif #endif @@ -88,7 +92,6 @@ struct JsonBundle { }; std::queue jsonQueue; -int queueLengthSum = 0; #endif StaticJsonDocument modulesBuffer; @@ -264,7 +267,7 @@ static String ota_server_cert = ""; # include # include -bool ProcessLock = false; // Process lock when we want to use a critical function like OTA for example +bool ProcessLock = true; // Process lock when we want to use a critical function like OTA for example, at start to true so as to wait for critical functions to be performed before BLE start # if !defined(NO_INT_TEMP_READING) // ESP32 internal temperature reading # include @@ -426,11 +429,16 @@ void pubMainCore(JsonObject& data) { // Add a document to the queue void enqueueJsonObject(const StaticJsonDocument& jsonDoc) { if (jsonDoc.size() == 0) { - Log.error(F("Empty JSON, not enqueued" CR)); + Log.error(F("Empty JSON, skipping" CR)); + return; + } +#if defined(ESP8266) || defined(ESP32) || defined(__AVR_ATmega2560__) || defined(__AVR_ATmega1280__) + if (queueLength >= QueueSize) { + Log.warning(F("%d Doc(s) in queue, doc blocked" CR), queueLength); + blockedMessages++; return; } Log.trace(F("Enqueue JSON" CR)); -#if defined(ESP8266) || defined(ESP32) || defined(__AVR_ATmega2560__) || defined(__AVR_ATmega1280__) JsonBundle bundle; bundle.doc = jsonDoc; jsonQueue.push(bundle); @@ -484,20 +492,20 @@ void buildTopicFromId(JsonObject& Jsondata, const char* origin) { // Empty the documents queue void emptyQueue() { - while (true) { - JsonBundle bundle; - - if (jsonQueue.empty()) { - break; - } - Log.trace(F("Dequeue JSON" CR)); - bundle = jsonQueue.front(); - jsonQueue.pop(); - - JsonObject obj = bundle.doc.as(); - pubMainCore(obj); - queueLengthSum++; + queueLength = jsonQueue.size(); + if (queueLength > maxQueueLength) { + maxQueueLength = queueLength; } + if (queueLength == 0) { + return; + } + JsonBundle bundle; + Log.trace(F("Dequeue JSON" CR)); + bundle = jsonQueue.front(); + jsonQueue.pop(); + JsonObject obj = bundle.doc.as(); + pubMainCore(obj); + queueLengthSum++; } #else void emptyQueue() {} @@ -523,6 +531,7 @@ void pub(const char* topicori, const char* payload, bool retainFlag) { */ void pub(const char* topicori, JsonObject& data) { String dataAsString = ""; + bool ret = sensor_Retain; #if defined(ESP8266) || defined(ESP32) # if message_UTCtimestamp == true @@ -532,6 +541,10 @@ void pub(const char* topicori, JsonObject& data) { data["unixtime"] = unixtimestamp(); # endif #endif + if (data.containsKey("retain") && data["retain"].is()) { + ret = data["retain"]; + data.remove("retain"); + } if (data.containsKey("origin")) { // temporary, in the future use this instead of topicori data.remove("origin"); } @@ -559,7 +572,7 @@ void pub(const char* topicori, JsonObject& data) { #if jsonPublishing Log.trace(F("jsonPubl - ON" CR)); - pubMQTT(topic, dataAsString.c_str()); + pubMQTT(topic.c_str(), dataAsString.c_str(), ret); pubWebUI(topicori, data); #endif @@ -2026,14 +2039,20 @@ void loop() { # endif } if (now > (timer_sys_checks + (TimeBetweenCheckingSYS * 1000)) || !timer_sys_checks) { -# if defined(ESP32) && defined(MQTT_HTTPS_FW_UPDATE) - checkForUpdates(); -# endif # if defined(ESP8266) || defined(ESP32) # if message_UTCtimestamp || message_unixtimestamp syncNTP(); # endif # endif + if (!timer_sys_checks) { // Update check at start up only +# if defined(ESP32) && defined(MQTT_HTTPS_FW_UPDATE) + checkForUpdates(); +# endif +# ifdef ZgatewayBT + ProcessLock = false; // Release BLE processes at start +# endif + } + timer_sys_checks = millis(); } #endif @@ -2293,7 +2312,7 @@ void eraseAndRestart() { #if defined(ESP8266) || defined(ESP32) || defined(__AVR_ATmega2560__) || defined(__AVR_ATmega1280__) String stateMeasures() { - StaticJsonDocument SYSdata; + StaticJsonDocument<1024> SYSdata; SYSdata["uptime"] = uptime(); @@ -2312,6 +2331,8 @@ String stateMeasures() { SYSdata["mqttsecure"] = mqtt_secure; SYSdata["msgproc"] = queueLengthSum; SYSdata["msgspeed"] = round2((float)queueLengthSum / uptime()); + SYSdata["msgblock"] = blockedMessages; + SYSdata["maxqueue"] = maxQueueLength; # ifdef ESP32 minFreeMem = ESP.getMinFreeHeap(); SYSdata["minfreemem"] = minFreeMem; @@ -2593,9 +2614,6 @@ String latestVersion; */ bool checkForUpdates() { Log.notice(F("Update check, free heap: %d"), ESP.getFreeHeap()); -# ifdef ZGatewayBT - ProcessLock = true; -# endif HTTPClient http; http.setTimeout((GeneralTimeOut - 1) * 1000); // -1 to avoid WDT http.setFollowRedirects(HTTPC_STRICT_FOLLOW_REDIRECTS); @@ -2616,6 +2634,7 @@ bool checkForUpdates() { Log.error(F("Error on HTTP request")); } http.end(); //Free the resources + Log.notice(F("Update check done, free heap: %d"), ESP.getFreeHeap()); if (jsondata.containsKey("latest_version")) { jsondata["installed_version"] = OMG_VERSION; jsondata["entity_picture"] = ENTITY_PICTURE; @@ -2623,6 +2642,7 @@ bool checkForUpdates() { jsondata["release_summary"] = ""; latestVersion = jsondata["latest_version"].as(); jsondata["origin"] = subjectRLStoMQTT; + jsondata["retain"] = true; enqueueJsonObject(jsondata); Log.trace(F("Update file found on server" CR)); return true; @@ -2630,10 +2650,6 @@ bool checkForUpdates() { Log.trace(F("No update file found on server" CR)); return false; } -# ifdef ZGatewayBT - ProcessLock = false; -# endif - Log.notice(F("Update check done, free heap: %d"), ESP.getFreeHeap()); } # else