Skip to content

Provide ability to send a status msg via MQTT #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mattbnz wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Provide ability to send a status msg via MQTT #14

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b16c665
Provide ability to send a status msg via MQTT
mattbnz Sep 2, 2022
f5cc3bc
Store entire MqttMessage not just 4 bytes...
mattbnz Sep 6, 2022
15f2684
Allow publishing of arbitrary status docs
mattbnz Sep 6, 2022
010f631
Fix missing deletes on error paths.
mattbnz Sep 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions include/mqtt.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ namespace mqtt {

void publishSensors(uint16_t mask);
void publishConfiguration();
void sendStatus(const char *statusmsg);

void mqttLoop(void* pvParameters);

Expand Down
39 changes: 37 additions & 2 deletions src/mqtt.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ namespace mqtt {

const uint8_t X_CMD_PUBLISH_SENSORS = bit(0);
const uint8_t X_CMD_PUBLISH_CONFIGURATION = bit(1);
const uint8_t X_CMD_PUBLISH_STATUSMSG = bit(2);

TaskHandle_t mqttTask;
QueueHandle_t mqttQueue;
Expand All @@ -45,6 +46,8 @@ namespace mqtt {
getSPS30StatusCallback_t getSPS30StatusCallback;

uint32_t lastReconnectAttempt = 0;
uint32_t connectionAttempts = 0;
char statusmsg[512];

void publishSensors(uint16_t mask) {
if (!WiFi.isConnected() || !mqtt_client->connected()) return;
Expand Down Expand Up @@ -255,12 +258,42 @@ namespace mqtt {
}
}

void sendStatus(const char *newmsg) {
MqttMessage msg;
msg.cmd = X_CMD_PUBLISH_STATUSMSG;
sprintf(statusmsg, "%s", newmsg);
Copy link
Owner

@oseiler2 oseiler2 Sep 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's a rare chance that a sendStatusInteral call is in progress while a sentStatus call is made from another thread which then overwrites/corrupts the statusmsg. Can we avoid that e.g. by passing the status string (or a constant that gets resolved to a string) with the message? How long do you anticipate the statue message to get?

Copy link
Owner

@oseiler2 oseiler2 Sep 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if there is a much better solution, and if its worth increasing the queue message size which would reserve a larger chunk of memory. If we could keep the status message short then adding it to the queue message could work, otherwise not so sure.

Copy link
Contributor Author

@mattbnz mattbnz Sep 6, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I originally was storing the string in the message, but couldn't make it work through the queue, so fell back to the global object, but you're right it's messy... I took another look and found the bug that was breaking the queue passing (wrong size being passed to the createQueue method).

Copy link
Owner

@oseiler2 oseiler2 Sep 8, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just wondering - wouldn't the same approach of sending a pointer to the JsonDocument have worked for a String pointer as well as long as the publishStatusInternal method deleted it to free the memory?
I do like the idea of having an option to publish structured data, but it feels a little over kill for publishing a simple status message. What do you think, and do you have future changes that would make use of a generic publish for structured data

if (strlen(statusmsg) > 0) {
if (mqttQueue) xQueueSendToBack(mqttQueue, (void*)&msg, pdMS_TO_TICKS(100));
}
}

void publishStatusInternal() {
if (!mqtt_client->connected()) {
return;
}
char msg[1024];
StaticJsonDocument<1024> json;
json["online"] = true;
if (strlen(statusmsg) > 0) {
json["status"] = statusmsg;
}
if (serializeJson(json, msg) == 0) {
ESP_LOGW(TAG, "Failed to serialise status payload");
return;
}
char buf[256];
sprintf(buf, "%s/%u/up/status", config.mqttTopic, config.deviceId);
mqtt_client->publish(buf, msg);
sprintf(statusmsg, "");
}

void reconnect() {
if (millis() - lastReconnectAttempt < 60000) return;
char buf[256];
sprintf(buf, "CO2Monitor-%u-%s", config.deviceId, WifiManager::getMac().c_str());
if (!WiFi.isConnected()) return;
lastReconnectAttempt = millis();
connectionAttempts++;
if (!mqtt_client->connected()) {
ESP_LOGD(TAG, "Attempting MQTT connection...");
if (mqtt_client->connect(buf, config.mqttUsername, config.mqttPassword)) {
Expand All @@ -269,8 +302,8 @@ namespace mqtt {
mqtt_client->subscribe(buf);
sprintf(buf, "%s/down/#", config.mqttTopic);
mqtt_client->subscribe(buf);
sprintf(buf, "%s/%u/up/status", config.mqttTopic, config.deviceId);
mqtt_client->publish(buf, "{\"online\":true}");
sprintf(statusmsg, "connected after %d attempts", connectionAttempts);
Copy link
Owner

@oseiler2 oseiler2 Sep 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you mind passing the reconnect count as a separate property instead of injecting it into the string?
I've got a mqtt_client->publish(buf, "{\"online\":false}"); message in the (so far unpublished) branch for the portable monitor to signal the device going to disconnect before sleep.

Copy link
Contributor Author

@mattbnz mattbnz Sep 6, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, see what you think about this reworked code... allows an arbitrary JSON doc to be provided as the status message (as a pointer, which the queue takes ownership of to free after serialization).

This allows sending structured data (like offline:false, and connectionAttempts) but still makes it pretty simple to provide a nice wrapper for sending a simple message.

Only 4 more bytes on the queue in the general case (e.g. other messages), but when providing a doc, the caller has control over how much memory to allocate.

publishStatusInternal();
} else {
ESP_LOGW(TAG, "MQTT connection failed, rc=%i", mqtt_client->state());
vTaskDelay(pdMS_TO_TICKS(1000));
Expand Down Expand Up @@ -347,6 +380,8 @@ namespace mqtt {
publishConfigurationInternal();
} else if (msg.cmd == X_CMD_PUBLISH_SENSORS) {
publishSensorsInternal(msg.mask);
} else if (msg.cmd == X_CMD_PUBLISH_STATUSMSG) {
publishStatusInternal();
}
}
if (!mqtt_client->connected()) {
Expand Down
3 changes: 3 additions & 0 deletions src/ota.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
#include <Arduino.h>
#include <config.h>
#include <mqtt.h>
#include <ota.h>
#include <esp32fota.h>
#include <Ticker.h>
Expand Down Expand Up @@ -36,6 +37,7 @@ namespace OTA {
if (shouldExecuteFirmwareUpdate) {
ESP_LOGD(TAG, "Firmware update available");
if (preUpdateCallback) preUpdateCallback();
mqtt::sendStatus("Starting OTA update");
esp32FOTA.execOTA();
} else {
ESP_LOGD(TAG, "No firmware update available");
Expand All @@ -52,6 +54,7 @@ namespace OTA {
ESP_LOGD(TAG, "Beginning forced OTA");
if (preUpdateCallback) preUpdateCallback();
esp32FOTA esp32FOTA(OTA_APP, APP_VERSION, LittleFS, false, false);
mqtt::sendStatus("Starting forced OTA update");
esp32FOTA.forceUpdate(forceUpdateURL, false);
forceUpdateURL = "";
ESP_LOGD(TAG, "Forced OTA done"); forceUpdateURL = "";
Expand Down