I'm in the process of migrating my projects from using a webserver to communiate status messages to using MQQT, which I'm finding to be significantly faster. The one problem I haven't solved is how to have a "persistent status message". What I mean by that is that it's a single string that contains the status of the system.
Examples of that are:
"loading"
"running"
"waiting"
That kind of thing.
It should be able to be checked by any device, even a week after posted, and it always retrieves the latest (and therefore only) status message. Any device can change that status, and it then changes that message everywhere.
In other words I'm just trying to have a single message, not a queue of messages, and that message should be persistent (non expiring).
Let me know if that's not clear, I know it's confusing, and I'm probably not helping, ha.
Here's how I'm currently posting and retrieving from MQQT:
#include <ArduinoJson.h>
#include <WiFi.h>
// WiFi
const char *ssid = "Nacho Cheese"; // Enter your WiFi name
const char *password = "asdf"; // Enter WiFi password
#include <PubSubClient.h>
WiFiClient espClient;
PubSubClient mqtt_client(espClient);
// MQTT Broker
const char *mqtt_broker = "broker.emqx.io";
const char *mqtt_topic = "esp32/testing123";
const char *mqtt_username = "emqx";
const char *mqtt_password = "public";
const int mqtt_port = 1883;
DynamicJsonDocument mqtt_doc(256); // remember to allow enough size!
void setup() {
Serial.begin(115200);
// connecting to a WiFi network
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.println("Connecting to WiFi..");
}
Serial.println("Connected to the WiFi network");
//connecting to a mqtt broker
mqtt_client.setServer(mqtt_broker, mqtt_port);
mqtt_client.setCallback(mqtt_callback);
while (!mqtt_client.connected()) {
String mqtt_client_id = "esp32-client-";
mqtt_client_id += String(WiFi.macAddress());
Serial.printf("The client %s connects to the public mqtt broker\n", mqtt_client_id.c_str());
if (mqtt_client.connect(mqtt_client_id.c_str(), mqtt_username, mqtt_password)) {
Serial.println("Public emqx mqtt broker connected");
} else {
Serial.print("failed with state ");
Serial.print(mqtt_client.state());
delay(2000);
}
}
// publish and subscribe
mqtt_client.publish(mqtt_topic, "Hi EMQX I'm ESP32 ^^");
mqtt_client.subscribe(mqtt_topic);
}
void mqtt_callback(char *mqtt_topic, byte *payload, unsigned int length) {
Serial.print("Message arrived in topic: ");
Serial.println(mqtt_topic);
Serial.println("Message:");
Serial.println();
for (int i = 0; i < length; i++) {
Serial.print((char) payload[i]);
}
Serial.println();
Serial.println("-----------------------");
deserializeJson(mqtt_doc, payload, length);
Serial.println(">>>");
String msg = mqtt_doc["msg"];
Serial.println(msg);
}
void loop() {
long s=millis();
mqtt_client.loop();
long d = millis()-s;
if (d>=2) Serial.println("Done in " + String(d));
}
From the PubSubClient API here just use the retained flag.
boolean publish (topic, payload, [length], [retained])
Publishes a message to the specified topic.
Parameters
topic const char[] - the topic to publish to
payload const char[], byte[] - the message to publish
length unsigned int (optional) - the length of the payload. Required if payload is a byte[]
retained boolean (optional) - whether the message should be retained
false - not retained
true - retained
Returns
false - publish failed, either connection lost or message too large
true - publish succeeded
Thanks, this works well. Now my problem is that I can't obtain the last message in an MQTT topic without subscribing to it. Since I only need to see the last message posted in the past, there's no need for a subscription. And since I'm just running a PHP script (or Python script) once on a shared webserver there's no way for me to subscribe to the topic, that I see at least, since I can't run long and wait for the callback to execute.
Can anyone think of a way to obtain the last message posted to a topic (in the past), without using a callback?
Here's my starter PHP code:
require('vendor/autoload.php');
use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;
if (1) {
$server = 'broker.emqx.io';
$port = 1883;
$clientId = 'test-publisher-wrybread';
$mqtt = new \PhpMqtt\Client\MqttClient($server, $port, $clientId);
$mqtt->connect();
$mqtt->subscribe("esp32/wrybread_test_mode", function ($topic, $message) {
printf("Received message on topic [%s]: %s\n", $topic, $message);
}, 0);
$mqtt->disconnect();
}
Or in Python:
import paho.mqtt.client as mqtt #pip install paho-mqtt
# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
print("Connected with result code "+str(rc))
client.subscribe("esp32/wrybread_test_mode")
# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
print(msg.topic+" "+str(msg.payload))
mqtt_msg = msg.payload
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.emqx.io", 1883, 60)
client.loop_forever()
There is a need because that's the way to get to that message.
If the topic was written with the retain flag your callback will be called almost immediately, so you don't have to wait long for that message to arrive.
# install library from here https://emqx.medium.com/how-to-use-mqtt-in-php-67edc9b41022
require('vendor/autoload.php');
use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;
if (1) {
$server = 'broker.emqx.io';
$port = 1883;
$clientId = 'test-publisher-wrybread';
$mqtt = new \PhpMqtt\Client\MqttClient($server, $port, $clientId);
$mqtt->connect();
$mqtt->subscribe("esp32/wrybread_test_mode", function ($topic, $message) {
printf("Received message on topic [%s]: %s\n", $topic, $message);
}, 0);
$mqtt->disconnect();
}
Python code to do the same:
#!/usr/bin/python
# pip install paho-mqtt
import paho.mqtt.client as mqtt
import os
import json
import cgi
import time
import threading
# how many seconds to wait for a message at most?
max_time = 5
start_time = time.time();
# A watchdog thread to terminate the script if no message is found.
# Because apparently mqtt doesn't have a way to simply return "no message found" for a given topic.
def thread_function(name):
while True:
if time.time() - start_time > max_time:
print ("Maximum time reached and no message was found! Exiting.")
os._exit(1)
time.sleep(.1)
watchdog_thread = threading.Thread(target=thread_function, args=(1,))
watchdog_thread.start()
def on_connect(client, userdata, flags, rc):
#print("Connected with result code "+str(rc))
# Subscribing in on_connect() means that if we lose the connection and
# reconnect then subscriptions will be renewed.
client.subscribe("esp32/teste")
def on_message(client, userdata, msg):
#print(msg.topic+" "+str(msg.payload))
mqtt_msg = msg.payload
execution_time = time.time() - start_time
deets = {
"current_mode": mqtt_msg,
"station_name": station_name,
#"execution_time": execution_time,
}
print ("Content-type: application/json\n")
print( json.dumps(deets) )
# exit right away!
#sys.exit()
os._exit(1)
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.emqx.io", 1883, 60)
client.loop_forever()
The trick is to exit the loop when the topic is read. The obvious problem to that approach is that if there's no message in the topic, the script will essentially hang as it waits for a message... I solved that in the python script by having a watchdog thread that will terminate the script if it runs too long, which happens when there's nothing in the topic I'm subscribing to.