chriz/OpenMQTT-Gateway
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

[SYS] Implement a central queue (#1767)

Browse Source 
to avoid concurrency issues when modules or/and gateways want to publish data
This commit is contained in:
Florian authored and GitHub committed 2023-11-08 07:26:21 -06:00
1 parent ca9ef664ab
commit 029409c0c3
39 files changed
+491 -407
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

main/ZgatewayBT.ino
+62 -139
View File
@@ -113,8 +113,9 @@ void btScanWDG() {
if (!ProcessLock &&
previousBtScanCount == scanCount &&
scanCount != 0 &&
(now - lastBtScan > BTConfig.BLEinterval)) {
(now - lastBtScan > ((BTConfig.BLEinterval + BTConfig.scanDuration) < GeneralTimeOut ? GeneralTimeOut + 1000 : (BTConfig.BLEinterval + BTConfig.scanDuration)))) {
Log.error(F("BLE Scan watchdog triggered at : %ds" CR), lastBtScan / 1000);
stopProcessing();
ESPRestart(4);
} else {
previousBtScanCount = scanCount;
@@ -148,10 +149,6 @@ String stateBTMeasures(bool start) {
jo["ignoreWBlist"] = BTConfig.ignoreWBlist;
jo["presenceawaytimer"] = BTConfig.presenceAwayTimer;
jo["movingtimer"] = BTConfig.movingTimer;
jo["btqblck"] = btQueueBlocked;
jo["btqsum"] = btQueueLengthSum;
jo["btqsnd"] = btQueueLengthCount;
jo["btqavg"] = (btQueueLengthCount > 0 ? btQueueLengthSum / (float)btQueueLengthCount : 0);
jo["bletaskstack"] = uxTaskGetStackHighWaterMark(xProcBLETaskHandle);
jo["blecoretaskstack"] = uxTaskGetStackHighWaterMark(xCoreTaskHandle);
@@ -163,7 +160,8 @@ String stateBTMeasures(bool start) {
}
String output;
serializeJson(jo, output);
pub(subjectBTtoMQTT, jo);
jo["origin"] = subjectBTtoMQTT;
enqueueJsonObject(jo);
return (output);
}
@@ -315,103 +313,10 @@ void BTConfig_load() {
}
}
void pubBTMainCore(JsonObject& data, bool haPresenceEnabled = true) {
if (abs((int)data["rssi"] | 0) < abs(BTConfig.minRssi) && data.containsKey("id")) {
String topic = data["id"].as<const char*>();
topic.replace(":", ""); // Initially publish topic ends with MAC address
if (BTConfig.pubBeaconUuidForTopic && !BTConfig.extDecoderEnable && data.containsKey("model_id") && data["model_id"].as<String>() == "IBEACON")
topic = data["uuid"].as<const char*>(); // If model_id is IBEACON, use uuid as topic
if (BTConfig.extDecoderEnable && !data.containsKey("model"))
topic = BTConfig.extDecoderTopic; // If external decoder, use this topic to send data
topic = subjectBTtoMQTT + String("/") + topic;
pub((char*)topic.c_str(), data);
}
if (haPresenceEnabled && data.containsKey("distance")) {
if (data.containsKey("servicedatauuid"))
data.remove("servicedatauuid");
if (data.containsKey("servicedata"))
data.remove("servicedata");
if (BTConfig.presenceUseBeaconUuid && data.containsKey("model_id") && data["model_id"].as<String>() == "IBEACON") {
data["mac"] = data["id"];
data["id"] = data["uuid"];
}
String topic = String(mqtt_topic) + BTConfig.presenceTopic + String(gateway_name);
Log.trace(F("Pub HA Presence %s" CR), topic.c_str());
pub_custom_topic((char*)topic.c_str(), data, false);
}
}
class JsonBundle {
public:
StaticJsonDocument<JSON_MSG_BUFFER> buffer;
JsonObject object;
bool haPresence;
JsonObject& createObject(const char* json = NULL, bool haPresenceEnabled = true) {
buffer.clear();
haPresence = haPresenceEnabled;
object = buffer.to<JsonObject>();
if (json != nullptr) {
auto error = deserializeJson(buffer, json);
if (error) {
Log.error(F("deserialize object failed: %s" CR), error.c_str());
}
}
return object;
}
};
void PublishDeviceData(JsonObject& BLEdata);
atomic_int forceBTScan;
JsonBundle jsonBTBufferQueue[BTQueueSize];
atomic_int jsonBTBufferQueueNext, jsonBTBufferQueueLast;
int btQueueBlocked = 0;
int btQueueLengthSum = 0;
int btQueueLengthCount = 0;
JsonObject& getBTJsonObject(const char* json, bool haPresenceEnabled) {
int next, last;
for (bool blocked = false;;) {
next = atomic_load_explicit(&jsonBTBufferQueueNext, ::memory_order_seq_cst); // use namespace std -> ambiguous error...
last = atomic_load_explicit(&jsonBTBufferQueueLast, ::memory_order_seq_cst); // use namespace std -> ambiguous error...
if ((2 * BTQueueSize + last - next) % (2 * BTQueueSize) != BTQueueSize) break;
if (!blocked) {
blocked = true;
btQueueBlocked++;
}
delay(1);
}
return jsonBTBufferQueue[last % BTQueueSize].createObject(json, haPresenceEnabled);
}
// should run from the BT core
void pubBT(JsonObject& data) {
int last = atomic_load_explicit(&jsonBTBufferQueueLast, ::memory_order_seq_cst);
atomic_store_explicit(&jsonBTBufferQueueLast, (last + 1) % (2 * BTQueueSize), ::memory_order_seq_cst); // use namespace std -> ambiguous error...
}
// should run from the main core
void emptyBTQueue() {
for (bool first = true;;) {
int next = atomic_load_explicit(&jsonBTBufferQueueNext, ::memory_order_seq_cst); // use namespace std -> ambiguous error...
int last = atomic_load_explicit(&jsonBTBufferQueueLast, ::memory_order_seq_cst); // use namespace std -> ambiguous error...
if (last == next) break;
if (first) {
int diff = (2 * BTQueueSize + last - next) % (2 * BTQueueSize);
btQueueLengthSum += diff;
btQueueLengthCount++;
first = false;
}
JsonBundle& bundle = jsonBTBufferQueue[next % BTQueueSize];
pubBTMainCore(bundle.object, bundle.haPresence);
atomic_store_explicit(&jsonBTBufferQueueNext, (next + 1) % (2 * BTQueueSize), ::memory_order_seq_cst); // use namespace std -> ambiguous error...
vTaskDelay(1);
}
}
void createOrUpdateDevice(const char* mac, uint8_t flags, int model, int mac_type = 0);
BLEdevice* getDeviceByMac(const char* mac); // Declared here to avoid pre-compilation issue (misplaced auto declaration by pio)
@@ -509,11 +414,13 @@ void updateDevicesStatus() {
p->sensorModel_id == TheengsDecoder::BLE_ID_NUM::RUUVITAG_RAWV2)) { // We apply the offline status only for tracking device, can be extended further to all the devices
if ((p->lastUpdate != 0) && (p->lastUpdate < (now - BTConfig.presenceAwayTimer) && (now > BTConfig.presenceAwayTimer)) &&
(BTConfig.ignoreWBlist || ((!oneWhite || isWhite(p)) && !isBlack(p)))) { // Only if WBlist is disabled OR ((no white MAC OR this MAC is white) AND not a black listed MAC)) {
JsonObject BLEdata = getBTJsonObject();
StaticJsonDocument<JSON_MSG_BUFFER> BLEdataBuffer;
JsonObject BLEdata = BLEdataBuffer.to<JsonObject>();
BLEdata["id"] = p->macAdr;
BLEdata["state"] = "offline";
pubBT(BLEdata);
// We set the lastUpdate to 0 to avoid republishing the offline state
buildTopicFromId(BLEdata, subjectBTtoMQTT);
enqueueJsonObject(BLEdata);
// We set the lastUpdate to 0 to avoid replublishing the offline state
p->lastUpdate = 0;
}
}
@@ -521,11 +428,13 @@ void updateDevicesStatus() {
if (p->sensorModel_id == TheengsDecoder::BLE_ID_NUM::BC08) {
if ((p->lastUpdate != 0) && (p->lastUpdate < (now - BTConfig.movingTimer) && (now > BTConfig.movingTimer)) &&
(BTConfig.ignoreWBlist || ((!oneWhite || isWhite(p)) && !isBlack(p)))) { // Only if WBlist is disabled OR ((no white MAC OR this MAC is white) AND not a black listed MAC)) {
JsonObject BLEdata = getBTJsonObject();
StaticJsonDocument<JSON_MSG_BUFFER> BLEdataBuffer;
JsonObject BLEdata = BLEdataBuffer.to<JsonObject>();
BLEdata["id"] = p->macAdr;
BLEdata["state"] = "offline";
pubBT(BLEdata);
// We set the lastUpdate to 0 to avoid republishing the offline state
buildTopicFromId(BLEdata, subjectBTtoMQTT);
enqueueJsonObject(BLEdata);
// We set the lastUpdate to 0 to avoid replublishing the offline state
p->lastUpdate = 0;
}
}
@@ -682,16 +591,22 @@ void procBLETask(void* pvParameters) {
for (;;) {
xQueueReceive(BLEQueue, &advertisedDevice, portMAX_DELAY);
// Feed the watchdog
//esp_task_wdt_reset();
if (!ProcessLock) {
Log.trace(F("Creating BLE buffer" CR));
JsonObject& BLEdata = getBTJsonObject();
String mac_adress = advertisedDevice->getAddress().toString().c_str();
mac_adress.toUpperCase();
BLEdata["id"] = (char*)mac_adress.c_str();
StaticJsonDocument<JSON_MSG_BUFFER> BLEdataBuffer;
JsonObject BLEdata = BLEdataBuffer.to<JsonObject>();
std::string mac_address = advertisedDevice->getAddress().toString();
for (char& c : mac_address) {
c = std::toupper(c);
}
BLEdata["id"] = mac_address;
BLEdata["mac_type"] = advertisedDevice->getAddress().getType();
BLEdata["adv_type"] = advertisedDevice->getAdvType();
Log.notice(F("Device detected: %s" CR), (char*)mac_adress.c_str());
Log.notice(F("Device detected: %s" CR), BLEdata["id"].as<const char*>());
BLEdevice* device = getDeviceByMac(BLEdata["id"].as<const char*>());
if (BTConfig.filterConnectable && device->connect) {
@@ -718,31 +633,21 @@ void procBLETask(void* pvParameters) {
if (advertisedDevice->haveServiceData()) {
int serviceDataCount = advertisedDevice->getServiceDataCount();
Log.trace(F("Get services data number: %d" CR), serviceDataCount);
char initialBLEDataCharArray[JSON_MSG_BUFFER];
serializeJson(BLEdata, initialBLEDataCharArray);
for (int j = 0; j < serviceDataCount; j++) {
StaticJsonDocument<JSON_MSG_BUFFER> BLEdataBufferTemp;
JsonObject BLEdataTemp = BLEdataBufferTemp.to<JsonObject>();
BLEdataBufferTemp = BLEdataBuffer;
std::string service_data = convertServiceData(advertisedDevice->getServiceData(j));
Log.trace(F("Service data: %s" CR), service_data.c_str());
std::string serviceDatauuid = advertisedDevice->getServiceDataUUID(j).toString();
Log.trace(F("Service data UUID: %s" CR), (char*)serviceDatauuid.c_str());
if (j == 0) {
BLEdata["servicedata"] = (char*)service_data.c_str();
BLEdata["servicedatauuid"] = (char*)serviceDatauuid.c_str();
PublishDeviceData(BLEdata);
} else {
StaticJsonDocument<JSON_MSG_BUFFER> doc;
deserializeJson(doc, initialBLEDataCharArray);
JsonObject& BLEdataTemp = getBTJsonObject();
BLEdataTemp = doc.as<JsonObject>();
BLEdataTemp["servicedata"] = (char*)service_data.c_str();
BLEdataTemp["servicedatauuid"] = (char*)serviceDatauuid.c_str();
PublishDeviceData(BLEdataTemp);
}
BLEdataTemp["servicedata"] = (char*)service_data.c_str();
BLEdataTemp["servicedatauuid"] = (char*)serviceDatauuid.c_str();
PublishDeviceData(BLEdataTemp);
}
} else {
PublishDeviceData(BLEdata);
}
} else {
Log.trace(F("Filtered MAC device" CR));
}
@@ -757,7 +662,7 @@ void procBLETask(void* pvParameters) {
*/
void BLEscan() {
// Don't start the next scan until processing of previous results is complete.
while (uxQueueMessagesWaiting(BLEQueue)) {
while (uxQueueMessagesWaiting(BLEQueue) || jsonQueue.size() != 0) {
yield();
}
Log.notice(F("Scan begin" CR));
@@ -967,7 +872,7 @@ void setupBTTasksAndBLE() {
xTaskCreatePinnedToCore(
procBLETask, /* Function to implement the task */
"procBLETask", /* Name of the task */
5120, /* Stack size in bytes */
8000, /* Stack size in bytes */
NULL, /* Task input parameter */
2, /* Priority of the task (set higher than core task) */
&xProcBLETaskHandle, /* Task handle. */
@@ -1000,8 +905,6 @@ void setupBT() {
Log.notice(F("Moving Timer: %d" CR), BTConfig.movingTimer);
atomic_init(&forceBTScan, 0); // in theory, we don't need this
atomic_init(&jsonBTBufferQueueNext, 0); // in theory, we don't need this
atomic_init(&jsonBTBufferQueueLast, 0); // in theory, we don't need this
semaphoreCreateOrUpdateDevice = xSemaphoreCreateBinary();
xSemaphoreGive(semaphoreCreateOrUpdateDevice);
@@ -1009,7 +912,7 @@ void setupBT() {
semaphoreBLEOperation = xSemaphoreCreateBinary();
xSemaphoreGive(semaphoreBLEOperation);
BLEQueue = xQueueCreate(32, sizeof(NimBLEAdvertisedDevice*));
BLEQueue = xQueueCreate(QueueSize, sizeof(NimBLEAdvertisedDevice*));
setupBTTasksAndBLE();
@@ -1191,6 +1094,7 @@ void PublishDeviceData(JsonObject& BLEdata) {
process_bledata(BLEdata);
// If the device is a random MAC and pubRandomMACs is false we don't publish this payload
if (!BTConfig.pubRandomMACs && (BLEdata["type"].as<string>()).compare("RMAC") == 0) {
Log.trace(F("Random MAC, device filtered" CR));
return;
}
// If pubAdvData is false we don't publish the adv data
@@ -1209,21 +1113,38 @@ void PublishDeviceData(JsonObject& BLEdata) {
}
// If the device is not a sensor and pubOnlySensors is true we don't publish this payload
if (!BTConfig.pubOnlySensors || BLEdata.containsKey("model") || BLEdata.containsKey("distance")) { // Identified device
pubBT(BLEdata);
buildTopicFromId(BLEdata, subjectBTtoMQTT);
enqueueJsonObject(BLEdata);
} else {
Log.notice(F("Not a sensor device filtered" CR));
return;
}
} else if (BLEdata.containsKey("distance")) {
pubBT(BLEdata);
if (BLEdata.containsKey("servicedatauuid"))
BLEdata.remove("servicedatauuid");
if (BLEdata.containsKey("servicedata"))
BLEdata.remove("servicedata");
if (BLEdata.containsKey("manufacturerdata"))
BLEdata.remove("manufacturerdata");
if (BTConfig.presenceUseBeaconUuid && BLEdata.containsKey("model_id") && BLEdata["model_id"].as<String>() == "IBEACON") {
BLEdata["mac"] = BLEdata["id"];
BLEdata["id"] = BLEdata["uuid"];
}
enqueueJsonObject(BLEdata);
} else {
Log.trace(F("Low rssi, device filtered" CR));
Log.notice(F("Low rssi, device filtered" CR));
return;
}
}
void process_bledata(JsonObject& BLEdata) {
yield();
if (!BLEdata.containsKey("id")) {
Log.error(F("No mac address in the payload" CR));
return;
}
const char* mac = BLEdata["id"].as<const char*>();
Log.trace(F("Processing BLE data %s" CR), BLEdata["id"].as<const char*>());
int model_id = BTConfig.extDecoderEnable ? -1 : decoder.decodeBLEJson(BLEdata);
int mac_type = BLEdata["mac_type"].as<int>();
if ((BLEdata["type"].as<string>()).compare("RMAC") != 0 && model_id != TheengsDecoder::BLE_ID_NUM::IBEACON) { // Do not store in memory the random mac devices and iBeacons
@@ -1350,10 +1271,12 @@ void immediateBTAction(void* pvParameters) {
xSemaphoreGive(semaphoreBLEOperation);
} else {
Log.error(F("BLE busy - command not sent" CR));
JsonObject result = getBTJsonObject();
result["id"] = BLEactions.back().addr;
result["success"] = false;
pubBT(result);
StaticJsonDocument<JSON_MSG_BUFFER> BLEdataBuffer;
JsonObject BLEdata = BLEdataBuffer.to<JsonObject>();
BLEdata["id"] = BLEactions.back().addr;
BLEdata["success"] = false;
buildTopicFromId(BLEdata, subjectBTtoMQTT);
enqueueJsonObject(BLEdata);
BLEactions.pop_back();
ProcessLock = false;
}