DirectHub.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_DIRECTHUB_HXX_
#define _UTILS_DIRECTHUB_HXX_
 
#include "executor/Executor.hxx"
#include "utils/DataBuffer.hxx"
 
class Service;
 
class HubSource
{ };
 
struct MessageMetadata
{
    void clear()
    {
        if (done_)
        {
            done_->notify();
            done_ = nullptr;
        }
        source_ = dst_ = nullptr;
        isFlush_ = false;
    }
 
    void set_done(BarrierNotifiable *done)
    {
        if (done_)
        {
            done_->notify();
            done_ = nullptr;
        }
        done_ = done;
    }
 
    BarrierNotifiable *done_ = nullptr;
    HubSource *source_ = nullptr;
    HubSource *dst_ = nullptr;
    bool isFlush_ = false;
};
 
template <class T> struct MessageAccessor : public MessageMetadata
{
public:
    void clear()
    {
        payload_.reset();
        MessageMetadata::clear();
    }
 
    BufferPtr<T> payload_;
};
 
template <> struct MessageAccessor<uint8_t[]> : public MessageMetadata
{
    void clear()
    {
        // Walks the buffer links and unrefs everything we own.
        buf_.reset();
        MessageMetadata::clear();
    }
    LinkedDataBufferPtr buf_;
};
 
class MessageSegmenter : public Destructable
{
public:
    virtual ssize_t segment_message(const void *data, size_t size) = 0;
 
    virtual void clear() = 0;
};
 
template <class T> class DirectHubPort : public HubSource
{
public:
    virtual void send(MessageAccessor<T> *msg) = 0;
};
 
template <class T> class DirectHubInterface : public Destructable
{
public:
    virtual Service *get_service() = 0;
 
    virtual void register_port(DirectHubPort<T> *port) = 0;
    virtual void unregister_port(DirectHubPort<T> *port) = 0;
    virtual void unregister_port(DirectHubPort<T> *port, Notifiable *done) = 0;
 
    virtual void enqueue_send(Executable *caller) = 0;
 
    virtual MessageAccessor<T> *mutable_message() = 0;
 
    virtual void do_send() = 0;
};
 
typedef DirectHubInterface<uint8_t[]> ByteDirectHubInterface;
 
ByteDirectHubInterface *create_hub(ExecutorBase *e);
 
void create_port_for_fd(ByteDirectHubInterface *hub, int fd,
    std::unique_ptr<MessageSegmenter> segmenter,
    Notifiable *on_error = nullptr);
 
void create_direct_gc_tcp_hub(ByteDirectHubInterface *hub, int port);
 
MessageSegmenter *create_gc_message_segmenter();
 
MessageSegmenter *create_trivial_message_segmenter();
 
// Forward declarations to avoid needing to include Hub.hxx here.
template <class T> class GenericHubFlow;
template <class T> class HubContainer;
struct CanFrameContainer;
typedef HubContainer<CanFrameContainer> CanHubData;
typedef GenericHubFlow<CanHubData> CanHubFlow;
 
Destructable *create_gc_to_legacy_can_bridge(
    DirectHubInterface<uint8_t[]> *gc_hub, CanHubFlow *can_hub);
 
#endif // _UTILS_DIRECTHUB_HXX_