If.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_IF_HXX_
#define _OPENLCB_IF_HXX_
 
#include <string>
 
#include "executor/Dispatcher.hxx"
#include "executor/Executor.hxx"
#include "executor/Service.hxx"
#include "openlcb/Convert.hxx"
#include "openlcb/Defs.hxx"
#include "openlcb/Node.hxx"
#include "utils/Buffer.hxx"
#include "utils/Map.hxx"
#include "utils/Queue.hxx"
 
namespace openlcb
{
 
class Node;
class StreamTransport;
 
extern void send_event(Node* src_node, uint64_t event_id);
 
struct GenMessage
{
    GenMessage()
        : src({0, 0}), dst({0, 0}), flagsSrc(0), flagsDst(0) {}
 
    void clear()
    {
        reset((Defs::MTI)0, 0, EMPTY_PAYLOAD);
    }
 
    void reset(Defs::MTI mti, NodeID src, NodeHandle dst, string payload)
    {
        this->mti = mti;
        this->src = {src, 0};
        this->dst = dst;
        this->payload = std::move(payload);
        this->dstNode = nullptr;
        this->flagsSrc = 0;
        this->flagsDst = 0;
    }
 
    void reset(Defs::MTI mti, NodeID src, string payload)
    {
        this->mti = mti;
        this->src = {src, 0};
        this->dst = {0, 0};
        this->payload = std::move(payload);
        this->dstNode = nullptr;
        this->flagsSrc = 0;
        this->flagsDst = 0;
    }
 
    NodeHandle src;
    NodeHandle dst;
    Defs::MTI mti;
    Node *dstNode;
    string payload;
 
    unsigned flagsSrc : 4;
    unsigned flagsDst : 4;
    unsigned get_flags_src() {
        return flagsSrc;
    }
    unsigned get_flags_dst() {
        return flagsDst;
    }
    void set_flag_src(unsigned flags) {
        flagsSrc |= flags;
    }
    void clear_flag_src(unsigned flags) {
        flagsSrc &= ~flags;
    }
    bool has_flag_src(unsigned flags) {
        return ((flagsSrc & flags) == flags);
    }
    void set_flag_dst(unsigned flags) {
        flagsDst |= flags;
    }
    void clear_flag_dst(unsigned flags) {
        flagsDst &= ~flags;
    }
    bool has_flag_dst(unsigned flags) {
        return ((flagsDst & flags) == flags);
    }
 
    typedef uint32_t id_type;
    id_type id() const
    {
        return static_cast<uint32_t>(mti);
    }
 
    unsigned priority()
    {
        return Defs::mti_priority(mti);
    }
 
    enum DstFlags {
        WAIT_FOR_LOCAL_LOOPBACK = 1,
        DSTFLAG_NOT_FIRST_MESSAGE = 2,
        DSTFLAG_NOT_LAST_MESSAGE  = 4,
        // 8: free
    };
    enum SrcFlags {
        // 1, 2, 4, 8: free
    };
};
 
typedef FlowInterface<Buffer<GenMessage>> MessageHandler;
 
class If : public Service
{
public:
    If(ExecutorBase *executor, int local_nodes_count);
 
    virtual ~If()
    {
    }
 
    MessageHandler *global_message_write_flow()
    {
        HASSERT(globalWriteFlow_);
        if (txHook_)
        {
            txHook_();
        }
        return globalWriteFlow_;
    }
    MessageHandler *addressed_message_write_flow()
    {
        HASSERT(addressedWriteFlow_);
        if (txHook_)
        {
            txHook_();
        }
        return addressedWriteFlow_;
    }
 
    typedef DispatchFlow<Buffer<GenMessage>, 4> MessageDispatchFlow;
 
    MessageDispatchFlow *dispatcher()
    {
        return &dispatcher_;
    }
 
    virtual void add_owned_flow(Executable *e) = 0;
 
    void add_local_node(Node *node)
    {
        NodeID id = node->node_id();
        HASSERT(localNodes_.find(id) == localNodes_.end());
        localNodes_[id] = node;
    }
 
    virtual void delete_local_node(Node *node) = 0;
 
    Node *lookup_local_node(NodeID id)
    {
        auto it = localNodes_.find(id);
        if (it == localNodes_.end())
        {
            return nullptr;
        }
        return it->second;
    }
 
    virtual Node *lookup_local_node_handle(NodeHandle handle)
    {
        return lookup_local_node(handle.id);
    }
 
    Node* first_local_node() {
        auto it = localNodes_.begin();
        if (it == localNodes_.end()) return nullptr;
        return it->second;
    }
 
    Node* next_local_node(NodeID previous) {
        auto it = localNodes_.find(previous);
        if (it == localNodes_.end())
        {
            return nullptr;
        }
        ++it;
        if (it == localNodes_.end())
        {
            return nullptr;
        }
        return it->second;
    }
 
    virtual bool matching_node(NodeHandle expected,
                               NodeHandle actual) = 0;
 
 
    virtual void canonicalize_handle(NodeHandle *h) {}
 
    virtual NodeID get_default_node_id() = 0;
 
    void set_tx_hook(std::function<void()> hook)
    {
        txHook_ = std::move(hook);
    }
 
    StreamTransport *stream_transport()
    {
        return streamTransport_;
    }
 
    void set_stream_transport(StreamTransport *s)
    {
        HASSERT(streamTransport_ == nullptr);
        streamTransport_ = s;
    }
 
protected:
    void remove_local_node_from_map(Node *node)
    {
        auto it = localNodes_.find(node->node_id());
        HASSERT(it != localNodes_.end());
        localNodes_.erase(it);
    }
 
    MessageHandler *globalWriteFlow_;
    MessageHandler *addressedWriteFlow_;
 
private:
    MessageDispatchFlow dispatcher_;
 
    std::function<void()> txHook_;
 
    typedef Map<NodeID, Node *> VNodeMap;
 
    VNodeMap localNodes_;
 
    StreamTransport *streamTransport_ {nullptr};
 
    friend class VerifyNodeIdHandler;
 
    DISALLOW_COPY_AND_ASSIGN(If);
};
 
typedef StateFlow<Buffer<GenMessage>, QList<4>> MessageStateFlowBase;
 
class IncomingMessageStateFlow
    : public MessageStateFlowBase
{
public:
    IncomingMessageStateFlow(If *iface)
        : MessageStateFlowBase(iface)
    {
    }
 
    If *iface()
    {
        return static_cast<If *>(service());
    }
 
    GenMessage *nmsg()
    {
        return message()->data();
    }
};
 
template <typename... Args> void send_message(Node *src_node, Defs::MTI mti, Args &&...args)
{
    Buffer<GenMessage> *msg;
    mainBufferPool->alloc(&msg);
    msg->data()->reset(mti, src_node->node_id(), std::forward<Args>(args)...);
    if (msg->data()->dst == NodeHandle())
    {
        src_node->iface()->global_message_write_flow()->send(msg);
    }
    else
    {
        src_node->iface()->addressed_message_write_flow()->send(msg);
    }
}
 
} // namespace openlcb
 
#endif // _OPENLCB_IF_HXX_