bridge.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <chrono>
#include <mutex>
#include <unordered_map>
#include <vector>
 
#include "spsp/local_addr_mac.hpp"
#include "spsp/logger.hpp"
#include "spsp/node.hpp"
#include "spsp/timer.hpp"
#include "spsp/wildcard_trie.hpp"
 
// Log tag
#define SPSP_LOG_TAG "SPSP/Bridge"
 
namespace SPSP::Nodes
{
    static constexpr auto BRIDGE_SUB_NO_EXPIRE = std::chrono::milliseconds::max();
 
    struct BridgeConfig
    {
        struct Reporting
        {
            bool probePayload = true;  
            bool rssiOnProbe = true;   
            bool rssiOnPub = true;     
            bool rssiOnSub = true;     
            bool rssiOnUnsub = true;   
        };
 
        struct SubDB
        {
            std::chrono::milliseconds interval = std::chrono::minutes(1);
 
            std::chrono::milliseconds subLifetime = std::chrono::minutes(15);
        };
 
        Reporting reporting;
        SubDB subDB;
    };
 
    template <typename TLocalLayer, typename TFarLayer>
    class Bridge : public ILocalAndFarNode<TLocalLayer, TFarLayer>
    {
    public:
        using LocalAddrT = typename TLocalLayer::LocalAddrT;
        using LocalMessageT = typename TLocalLayer::LocalMessageT;
 
    protected:
        struct SubDBEntry
        {
            std::chrono::milliseconds lifetime;  
            SPSP::SubscribeCb cb = nullptr;      
        };
 
        using SubDBMapT = std::unordered_map<LocalAddrT, SubDBEntry>;
 
        std::mutex m_mutex;               
        BridgeConfig m_conf;              
        WildcardTrie<SubDBMapT> m_subDB;  
        Timer m_subDBTimer;               
 
    public:
        Bridge(TLocalLayer* ll, TFarLayer* fl, BridgeConfig conf = {})
            : ILocalAndFarNode<TLocalLayer, TFarLayer>{ll, fl},
              m_conf{conf},
              m_subDBTimer{conf.subDB.interval,
                           std::bind(&Bridge<TLocalLayer, TFarLayer>::subDBTick,
                           this)}
        {
            SPSP_LOGI("Initialized");
        }
 
        ~Bridge()
        {
            SPSP_LOGI("Deinitialized");
        }
 
        bool receiveFar(const std::string& topic, const std::string& payload)
        {
            const std::scoped_lock lock(m_mutex);
 
            SPSP_LOGD("Received far msg: topic '%s', payload '%s'",
                      topic.c_str(), payload.c_str());
 
            // Get matching entries
            auto entries = m_subDB.find(topic);
 
            for (auto& [entryTopic, entryMap] : entries) {
                for (auto& [addr, entry] : entryMap) {
                    if (addr == LocalAddrT{}) {
                        // This node's subscription - call callback
                        SPSP_LOGD("Calling user callback for topic '%s' in new thread",
                                  topic.c_str());
                        std::thread t(entry.cb, topic, payload);
                        t.detach();
                    } else {
                        // Local layer subscription
                        std::thread t(&Bridge<TLocalLayer, TFarLayer>::publishSubData,
                                      this, addr, topic, payload);
                        t.detach();
                    }
                }
            }
 
            return true;
        }
 
        bool publish(const std::string& topic, const std::string& payload)
        {
            SPSP_LOGD("Publishing locally: topic '%s', payload '%s'",
                      topic.c_str(), payload.c_str());
 
            if (topic.empty()) {
                SPSP_LOGW("Can't publish to empty topic");
                return false;
            }
 
            return this->getFarLayer()->publish(LocalAddrMAC::local().str,
                                                topic, payload);
        }
 
        bool subscribe(const std::string& topic, SubscribeCb cb)
        {
            const std::scoped_lock lock(m_mutex);
 
            SPSP_LOGD("Subscribing locally to topic '%s'", topic.c_str());
 
            if (topic.empty()) {
                SPSP_LOGW("Can't subscribe to empty topic");
                return false;
            }
 
            auto& entryMap = m_subDB[topic];
 
            // Attempt to subscribe to new topic
            if (entryMap.empty()) {
                if (!this->getFarLayer()->subscribe(topic)) {
                    return false;
                }
            }
 
            entryMap[LocalAddrT{}] = SubDBEntry{
                .lifetime = BRIDGE_SUB_NO_EXPIRE,
                .cb = cb
            };
 
            return true;
        }
 
        void resubscribeAll()
        {
            const std::scoped_lock lock(m_mutex);
 
            m_subDB.forEach(
                [this](const std::string& topic, const SubDBMapT& topicEntries) {
                    if (!this->getFarLayer()->subscribe(topic)) {
                        SPSP_LOGW("Resubscribe to topic %s failed",
                                  topic.c_str());
                    }
                }
            );
        }
 
        bool unsubscribe(const std::string& topic)
        {
            SPSP_LOGD("Unsubscribing locally from topic '%s'", topic.c_str());
 
            if (topic.empty()) {
                SPSP_LOGW("Can't unsubscribe from empty topic");
                return false;
            }
 
            {
                const std::scoped_lock lock(m_mutex);
                if (!m_subDB[topic].erase(LocalAddrT{})) {
                    // Entry doesn't exist
                    SPSP_LOGD("Can't unsubscribe from not-subscribed topic '%s'",
                              topic.c_str());
                    return false;
                }
            }
 
            // Remove unused topics
            this->subDBRemoveUnusedTopics();
 
            return true;
        }
 
    protected:
        bool processProbeReq(const LocalMessageT& req,
                             int rssi = NODE_RSSI_UNKNOWN)
        {
            LocalMessageT res = req;
            res.type = LocalMessageType::PROBE_RES;
            res.payload = "";
 
            // Publish RSSI
            if (m_conf.reporting.rssiOnProbe) {
                this->publishRssi(req.addr, rssi);
            }
 
            // Publish payload
            if (m_conf.reporting.probePayload && !req.payload.empty()) {
                std::string probePayloadReportTopic =
                    NODE_REPORTING_TOPIC + "/" +
                    NODE_REPORTING_PROBE_PAYLOAD_SUBTOPIC + "/" +
                    req.addr.str;
 
                this->publish(probePayloadReportTopic, req.payload);
            }
 
            return this->sendLocal(res);
        }
 
        bool processProbeRes(const LocalMessageT& req,
                             int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processPub(const LocalMessageT& req,
                        int rssi = NODE_RSSI_UNKNOWN)
        {
            // Publish RSSI
            if (m_conf.reporting.rssiOnPub) {
                this->publishRssi(req.addr, rssi);
            }
 
            if (req.topic.empty()) {
                SPSP_LOGW("Can't publish to empty topic");
                return false;
            }
 
            return this->getFarLayer()->publish(req.addr.str, req.topic,
                                                req.payload);
        }
 
        bool processSubReq(const LocalMessageT& req,
                           int rssi = NODE_RSSI_UNKNOWN)
        {
            // Publish RSSI
            if (m_conf.reporting.rssiOnSub) {
                this->publishRssi(req.addr, rssi);
            }
 
            if (req.topic.empty()) {
                SPSP_LOGW("Can't subscribe to empty topic");
                return false;
            }
 
            {
                const std::scoped_lock lock(m_mutex);
 
                auto& entryMap = m_subDB[req.topic];
 
                // Attempt to subscribe to new topic
                if (entryMap.empty()) {
                    if (!this->getFarLayer()->subscribe(req.topic)) {
                        return false;
                    }
                }
 
                entryMap[req.addr] = SubDBEntry{
                    .lifetime = m_conf.subDB.subLifetime,
                    .cb = nullptr
                };
            }
 
            return true;
        }
 
        bool processSubData(const LocalMessageT& req,
                            int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processUnsub(const LocalMessageT& req,
                          int rssi = NODE_RSSI_UNKNOWN)
        {
            // Publish RSSI
            if (m_conf.reporting.rssiOnUnsub) {
                this->publishRssi(req.addr, rssi);
            }
 
            if (req.topic.empty()) {
                SPSP_LOGW("Can't unsubscribe from empty topic");
                return false;
            }
 
            {
                const std::scoped_lock lock(m_mutex);
                m_subDB[req.topic].erase(req.addr);
            }
 
            // Remove unused topics
            this->subDBRemoveUnusedTopics();
 
            return true;
        }
 
        bool processTimeReq(const LocalMessageT& req,
                            int rssi = NODE_RSSI_UNKNOWN)
        {
            // Get current time (millisecond accuracy)
            auto now = std::chrono::system_clock::now().time_since_epoch();
            auto nowMilliseconds =
                std::chrono::duration_cast<std::chrono::milliseconds>(now);
 
            LocalMessageT res = req;
            res.type = LocalMessageType::TIME_RES;
            res.payload = std::to_string(nowMilliseconds.count());
 
            return this->sendLocal(res);
        }
 
        bool processTimeRes(const LocalMessageT& req,
                            int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool publishSubData(const LocalAddrT& addr, const std::string& topic,
                            const std::string& payload)
        {
            SPSP_LOGD("Sending SUB_DATA to %s: topic '%s'",
                      addr.str.c_str(), topic.c_str());
 
            LocalMessageT msg = {};
            msg.addr = addr;
            msg.type = LocalMessageType::SUB_DATA;
            msg.topic = topic;
            msg.payload = payload;
 
            return this->sendLocal(msg);
        }
 
        void subDBTick()
        {
            SPSP_LOGD("SubDB: Tick running");
 
            this->subDBDecrementLifetimes();
            this->subDBRemoveExpiredEntries();
            this->subDBRemoveUnusedTopics();
 
            SPSP_LOGD("SubDB: Tick done");
        }
 
        void subDBDecrementLifetimes()
        {
            const std::scoped_lock lock(m_mutex);
 
            m_subDB.forEach([this] (const std::string& topic,
                                    const SubDBMapT& entryMap) {
                for (auto& [addr, entry] : entryMap) {
                    // Don't decrement entries with infinite lifetime
                    if (entry.lifetime != BRIDGE_SUB_NO_EXPIRE) {
                        m_subDB[topic][addr].lifetime -= m_conf.subDB.interval;
                    }
                }
            });
        }
 
        void subDBRemoveExpiredEntries()
        {
            using namespace std::chrono_literals;
 
            const std::scoped_lock lock(m_mutex);
 
            m_subDB.forEach([this] (const std::string& topic,
                                    const SubDBMapT& entryMap) {
                auto entryIt = entryMap.begin();
                while (entryIt != entryMap.end()) {
                    auto addr = entryIt->first;
 
                    if (entryIt->second.lifetime <= 0ms) {
                        // Expired
                        entryIt = m_subDB[topic].erase(entryIt);
                        SPSP_LOGD("SubDB: Removed addr %s from topic '%s'",
                                  addr.str.c_str(), topic.c_str());
                    } else {
                        // Continue
                        entryIt++;
                    }
                }
            });
        }
 
        void subDBRemoveUnusedTopics()
        {
            const std::scoped_lock lock(m_mutex);
 
            std::vector<std::string> unusedTopics;
 
            // Get unused topics
            m_subDB.forEach([&unusedTopics] (const std::string& topic,
                                             const SubDBMapT& entryMap) {
                if (entryMap.empty()) {
                    unusedTopics.push_back(topic);
                }
            });
 
            // Unsubscribe from them
            for (auto& topic : unusedTopics) {
                if (this->getFarLayer()->unsubscribe(topic)) {
                    // Unsub successful, remove topic from sub DB
                    m_subDB.remove(topic);
                    SPSP_LOGD("SubDB: Removed unused topic '%s'", topic.c_str());
                } else {
                    SPSP_LOGW("SubDB: Topic '%s' can't be unsubscribed. Will try again in next tick.",
                              topic.c_str());
                }
            }
        }
    };
} // namespace SPSP::Nodes
 
#undef SPSP_LOG_TAG