node.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <chrono>
#include <cinttypes>
#include <climits>
#include <functional>
#include <memory>
#include <mutex>
#include <thread>
 
#include "spsp/layers.hpp"
#include "spsp/local_message.hpp"
#include "spsp/logger.hpp"
#include "spsp/version.hpp"
 
// Log tag
#define SPSP_LOG_TAG "SPSP/Node"
 
namespace SPSP
{
    // Topics for reporting
    static const std::string NODE_REPORTING_TOPIC = "_report";
    static const std::string NODE_REPORTING_RSSI_SUBTOPIC = "rssi";
    static const std::string NODE_REPORTING_PROBE_PAYLOAD_SUBTOPIC = "probe_payload";
 
    static const int NODE_RSSI_UNKNOWN = INT_MIN;  
 
    using SubscribeCb = std::function<void(const std::string& topic,
                                           const std::string& payload)>;
 
    class INode
    {
    public:
        INode()
        {
            SPSP_LOGI("SPSP version: %s", SPSP::VERSION);
        }
 
        virtual bool publish(const std::string& topic,
                             const std::string& payload) = 0;
 
        virtual bool subscribe(const std::string& topic, SubscribeCb cb) = 0;
 
        virtual bool unsubscribe(const std::string& topic) = 0;
 
        virtual void resubscribeAll() = 0;
    };
 
    template <typename TLocalLayer>
    class ILocalNode : virtual public INode
    {
        using LocalAddrT = typename TLocalLayer::LocalAddrT;
        using LocalMessageT = typename TLocalLayer::LocalMessageT;
        using LocalRecvSendCb = std::function<void(const LocalMessageT&)>;
 
        TLocalLayer* m_ll;
        LocalRecvSendCb m_localRecvSendCb = nullptr;
 
    public:
        ILocalNode(TLocalLayer* ll) : m_ll{ll}
        {
            m_ll->setNode(this);
        }
 
        void receiveLocal(const LocalMessageT& msg, int rssi = NODE_RSSI_UNKNOWN)
        {
            if (rssi != NODE_RSSI_UNKNOWN) {
                SPSP_LOGI("Received local msg: %s (%d dBm)",
                          msg.toString().c_str(), rssi);
            } else {
                SPSP_LOGI("Received local msg: %s", msg.toString().c_str());
            }
 
            // Call receive/send callback
            if (m_localRecvSendCb != nullptr) {
                SPSP_LOGD("Calling receive/send callback");
                m_localRecvSendCb(msg);
            }
 
            bool processed = false;
            auto processingTimeBegin = std::chrono::steady_clock::now();
 
            // Call responsible handler
            switch (msg.type) {
            case LocalMessageType::PROBE_REQ: processed = processProbeReq(msg, rssi); break;
            case LocalMessageType::PROBE_RES: processed = processProbeRes(msg, rssi); break;
            case LocalMessageType::PUB: processed = processPub(msg, rssi); break;
            case LocalMessageType::SUB_REQ: processed = processSubReq(msg, rssi); break;
            case LocalMessageType::SUB_DATA: processed = processSubData(msg, rssi); break;
            case LocalMessageType::UNSUB: processed = processUnsub(msg, rssi); break;
            case LocalMessageType::TIME_REQ: processed = processTimeReq(msg, rssi); break;
            case LocalMessageType::TIME_RES: processed = processTimeRes(msg, rssi); break;
            default:
                SPSP_LOGW("Unprocessable message type %s (%d)",
                        localMessageTypeToStr(msg.type),
                        static_cast<uint8_t>(msg.type));
                break;
            }
 
            auto processingTimeEnd = std::chrono::steady_clock::now();
            auto processingDuration = std::chrono::duration_cast
                <std::chrono::milliseconds>(processingTimeEnd
                                            - processingTimeBegin);
 
            if (processed) {
                SPSP_LOGD("Message processed (%" PRId64 " ms): %s",
                          processingDuration.count(), msg.toString().c_str());
            } else {
                SPSP_LOGW("Message not processed (%" PRId64 " ms): %s",
                          processingDuration.count(), msg.toString().c_str());
            }
        }
 
        void setLocalRecvSendCb(LocalRecvSendCb cb)
        {
            m_localRecvSendCb = cb;
        }
 
    protected:
        inline TLocalLayer* getLocalLayer() const
        {
            return m_ll;
        }
 
        bool sendLocal(const LocalMessageT& msg)
        {
            SPSP_LOGI("Sending local msg: %s", msg.toString().c_str());
 
            // Send to local layer
            bool delivered = m_ll->send(msg);
 
            if (delivered) {
                SPSP_LOGD("Message delivered: %s", msg.toString().c_str());
            } else {
                SPSP_LOGW("Message not delivered: %s", msg.toString().c_str());
            }
 
            // Call receive/send callback
            if (m_localRecvSendCb != nullptr) {
                SPSP_LOGD("Calling receive/send callback");
                m_localRecvSendCb(msg);
            }
 
            return delivered;
        }
 
        void publishRssi(const LocalAddrT& addr, int rssi)
        {
            if (rssi == NODE_RSSI_UNKNOWN) return;
 
            // Spawn new thread for this publish
            std::thread t([this, addr, rssi] {
                std::string topic = NODE_REPORTING_TOPIC + "/"
                                  + NODE_REPORTING_RSSI_SUBTOPIC + "/"
                                  + addr.str;
 
                this->publish(topic, std::to_string(rssi));
            });
            t.detach();
        }
 
        virtual bool processProbeReq(const LocalMessageT& req,
                                     int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processProbeRes(const LocalMessageT& req,
                                     int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processPub(const LocalMessageT& req,
                                int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processSubReq(const LocalMessageT& req,
                                   int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processSubData(const LocalMessageT& req,
                                    int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processUnsub(const LocalMessageT& req,
                                  int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processTimeReq(const LocalMessageT& req,
                                    int rssi = NODE_RSSI_UNKNOWN) = 0;
 
        virtual bool processTimeRes(const LocalMessageT& req,
                                    int rssi = NODE_RSSI_UNKNOWN) = 0;
    };
 
    template <typename TFarLayer>
    class IFarNode : virtual public INode
    {
        TFarLayer* m_fl;
 
    public:
        IFarNode(TFarLayer* fl) : m_fl{fl}
        {
            m_fl->setNode(this);
        }
 
        virtual bool receiveFar(const std::string& topic,
                                const std::string& payload) = 0;
 
    protected:
        inline TFarLayer* getFarLayer() const
        {
            return m_fl;
        }
    };
 
    template<typename TLocalLayer, typename TFarLayer>
    class ILocalAndFarNode : public ILocalNode<TLocalLayer>,
                             public IFarNode<TFarLayer>
    {
    public:
        ILocalAndFarNode(TLocalLayer *ll, TFarLayer *fl)
            : ILocalNode<TLocalLayer>(ll), IFarNode<TFarLayer>(fl) {}
    };
} // namespace SPSP
 
#undef SPSP_LOG_TAG