WriteHelper.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_WRITEHELPER_HXX_
#define _OPENLCB_WRITEHELPER_HXX_
 
#include <string>
 
#include "nmranet_config.h"
 
#include "openlcb/If.hxx"
#include "openlcb/Node.hxx"
 
namespace openlcb
{
 
class WriteHelper : public Executable
{
public:
    typedef Node *node_type;
    typedef string payload_type;
 
    static NodeHandle global()
    {
        return {0, 0};
    }
 
    WriteHelper()
        : waitForLocalLoopback_(0)
    {
    }
 
    const payload_type &last_payload()
    {
        return buffer_;
    }
 
    void clear_last_payload()
    {
        buffer_.clear();
    }
 
    void set_wait_for_local_loopback(bool wait = true)
    {
        waitForLocalLoopback_ = (wait ? 1 : 0);
    }
 
    void WriteAsync(Node *node, Defs::MTI mti, NodeHandle dst,
                    const payload_type &buffer, Notifiable *done)
    {
        if (done)
        {
            done_.reset(done);
        }
        else
        {
            // We don't support synchronous sending anymore.
            HASSERT(0);
        }
        if (!node ||
            (!node->is_initialized() && mti != Defs::MTI_INITIALIZATION_COMPLETE))
        {
            done_.notify();
            return;
        }
        node_ = node;
        mti_ = mti;
        dst_ = dst;
        buffer_ = buffer;
        if (dst == global())
        {
            node->iface()->global_message_write_flow()->alloc_async(this);
        }
        else
        {
            node->iface()->addressed_message_write_flow()->alloc_async(
                this);
        }
    }
 
private:
    // Callback from the allocator.
    void alloc_result(QMember *entry) override
    {
        /* NOTE(balazs.racz): We could choose not to pass on the done_
         * callback. That will allow the current write flow to be released
         * earlier for reuse, but breaks the assumption that done means that
         * the current packet is enqueued on the physical layer. */
        if (dst_ == global())
        {
            auto *f = node_->iface()->global_message_write_flow();
            Buffer<GenMessage> *b = f->cast_alloc(entry);
            b->data()->reset(mti_, node_->node_id(), buffer_);
            if (waitForLocalLoopback_)
            {
                b->data()->set_flag_dst(
                    GenMessage::WAIT_FOR_LOCAL_LOOPBACK);
            }
            b->set_done(&done_);
            f->send(b, b->data()->priority());
        }
        else
        {
            auto *f = node_->iface()->addressed_message_write_flow();
            auto *b = f->cast_alloc(entry);
            b->data()->reset(mti_, node_->node_id(), dst_, buffer_);
            if (waitForLocalLoopback_)
            {
                b->data()->set_flag_dst(
                    GenMessage::WAIT_FOR_LOCAL_LOOPBACK);
            }
            b->set_done(&done_);
            f->send(b, b->data()->priority());
        }
    }
 
    void run() override
    {
        HASSERT(0);
    }
 
    unsigned waitForLocalLoopback_ : 1;
    NodeHandle dst_;
    Defs::MTI mti_;
    Node *node_;
    payload_type buffer_;
    BarrierNotifiable done_;
};
 
}; /* namespace openlcb */
 
#endif // _OPENLCB_WRITEHELPER_HXX_