MQTTKit.h

Go to the documentation of this file.

/*
 * This is a part of the Uniot project.
 * Copyright (C) 2016-2024 Uniot <contact@uniot.io>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */



 
#pragma once
 
#if defined(ESP8266)
#include "ESP8266WiFi.h"
#elif defined(ESP32)
#include "WiFi.h"
#endif
 
#include <Bytes.h>
#include <CBORObject.h>
#include <COSEMessage.h>
#include <ClearQueue.h>
#include <Common.h>
#include <Date.h>
#include <EventListener.h>
#include <NetworkEvents.h>
#include <NetworkScheduler.h>
#include <MQTTEvents.h>
#include <PubSubClient.h>
#include <TaskScheduler.h>
 
#include <atomic>
#include <functional>
 
#if defined(ESP32)
#include <lwip/tcp.h>
#include <sys/socket.h>
#endif
 
#include "CallbackMQTTDevice.h"
#include "MQTTDevice.h"
#include "MQTTPath.h"
 
namespace uniot {
class MQTTKit : public ISchedulerConnectionKit, public CoreEventListener {
  typedef std::function<void(CBORObject &)> CBORExtender;
  friend class MQTTDevice;
 
 public:
  MQTTKit(const Credentials &credentials, CBORExtender infoExtender = nullptr)
      : mpCredentials(&credentials),
        mPath(credentials),
        mInfoExtender(infoExtender),
        mPubSubClient(mWiFiClient),
        mNetworkConnected(false),
        mConnectionId(0),
        mMqttConnected(false) {
    mPubSubClient.setCallback([this](char *topic, uint8_t *payload, unsigned int length) {
      mDevices.forEach([&](MQTTDevice *device) {
        if (device->isSubscribed(String(topic))) {
          if (!length) {
            device->handle(topic, Bytes());
            return;
          }
 
          Bytes decoded;
          if (_readCOSEMessage(Bytes(payload, length), decoded)) {
            device->handle(topic, decoded);
          } else {
            UNIOT_LOG_ERROR("failed to decode message on topic: %s", topic);
          }
        }
      });
    });
    _initTasks();
    CoreEventListener::listenToEvent(events::network::Topic::CONNECTION);
    CoreEventListener::listenToEvent(events::date::Topic::TIME);
 
    // mWiFiClient.allowSelfSignedCerts();
    // mWiFiClient.setInsecure();
  }

  ~MQTTKit() {
    CoreEventListener::stopListeningToEvent(events::network::Topic::CONNECTION);
    CoreEventListener::stopListeningToEvent(events::date::Topic::TIME);
    // TODO: implement, remove all devices
  }

  void setServer(const char *domain, uint16_t port) {
    mPubSubClient.setServer(domain, port);
  }

  void addDevice(MQTTDevice &device) {
    if (mDevices.pushUnique(&device)) {
      device.kit(this);
      device.topics()->forEach([this](String topic) {
        mPubSubClient.subscribe(topic.c_str());
      });
    }
  }

  void removeDevice(MQTTDevice &device) {
    if (mDevices.removeOne(&device)) {
      device.kit(nullptr);
      device.topics()->forEach([this](String topic) {
        mPubSubClient.unsubscribe(topic.c_str());
      });
    }
  }

  const MQTTPath &getPath() {
    return mPath;
  }

  void renewSubscriptions() {
    mDevices.forEach([this](MQTTDevice *device) {
      device->unsubscribeFromAll();
      device->syncSubscriptions();
    });
  }

  virtual void pushTo(TaskScheduler &scheduler) override {
    scheduler.push("mqtt", mTaskMQTT);
    scheduler.push("ntp_retry", mTaskNtpRetry);
  }

  virtual void attach() override {}

  virtual void onEventReceived(unsigned int topic, int msg) override {
    if (events::network::Topic::CONNECTION == topic) {
      switch (msg) {
        case events::network::Msg::SUCCESS:
          mNetworkConnected = true;
          // forceSync() blocks for up to ~2 s. Run it from a scheduler task
          // rather than here, so the event bus dispatch is not stalled.
          if (!mTaskNtpRetry->isAttached()) {
            mTaskNtpRetry->once(1);
          }
          break;
        case events::network::Msg::ACCESS_POINT:
        case events::network::Msg::AVAILABLE:
        case events::network::Msg::CONNECTING:
        case events::network::Msg::DISCONNECTING:
        case events::network::Msg::DISCONNECTED:
        case events::network::Msg::FAILED:
        default:
          mNetworkConnected = false;
          mMqttConnected.store(false);
          mTaskMQTT->detach();
          mTaskNtpRetry->detach();
          break;
      }
      return;
    }
    if (events::date::Topic::TIME == topic) {
      switch (msg) {
        case events::date::Msg::SYNCED:
          mTaskNtpRetry->detach();
          if (!mTaskMQTT->isAttached()) {
            mTaskMQTT->attach(10);
          }
          break;
        case events::date::Msg::SYNC_FAILED:
          // MQTT cannot start without a valid clock, so keep retrying.
          // Re-armed unconditionally: once() on an already-attached task simply
          // restarts the timer, and guarding on isAttached() here would stall
          // the retry chain permanently if the task were still attached.
          mTaskNtpRetry->once(3000);
          break;
        default:
          break;
      }
      return;
    }
  }

  bool isMqttConnected() const { return mMqttConnected.load(); }

  void forceDisconnect() {
    Bytes packetExtention;
    if (mInfoExtender) {
      CBORObject packet;
      mInfoExtender(packet);
      packetExtention = packet.build();
    }
    CBORObject offlineCBOR(packetExtention);
    _prepareOfflinePacket(offlineCBOR);
    auto offlinePacket = _buildCOSEMessage(offlineCBOR.build());
    mPubSubClient.publish(
        mPath.buildDevicePath("status").c_str(),
        offlinePacket.raw(),
        offlinePacket.size(),
        true);
    mPubSubClient.disconnect();
    mMqttConnected.store(false);
  }
 
 protected:
  PubSubClient *client() {
    return &mPubSubClient;
  }
 
 private:
  inline void _initTasks() {
    mTaskNtpRetry = TaskScheduler::make([this](SchedulerTask &self, short t) {
      // Active only until NTP succeeds; detached on TIME/SYNCED. forceSync() is
      // blocking, which a scheduler tick tolerates but the event bus does not.
      if (mNetworkConnected) {
        UNIOT_LOG_DEBUG("NTP retry, attempting forced sync");
        Date::getInstance().forceSync();
      }
    });
 
    mTaskMQTT = TaskScheduler::make([this](SchedulerTask &self, short t) {
      if (!mNetworkConnected) {
        UNIOT_LOG_DEBUG("network is not connected");
        return;
      }
      if (!mPubSubClient.connected()) {
        mMqttConnected.store(false);
        UNIOT_LOG_DEBUG("attempting MQTT connection #%d", mConnectionId);
        Bytes packetExtention;
        if (mInfoExtender) {
          CBORObject packet;
          mInfoExtender(packet);
          packetExtention = packet.build();
        }
 
        CBORObject offlineCBOR(packetExtention);
        _prepareOfflinePacket(offlineCBOR);
        auto offlinePacket = _buildCOSEMessage(offlineCBOR.build());
        auto password = _getUserPassword();
        if (mPubSubClient.connect(
                _getClientId().c_str(),
                _getUserLogin().c_str(),
                (const char *)password.raw(),
                password.size(),
                mPath.buildDevicePath("status").c_str(),
                0,
                true,
                (const char *)offlinePacket.raw(),
                offlinePacket.size(),
                true)) {
          CBORObject onlineCBOR(packetExtention);
          _prepareOnlinePacket(onlineCBOR);
          auto onlinePacket = _buildCOSEMessage(onlineCBOR.build());
          mPubSubClient.publish(
              mPath.buildDevicePath("status").c_str(),
              onlinePacket.raw(),
              onlinePacket.size(),
              true);  // publish an announcement
          mDevices.forEach([this](MQTTDevice *device) {
            device->topics()->forEach([this](String topic) {
              mPubSubClient.subscribe(topic.c_str());
            });
          });
          _applyTcpKeepalive();
          mMqttConnected.store(true);
          CoreEventEmitter::emitEvent(events::mqtt::Topic::CONNECTION, events::mqtt::Msg::SUCCESS);
        } else {
          mMqttConnected.store(false);
          CoreEventEmitter::emitEvent(events::mqtt::Topic::CONNECTION, events::mqtt::Msg::FAILED);
        }
      }
      mPubSubClient.loop();
    });
  }

  void _applyTcpKeepalive() {
#if defined(ESP32)
    int fd = mWiFiClient.fd();
    if (fd < 0) {
      UNIOT_LOG_WARN("TCP keepalive skipped, invalid socket fd");
      return;
    }
    static constexpr int kEnable = 1;
    static constexpr int kIdle = 10;   // seconds before the first probe; must be < MQTT_KEEPALIVE
    static constexpr int kIntvl = 5;   // seconds between probes
    static constexpr int kCount = 3;   // probes before giving up
    setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &kEnable, sizeof(kEnable));
    setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &kIdle, sizeof(kIdle));
    setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &kIntvl, sizeof(kIntvl));
    setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &kCount, sizeof(kCount));
    UNIOT_LOG_DEBUG("TCP keepalive applied, idle: %ds, interval: %ds, count: %d", kIdle, kIntvl, kCount);
#endif
  }

  Bytes _buildCOSEMessage(const Bytes &payload, bool sign = false) {
    COSEMessage obj;
    obj.setPayload(payload);
    auto kid = mpCredentials->keyId();  // FIXME: dynamic data must be within the scope of the obj.build() function
    if (sign) {
      obj.sign(*mpCredentials);
      obj.setUnprotectedKid(kid);
    }
    return obj.build();
  }

  bool _readCOSEMessage(const Bytes &message, Bytes &outPayload) {
    COSEMessage obj(message);
    if (obj.wasReadSuccessful()) {
      outPayload = obj.getPayload();
      return true;
    }
    return false;
  }

  void _prepareOnlinePacket(CBORObject &packet) {
    packet
        .put("online", 1)
        .put("connection_id", mConnectionId++);
  }

  void _prepareOfflinePacket(CBORObject &packet) {
    packet
        .put("online", 0)
        .put("connection_id", mConnectionId);
  }

  String _getClientId() {
    return _getUserLogin();
  }

  String _getUserLogin() {
    return "device:" + mpCredentials->getOwnerId() + "@" + mpCredentials->getDeviceId();
  }

  Bytes _getUserPassword() {
    CBORObject claims;
    claims.put("device", mpCredentials->getDeviceId().c_str());
    claims.put("owner", mpCredentials->getOwnerId().c_str());
    claims.put("creator", mpCredentials->getCreatorId().c_str());
    claims.put("timestamp", static_cast<int64_t>(Date::now()));
 
    COSEMessage password;
    auto kid = mpCredentials->keyId();  // FIXME: dynamic data must be within the scope of the obj.build() function
    password.setUnprotectedKid(kid);
    password.setPayload(claims.build());
    password.sign(*mpCredentials);
    return password.build();
  }
 
  const Credentials *mpCredentials;  
 
  MQTTPath mPath;                   
  CBORExtender mInfoExtender;       
  PubSubClient mPubSubClient;       
 
  bool mNetworkConnected;           
  int mConnectionId;                
  std::atomic<bool> mMqttConnected; 
 
  WiFiClient mWiFiClient;           
  // WiFiClientSecure mWiFiClient;  /**< Secure TCP client (commented out) */
  ClearQueue<MQTTDevice *> mDevices;    
  TaskScheduler::TaskPtr mTaskMQTT;     
  TaskScheduler::TaskPtr mTaskNtpRetry; 
};

}  // namespace uniot