mqtt.h

// MQTT.h
 
#pragma once
 
// #if defined(__ESP__)
 
#include <functional>
 
#include <Arduino_JSON.h>
#include <PubSubClient.h>  // ESP32 requires v2.7 or later
 
#include "ustd_platform.h"
#include "ustd_array.h"
#include "ustd_map.h"
 
#include "scheduler.h"
#include "timeout.h"
 
#include "munet.h"
 
namespace ustd {
 
class Mqtt {
  private:
    // muwerk task management
    Scheduler *pSched;
    int tID;
 
    // mqtt client
    WiFiClient wifiClient;
    PubSubClient mqttClient;
 
    // active configuration
    String mqttServer;
    uint16_t mqttServerPort;
    String mqttUsername;
    String mqttPassword;
    bool mqttRetained;
    String clientName;
    String domainToken;
    String outDomainToken;
    String lwTopic;
    String lwMsg;
    // computed configuration
    ustd::array<String> ownedPrefixes;
    bool bStateRetained;
    String outDomainPrefix;  // outDomainToken + '/' + clientName, or just clientName, if
                             // outDomainToken==""
 
    // persistently initialized tables
    ustd::array<String> subsList;
    ustd::array<String> outgoingBlockList;
    ustd::array<String> incomingBlockList;
 
    // runtime control - state management
    bool isOn = false;
    bool netUp = false;
    bool bMqInit = false;
    bool bWarned = false;
    bool bCheckConnection = false;
    bool mqttConnected = false;
    ustd::timeout mqttTickerTimeout = 5000L;
 
  public:
    Mqtt() {
        mqttClient = wifiClient;
    }
 
    ~Mqtt() {
    }
 
    void begin(Scheduler *_pSched, String _mqttServer = "", uint16_t _mqttServerPort = 1883,
               bool _mqttRetained = false, String _clientName = "${hostname}",
               String _domainToken = "mu", String _outDomainToken = "omu",
               String _mqttUsername = "", String _mqttPassword = "", String _willTopic = "",
               String _willMessage = "") {
        ustd::jsonfile conf;
 
        // read configuration
        mqttServer = conf.readString("mqtt/host", _mqttServer);
        mqttServerPort = (uint16_t)conf.readLong("mqtt/port", 1, 65535, _mqttServerPort);
        mqttUsername = conf.readString("mqtt/username", _mqttUsername);
        mqttPassword = conf.readString("mqtt/password", _mqttPassword);
        mqttRetained = conf.readBool("mqtt/alwaysRetained", _mqttRetained);
        clientName = conf.readString("mqtt/clientName", 1, _clientName);
        domainToken = conf.readString("mqtt/domainToken", 1, _domainToken);
        outDomainToken = conf.readString("mqtt/outDomainToken", _outDomainToken);
        lwTopic = conf.readString("mqtt/lastWillTopic", _willTopic);
        lwMsg = conf.readString("mqtt/lastWillMessage", _willMessage);
 
        // persistently initialized tables
        conf.readStringArray("mqtt/subscriptions", subsList);
        conf.readStringArray("mqtt/outgoingBlackList", outgoingBlockList);
        conf.readStringArray("mqtt/incomingBlackList", incomingBlockList);
 
        // This configuration is preliminary but it is ok. Currently we have no network connection
        // and nothing can happen with this prelimiary information. As soon as a network connection
        // is established, the configuration information will be finalized. This is not possible now
        // since the replacement of placeholders must be able to access some network stack
        // information like the mac id or the hostname but this information is inaccesible if the
        // network stack has not been enabled and configured.
 
        // init scheduler
        pSched = _pSched;
        tID = pSched->add([this]() { this->loop(); }, "mqtt");
 
        // subscribe to all messages
        pSched->subscribe(tID, "#", [this](String topic, String msg, String originator) {
            this->subsMsg(topic, msg, originator);
        });
 
        if (mqttServer.length()) {
            // query update from network stack
            pSched->publish("net/network/get");
        } else {
            DBG("mqtt: WARNING - no server defined.");
        }
 
        // initialize runtime
        isOn = true;
        netUp = false;
        bMqInit = configureMqttClient();
        bWarned = false;
        bStateRetained = false;
        bCheckConnection = false;
        mqttConnected = false;
        mqttTickerTimeout = 5000L;  // 5 seconds
 
        publishState();
    }
 
    int addSubscription(int taskID, String topic, T_SUBS subs, String originator = "") {
        int handle;
        pSched->subscribe(taskID, topic, subs, originator);
        for (unsigned int i = 0; i < subsList.length(); i++) {
            if (topic == subsList[i])
                return handle;  // Already subbed via mqtt.
        }
        if (mqttConnected) {
            mqttClient.subscribe(topic.c_str());
        }
        subsList.add(topic);
        return handle;
    }
 
    bool removeSubscription(int subscriptionHandle, String topic) {
        bool ret = pSched->unsubscribe(subscriptionHandle);
        for (unsigned int i = 0; i < subsList.length(); i++) {
            if (topic == subsList[i])
                subsList.erase(i);
        }
        return ret;
    }
 
    bool outgoingBlockSet(String topic) {
        for (unsigned int i = 0; i < outgoingBlockList.length(); i++) {
            if (outgoingBlockList[i] == topic)
                return true;
        }
        if (outgoingBlockList.add(topic) == -1)
            return false;
        return true;
    }
 
    bool outgoingBlockRemove(String topic) {
        for (unsigned int i = 0; i < outgoingBlockList.length(); i++) {
            if (outgoingBlockList[i] == topic) {
                if (!outgoingBlockList.erase(i))
                    return false;
                return true;
            }
        }
        return false;
    }
 
    bool incomingBlockSet(String topic) {
        for (unsigned int i = 0; i < incomingBlockList.length(); i++) {
            if (incomingBlockList[i] == topic)
                return true;
        }
        if (incomingBlockList.add(topic) == -1)
            return false;
        return true;
    }
 
    bool incomingBlockRemove(String topic) {
        for (unsigned int i = 0; i < incomingBlockList.length(); i++) {
            if (incomingBlockList[i] == topic) {
                if (!incomingBlockList.erase(i))
                    return false;
                return true;
            }
        }
        return false;
    }
 
  private:
    inline void publishState() {
        pSched->publish("mqtt/state", mqttConnected ? "connected" : "disconnected");
    }
 
    void loop() {
        if (!isOn || !netUp || mqttServer.length() == 0) {
            return;
        }
        if (mqttConnected) {
            mqttClient.loop();
        }
        if (bCheckConnection || mqttTickerTimeout.test()) {
            mqttTickerTimeout.reset();
            bCheckConnection = false;
            if (!mqttClient.connected()) {
                // Attempt to connect
                const char *usr = mqttUsername.length() ? mqttUsername.c_str() : NULL;
                const char *pwd = mqttPassword.length() ? mqttPassword.c_str() : NULL;
                bool conRes = mqttClient.connect(clientName.c_str(), usr, pwd, lwTopic.c_str(), 0,
                                                 true, lwMsg.c_str());
                if (conRes) {
                    DBG2("Connected to mqtt server");
                    mqttConnected = true;
                    mqttClient.subscribe((clientName + "/#").c_str());
                    mqttClient.subscribe((domainToken + "/#").c_str());
                    for (unsigned int i = 0; i < subsList.length(); i++) {
                        mqttClient.subscribe(subsList[i].c_str());
                    }
                    bWarned = false;
                    pSched->publish("mqtt/config", outDomainPrefix + "+" + lwTopic + "+" + lwMsg);
                    publishState();
                } else {
                    mqttConnected = false;
                    if (!bWarned) {
                        bWarned = true;
                        publishState();
                        DBG2("MQTT disconnected.");
                    }
                }
            }
        }
    }
 
    void mqttReceive(char *ctopic, unsigned char *payload, unsigned int length) {
        String msg;
        String topic;
 
        // prepare message and topic
        topic = (const char *)ctopic;
        if (length && payload) {
            char *szBuffer = (char *)malloc(length + 1);
            if (szBuffer) {
                memcpy(szBuffer, payload, length);
                szBuffer[length] = 0;
                msg = szBuffer;
                free(szBuffer);
            } else {
                DBG("mqtt: ERROR - message body lost due to memory outage");
            }
        }
 
        for (unsigned int i = 0; i < incomingBlockList.length(); i++) {
            if (Scheduler::mqttmatch(topic, incomingBlockList[i])) {
                // blocked incoming
                DBG2("mqtt: Blocked " + topic);
                return;
            }
        }
        for (unsigned int i = 0; i < subsList.length(); i++) {
            if (Scheduler::mqttmatch(topic, subsList[i])) {
                DBG2("mqtt: subscribed topic " + topic);
                pSched->publish(topic, msg, "mqtt");
                return;
            }
        }
        // strip the client name token or the domain token in messages for us
        for (unsigned int i = 0; i < ownedPrefixes.length(); i++) {
            if (ownedPrefixes[i].length() <= topic.length()) {
                // basically this comparison is not really needed since at this point we could
                // ONLY have messages that match either the domainToken or the clientName since
                // we have exactly subscribed to those. But who knows....
                if (ownedPrefixes[i] == topic.substring(0, ownedPrefixes[i].length())) {
                    topic = (const char *)(ctopic + ownedPrefixes[i].length());
                    pSched->publish(topic, msg, "mqtt");
                }
            }
        }
    }
 
    void subsMsg(String topic, String msg, String originator) {
        if (originator == "mqtt") {
            return;  // avoid loops
        }
 
        // router function
        if (mqttConnected) {
            unsigned int len = msg.length() + 1;
            for (unsigned int i = 0; i < outgoingBlockList.length(); i++) {
                if (Scheduler::mqttmatch(topic, outgoingBlockList[i])) {
                    // Item is blocked.
                    return;
                }
            }
            String tpc;
            if (topic.c_str()[0] == '!') {
                tpc = &(topic.c_str()[1]);
            } else {
                tpc = outDomainPrefix + "/" + topic;
            }
 
            bool bRetain = mqttRetained;
            if (tpc.c_str()[0] == '!') {
                // remove second exclamation point
                tpc = &(topic.c_str()[2]);
                bRetain = true;
            }
            if (!bRetain && bStateRetained && topic == "mqtt/state") {
                // the state topic shall always be retained
                bRetain = true;
            }
 
            DBG3("mqtt: publishing...");
            if (mqttClient.publish(tpc.c_str(), msg.c_str(), bRetain)) {
                DBG2("mqtt publish: " + topic + " | " + msg);
            } else {
                DBG("mqtt: ERROR len=" + String(len) + ", not published: " + topic + " | " + msg);
                if (len > 128) {
                    DBG("mqtt: FATAL ERROR: you need to re-compile the PubSubClient library "
                        "and "
                        "increase #define MQTT_MAX_PACKET_SIZE.");
                }
            }
        } else {
            DBG2("mqtt: NO CONNECTION, not published: " + topic + " | " + msg);
        }
 
        // internal processing
        if (topic == "mqtt/state/get") {
            publishState();
        } else if (topic == "mqtt/config/get") {
            pSched->publish("mqtt/config", outDomainPrefix + "+" + lwTopic + "+" + lwMsg);
        } else if (topic == "mqtt/outgoingblock/set") {
            outgoingBlockSet(msg);
        } else if (topic == "mqtt/outgoingblock/remove") {
            outgoingBlockRemove(msg);
        } else if (topic == "mqtt/incomingblock/set") {
            incomingBlockSet(msg);
        } else if (topic == "mqtt/incomingblock/remove") {
            incomingBlockRemove(msg);
        } else if (topic == "net/network") {
            // network state received:
            JSONVar jsonState = JSON.parse(msg);
            if (JSON.typeof(jsonState) != "object") {
                DBG("mqtt: Received broken network state " + msg);
                return;
            }
            String state = (const char *)jsonState["state"];
            String hostname = (const char *)jsonState["hostname"];
            String mac = (const char *)jsonState["mac"];
            if (state == "connected") {
                DBG3("mqtt: received network connect");
                if (!netUp) {
                    DBG2("mqtt: net state online");
                    finalizeConfiguration(hostname, mac);
                    netUp = true;
                    bCheckConnection = true;
                }
            } else {
                netUp = false;
                mqttConnected = false;
                publishState();
                DBG2("mqtt: net state offline");
            }
        }
    }
 
    bool configureMqttClient() {
        if (mqttServer.length() == 0) {
            DBG2("mqtt: No mqtt host defined. Ignoring configuration...");
            return false;
        }
        bCheckConnection = true;
        mqttClient.setServer(mqttServer.c_str(), mqttServerPort);
 
        if (!bMqInit) {
            mqttClient.setCallback([this](char *topic, unsigned char *msg, unsigned int len) {
                this->mqttReceive(topic, msg, len);
            });
            bMqInit = true;
        }
        return true;
    }
 
    void finalizeConfiguration(String &hostname, String &mac) {
        // get network information
        if (hostname.length() == 0) {
#if defined(__ESP32__)
            String hostname = WiFi.getHostname();
#else
            String hostname = WiFi.hostname();
#endif
        }
        if (mac.length() == 0) {
            mac = WiFi.macAddress();
        }
        mac.replace(":", "");
 
        // transform and integrate missing configuration data
        clientName = replaceVars(clientName, hostname, mac);
        if (outDomainToken.length()) {
            outDomainPrefix = outDomainToken + "/" + clientName;
        } else {
            outDomainPrefix = clientName;
        }
        if (lwTopic.length()) {
            lwMsg = replaceVars(lwMsg, hostname, mac);
        } else {
            lwTopic = outDomainPrefix + "/mqtt/state";
            lwMsg = "disconnected";
            bStateRetained = true;
        }
        String clientPrefix = clientName + "/";
        String domainPrefix = domainToken + "/";
        ownedPrefixes.erase();
        ownedPrefixes.add(clientPrefix);
        ownedPrefixes.add(domainPrefix);
    }
 
    String replaceVars(String val, String &hostname, String &macAddress) {
        val.replace("${hostname}", hostname);
        val.replace("${mac}", macAddress);
        val.replace("${macls}", macAddress.substring(6));
        val.replace("${macfs}", macAddress.substring(0, 5));
        return val;
    }
};
 
}  // namespace ustd
 
// #endif  // defined(__ESP__)