client.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <chrono>
#include <cinttypes>
#include <future>
#include <mutex>
#include <unordered_map>
#include <sys/time.h>  // Unix and ESP
 
#include "spsp/logger.hpp"
#include "spsp/node.hpp"
#include "spsp/timer.hpp"
#include "spsp/wildcard_trie.hpp"
 
// Log tag
#define SPSP_LOG_TAG "SPSP/Client"
 
namespace SPSP::Nodes
{
    struct ClientConfig
    {
        struct Reporting
        {
            bool rssiOnProbe = true;  
        };
 
        struct SubDB
        {
            std::chrono::milliseconds interval = std::chrono::minutes(1);
 
            std::chrono::milliseconds subLifetime = std::chrono::minutes(10);
        };
 
        Reporting reporting;
        SubDB subDB;
 
        std::chrono::milliseconds timeSyncTimeout = std::chrono::seconds(2);
    };
 
    template <typename TLocalLayer>
    class Client : public ILocalNode<TLocalLayer>
    {
    protected:
        struct SubDBEntry
        {
            std::chrono::milliseconds lifetime;  
            SPSP::SubscribeCb cb = nullptr;      
        };
 
        std::mutex m_mutex;                    
        ClientConfig m_conf;                   
        WildcardTrie<SubDBEntry> m_subDB;      
        Timer m_subDBTimer;                    
        bool m_timeSyncOngoing = false;        
        std::promise<bool> m_timeSyncPromise;  
 
    public:
        using LocalAddrT = typename TLocalLayer::LocalAddrT;
        using LocalMessageT = typename TLocalLayer::LocalMessageT;
 
        Client(TLocalLayer* ll, ClientConfig conf = {})
            : ILocalNode<TLocalLayer>{ll}, m_conf{conf}, m_subDB{},
              m_subDBTimer{conf.subDB.interval,
                           std::bind(&Client<TLocalLayer>::subDBTick, this)}
        {
            SPSP_LOGI("Initialized");
        }
 
        ~Client()
        {
            SPSP_LOGI("Deinitialized");
        }
 
        bool publish(const std::string& topic, const std::string& payload)
        {
            SPSP_LOGD("Publishing: topic '%s', payload '%s'",
                      topic.c_str(), payload.c_str());
 
            if (topic.empty()) {
                SPSP_LOGW("Can't publish to empty topic");
                return false;
            }
 
            LocalMessageT msg = {};
            // msg.addr is default => send to the bridge node
            msg.type = LocalMessageType::PUB;
            msg.topic = topic;
            msg.payload = payload;
 
            return this->sendLocal(msg);
        }
 
        bool subscribe(const std::string& topic, SubscribeCb cb)
        {
            SPSP_LOGD("Subscribing to topic '%s'", topic.c_str());
 
            if (topic.empty()) {
                SPSP_LOGW("Can't subscribe to empty topic");
                return false;
            }
 
            if (this->sendSubscribe(topic)) {
                // Subscribe request delivered successfully
                const std::scoped_lock lock(m_mutex);
 
                // Add to sub DB
                SubDBEntry subDBEntry = {
                    .lifetime = m_conf.subDB.subLifetime,
                    .cb = cb
                };
                m_subDB.insert(topic, subDBEntry);
 
                return true;
            }
 
            return false;
        }
 
        bool unsubscribe(const std::string& topic)
        {
            SPSP_LOGD("Unsubscribing from topic '%s'", topic.c_str());
 
            if (topic.empty()) {
                SPSP_LOGW("Can't unsubscribe from empty topic");
                return false;
            }
 
            LocalMessageT msg = {};
            // msg.addr is default => send to the bridge node
            msg.type = LocalMessageType::UNSUB;
            msg.topic = topic;
            // msg.payload is empty
 
            // Remove from sub DB
            {
                const std::scoped_lock lock(m_mutex);
                if (!m_subDB.remove(topic)) {
                    // Not subscribed to this topic
                    SPSP_LOGW("Can't unsubscribe from not-subscribed topic '%s'",
                              topic.c_str());
                    return false;
                }
            }
 
            // Explicitly unsubscribe from bridge
            // If this fails, the timeout on bridge will just expire in
            // couple of minutes.
            this->sendLocal(msg);
 
            return true;
        }
 
        void resubscribeAll()
        {
            const std::scoped_lock lock(m_mutex);
 
            m_subDB.forEach([this](const std::string& topic, const SubDBEntry&) {
                if (!this->sendSubscribe(topic)) {
                    SPSP_LOGW("Resubscribe to topic %s failed", topic.c_str());
                }
            });
        }
 
        bool syncTime()
        {
            SPSP_LOGD("Time sync: start");
 
            // Cleanup lambda to reset state
            auto cleanup = [this]() {
                const std::scoped_lock lock(m_mutex);
                m_timeSyncOngoing = false;
                m_timeSyncPromise = std::promise<bool>{};
            };
 
            LocalMessageT msg = {};
            // msg.addr is default => send to the bridge node
            msg.type = LocalMessageType::TIME_REQ;
            // msg.topic is empty
            // msg.payload is empty
 
            // Set state and get future
            std::future<bool> future;
            {
                const std::scoped_lock lock(m_mutex);
                m_timeSyncOngoing = true;
                future = m_timeSyncPromise.get_future();
            }
 
            // Try to send sync request
            if (!this->sendLocal(msg)) {
                cleanup();
                SPSP_LOGE("Time sync: request can't be sent");
                return false;
            }
 
            if (future.wait_for(m_conf.timeSyncTimeout) == std::future_status::timeout) {
                cleanup();
                SPSP_LOGE("Time sync: response timeout");
                return false;
            }
 
            if (future.get() == false) {
                cleanup();
                SPSP_LOGE("Time sync: invalid bridge response");
                return false;
            }
 
            cleanup();
            SPSP_LOGD("Time sync: success");
            return true;
        }
 
    protected:
        bool processProbeReq(const LocalMessageT& req,
                             int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processProbeRes(const LocalMessageT& req,
                             int rssi = NODE_RSSI_UNKNOWN)
        {
            if (m_conf.reporting.rssiOnProbe) {
                this->publishRssi(req.addr, rssi);
            }
            return true;
        }
 
        bool processPub(const LocalMessageT& req,
                        int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processSubReq(const LocalMessageT& req,
                           int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processSubData(const LocalMessageT& req,
                            int rssi = NODE_RSSI_UNKNOWN)
        {
            // Get matching entries
            std::unordered_map<std::string, const SubDBEntry&> entries;
            {
                const std::scoped_lock lock(m_mutex);
                entries = m_subDB.find(req.topic);
            }
 
            for (auto& [subTopic, entry] : entries) {
                SPSP_LOGD("Calling user callback for topic '%s'",
                          req.topic.c_str());
                entry.cb(req.topic, req.payload);
            }
 
            return true;
        }
 
        bool processUnsub(const LocalMessageT& req,
                          int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processTimeReq(const LocalMessageT& req,
                            int rssi = NODE_RSSI_UNKNOWN) { return false; }
 
        bool processTimeRes(const LocalMessageT& req,
                            int rssi = NODE_RSSI_UNKNOWN)
        {
            const std::scoped_lock lock(m_mutex);
 
            auto nowMilliseconds = stoull(req.payload);
 
            // No time sync ongoing now
            if (!m_timeSyncOngoing) {
                return false;
            }
 
            // Timestamp must have at least 13 digits
            if (nowMilliseconds < 1e12) {
                SPSP_LOGE("Time sync: invalid time received from bridge: '%s'",
                          req.payload.c_str());
                m_timeSyncPromise.set_value(false);
                return false;
            }
 
            // Set current time
            struct timeval tv = {
                .tv_sec = static_cast<long int>(nowMilliseconds / 1000),
                .tv_usec = static_cast<long int>((nowMilliseconds % 1000) * 1000),
            };
 
            if (settimeofday(&tv, nullptr) != 0) {
                SPSP_LOGE("Time sync: settimeofday failed");
                m_timeSyncPromise.set_value(false);
                return false;
            }
 
            SPSP_LOGI("Time sync: set current time to %llu",
                      nowMilliseconds);
            m_timeSyncPromise.set_value(true);
            return true;
        }
 
        bool sendSubscribe(const std::string& topic)
        {
            LocalMessageT msg = {};
            // msg.addr is default => send to the bridge node
            msg.type = LocalMessageType::SUB_REQ;
            msg.topic = topic;
            // msg.payload is empty
 
            return this->sendLocal(msg);
        }
 
        void subDBTick()
        {
            using namespace std::chrono_literals;
 
            const std::scoped_lock lock(m_mutex);
 
            SPSP_LOGD("SubDB: Tick running");
 
            m_subDB.forEach([this](const std::string& topic, const SubDBEntry& entry) {
                m_subDB[topic].lifetime -= m_conf.subDB.interval;
 
                if (entry.lifetime <= 0ms) {
                    SPSP_LOGD("SubDB: Topic '%s' expired (renewing)", topic.c_str());
 
                    bool extended = this->sendSubscribe(topic);
                    if (extended) {
                        m_subDB[topic].lifetime = m_conf.subDB.subLifetime;
                    } else {
                        SPSP_LOGE("SubDB: Topic '%s' can't be extended. Will try again in next tick.",
                                  topic.c_str());
                    }
                }
            });
 
            SPSP_LOGD("SubDB: Tick done");
        }
    };
} // namespace SPSP::Nodes
 
#undef SPSP_LOG_TAG