WordClockESP/lib/framework/MqttPubSub.h
rjwats 0e2124062f
Use references & flash strings where approperate (#110)
* pass originId as const reference
* store strings for serial logging in flash
* Use string references where approperate.
2020-05-21 08:42:21 +01:00

167 lines
5.0 KiB
C++

#ifndef MqttPubSub_h
#define MqttPubSub_h
#include <StatefulService.h>
#include <JsonSerializer.h>
#include <JsonDeserializer.h>
#include <AsyncMqttClient.h>
#define MAX_MESSAGE_SIZE 1024
#define MQTT_ORIGIN_ID "mqtt"
template <class T>
class MqttConnector {
protected:
StatefulService<T>* _statefulService;
AsyncMqttClient* _mqttClient;
MqttConnector(StatefulService<T>* statefulService, AsyncMqttClient* mqttClient) :
_statefulService(statefulService), _mqttClient(mqttClient) {
_mqttClient->onConnect(std::bind(&MqttConnector::onConnect, this));
}
virtual void onConnect() = 0;
public:
inline AsyncMqttClient* getMqttClient() const {
return _mqttClient;
}
};
template <class T>
class MqttPub : virtual public MqttConnector<T> {
public:
MqttPub(JsonSerializer<T> jsonSerializer,
StatefulService<T>* statefulService,
AsyncMqttClient* mqttClient,
const String& pubTopic = "") :
MqttConnector<T>(statefulService, mqttClient), _jsonSerializer(jsonSerializer), _pubTopic(pubTopic) {
MqttConnector<T>::_statefulService->addUpdateHandler([&](const String& originId) { publish(); }, false);
}
void setPubTopic(const String& pubTopic) {
_pubTopic = pubTopic;
publish();
}
protected:
virtual void onConnect() {
publish();
}
private:
JsonSerializer<T> _jsonSerializer;
String _pubTopic;
void publish() {
if (_pubTopic.length() > 0 && MqttConnector<T>::_mqttClient->connected()) {
// serialize to json doc
DynamicJsonDocument json(MAX_MESSAGE_SIZE);
JsonObject jsonObject = json.to<JsonObject>();
MqttConnector<T>::_statefulService->read(jsonObject, _jsonSerializer);
// serialize to string
String payload;
serializeJson(json, payload);
// publish the payload
MqttConnector<T>::_mqttClient->publish(_pubTopic.c_str(), 0, false, payload.c_str());
}
}
};
template <class T>
class MqttSub : virtual public MqttConnector<T> {
public:
MqttSub(JsonDeserializer<T> jsonDeserializer,
StatefulService<T>* statefulService,
AsyncMqttClient* mqttClient,
const String& subTopic = "") :
MqttConnector<T>(statefulService, mqttClient), _jsonDeserializer(jsonDeserializer), _subTopic(subTopic) {
MqttConnector<T>::_mqttClient->onMessage(std::bind(&MqttSub::onMqttMessage,
this,
std::placeholders::_1,
std::placeholders::_2,
std::placeholders::_3,
std::placeholders::_4,
std::placeholders::_5,
std::placeholders::_6));
}
void setSubTopic(const String& subTopic) {
if (!_subTopic.equals(subTopic)) {
// unsubscribe from the existing topic if one was set
if (_subTopic.length() > 0) {
MqttConnector<T>::_mqttClient->unsubscribe(_subTopic.c_str());
}
// set the new topic and re-configure the subscription
_subTopic = subTopic;
subscribe();
}
}
protected:
virtual void onConnect() {
subscribe();
}
private:
JsonDeserializer<T> _jsonDeserializer;
String _subTopic;
void subscribe() {
if (_subTopic.length() > 0) {
MqttConnector<T>::_mqttClient->subscribe(_subTopic.c_str(), 2);
}
}
void onMqttMessage(char* topic,
char* payload,
AsyncMqttClientMessageProperties properties,
size_t len,
size_t index,
size_t total) {
// we only care about the topic we are watching in this class
if (strcmp(_subTopic.c_str(), topic)) {
return;
}
// deserialize from string
DynamicJsonDocument json(MAX_MESSAGE_SIZE);
DeserializationError error = deserializeJson(json, payload, len);
if (!error && json.is<JsonObject>()) {
JsonObject jsonObject = json.as<JsonObject>();
MqttConnector<T>::_statefulService->update(jsonObject, _jsonDeserializer, MQTT_ORIGIN_ID);
}
}
};
template <class T>
class MqttPubSub : public MqttPub<T>, public MqttSub<T> {
public:
MqttPubSub(JsonSerializer<T> jsonSerializer,
JsonDeserializer<T> jsonDeserializer,
StatefulService<T>* statefulService,
AsyncMqttClient* mqttClient,
const String& pubTopic = "",
const String& subTopic = "") :
MqttConnector<T>(statefulService, mqttClient),
MqttPub<T>(jsonSerializer, statefulService, mqttClient, pubTopic),
MqttSub<T>(jsonDeserializer, statefulService, mqttClient, subTopic) {
}
public:
void configureTopics(const String& pubTopic, const String& subTopic) {
MqttSub<T>::setSubTopic(subTopic);
MqttPub<T>::setPubTopic(pubTopic);
}
protected:
void onConnect() {
MqttSub<T>::onConnect();
MqttPub<T>::onConnect();
}
};
#endif // end MqttPubSub