DirectHubLegacy.cxx

Go to the documentation of this file.

 
#include "utils/DirectHub.hxx"
#include "utils/Hub.hxx"
#include "utils/gc_format.h"
 
extern DataBufferPool g_direct_hub_kbyte_pool;
 
class HubToGcPort : public CanHubPort, public DirectHubPort<uint8_t[]>
{
public:
    HubToGcPort(DirectHubInterface<uint8_t[]> *gc_hub, CanHubFlow *can_hub)
        : CanHubPort(gc_hub->get_service())
        , targetHub_(gc_hub)
        , sourceHub_(can_hub)
    {
        sourceHub_->register_port(this);
        targetHub_->register_port(this);
    }
 
    ~HubToGcPort()
    {
        sourceHub_->unregister_port(this);
        targetHub_->unregister_port(this);
    }
 
    Action entry() override
    {
        // Allocates output buffer if needed.
        if (buf_.free() < MIN_GC_FREE)
        {
            // Need more output buffer.
            DataBuffer *b;
            g_direct_hub_kbyte_pool.alloc(&b);
            buf_.append_empty_buffer(b);
        }
        // Generates gridconnect message and commits to buffer.
        char *start = (char *)buf_.data_write_pointer();
        char *end = gc_format_generate(message()->data(), start, 0);
        packetSize_ = end - start;
        buf_.data_write_advance(packetSize_);
        pktDone_ = message()->new_child();
        release();
        // Sends off output message.
        wait_and_call(STATE(do_send));
        inlineRun_ = true;
        inlineComplete_ = false;
        targetHub_->enqueue_send(this);
        inlineRun_ = false;
        if (inlineComplete_)
        {
            return exit();
        }
        else
        {
            return wait();
        }
    }
 
    Action do_send()
    {
        auto *m = targetHub_->mutable_message();
        m->buf_ = buf_.transfer_head(packetSize_);
        m->source_ = (DirectHubPort<uint8_t[]> *)this;
        m->done_ = pktDone_;
        targetHub_->do_send();
        if (inlineRun_)
        {
            inlineComplete_ = true;
            return wait();
        }
        else
        {
            return exit();
        }
    }
 
    void send(MessageAccessor<uint8_t[]> *msg) override
    {
        auto &buf = msg->buf_;
        uint8_t *p;
        unsigned available;
        buf.head()->get_read_pointer(buf.skip(), &p, &available);
        if (buf.size() == 0 || *p != ':')
        {
            // Not a gridconnect packet. Do not do anything.
            return;
        }
        Buffer<CanHubData> *can_buf = sourceHub_->alloc();
        if (msg->done_)
        {
            can_buf->set_done(msg->done_->new_child());
        }
        can_buf->data()->skipMember_ = (CanHubPort *)this;
        const char *text_packet = nullptr;
        string assembled_packet;
        if (available == buf.size())
        {
            // One block of data. Convert in place.
            text_packet = (const char *)p;
        }
        else
        {
            buf.append_to(&assembled_packet);
            text_packet = assembled_packet.c_str();
        }
        if (gc_format_parse(text_packet, can_buf->data()) < 0)
        {
            string debug(text_packet, buf.size());
            LOG(INFO, "Failed to parse gridconnect packet: '%s'",
                debug.c_str());
            can_buf->unref();
            return;
        }
        sourceHub_->send(can_buf, 0);
    }
 
private:
    LinkedDataBufferPtr buf_;
    DirectHubInterface<uint8_t[]> *targetHub_;
    BarrierNotifiable *pktDone_ = nullptr;
    CanHubFlow *sourceHub_;
    HubSource *me_;
    bool inlineRun_ : 1;
    bool inlineComplete_ : 1;
    uint16_t packetSize_;
    static constexpr unsigned MIN_GC_FREE = 29;
};
 
Destructable *create_gc_to_legacy_can_bridge(
    DirectHubInterface<uint8_t[]> *gc_hub, CanHubFlow *can_hub)
{
 
    return new HubToGcPort(gc_hub, can_hub);
}