integration/luprex/core/cpp/drivenengine.cpp

#include "drivenengine.hpp"


Channel::Channel(DrivenEngine *de, int chid, int port, const std::string &target) {
    driven_ = de;
    chid_ = chid;
    sb_in_.reset(new StreamBuffer);
    sb_out_.reset(new StreamBuffer);
    port_ = port;
    closed_ = false;
    target_ = target;
    readline_enabled_ = (chid == 0);
    readline_len_ = 0;
    readline_lastc_ = 0;
    echo_len_ = 0;
    assert(driven_->channels_[chid_] == nullptr);
    driven_->channels_[chid_] = this;
}

Channel::~Channel() {
    assert(driven_->channels_[chid_] == this);
    driven_->new_closed_.insert(chid_);
    driven_->new_outgoing_.erase(chid_);
    driven_->channels_[chid_] = nullptr;
}


void Channel::feed_readline(int nbytes, const char *bytes) {
    for (int i = 0; i < nbytes; i++) {
        char c = bytes[i];
        if ((c == '\n') && (readline_lastc_ == '\r')) {
            // Ignore newline immediately after carriage return.
            // Otherwise, crlf produces two newlines.
        } else if ((c == '\r') || (c == '\n')) {
            if ((echo_space() >= 3) && (readline_space() >= 1)) {
                echo_buf_[echo_len_++] = ' ';
                echo_buf_[echo_len_++] = '\r';
                echo_buf_[echo_len_++] = '\n';
                readline_buf_[readline_len_++] = '\n';
                sb_in_->write_bytes(readline_buf_, readline_len_);
                readline_len_ = 0;
            }
        } else if (c == '\b') {
            if ((readline_len_ >= 1) && (echo_space() >= 3)) {
                echo_buf_[echo_len_++] = '\b';
                echo_buf_[echo_len_++] = ' ';
                echo_buf_[echo_len_++] = '\b';
                readline_len_ -= 1;
            }
        } else if (c >= 32) {
            // Don't use up the last character in the readline buffer: save
            // it for the newline.
            if ((readline_space() >= 2) && (echo_space() >= 1)) {
                echo_buf_[echo_len_++] = c;
                readline_buf_[readline_len_++] = c;
            }
        }
        readline_lastc_ = c;
    }
}

void Channel::set_readline(bool e) {
    if (e != readline_enabled_) {
        readline_enabled_ = e;
        readline_len_ = 0;
        readline_lastc_ = 0;
        echo_len_ = 0;
    }
}

int DrivenEngine::find_unused_chid() {
    // Note: channel ID zero is special, it is never reused.
    for (int i = 0; i < MAX_CHAN; i++) {
        int id = next_unused_chid_++;
        if (next_unused_chid_ == MAX_CHAN) next_unused_chid_ = 1;
        if (channels_[id] == nullptr) return id;
    }
    assert(false);
    return 0;
}

Channel *DrivenEngine::get_chid(int chid) {
    assert(unsigned(chid) < MAX_CHAN);
    assert(channels_[chid] != nullptr);
    return channels_[chid];
}

void DrivenEngine::listen_port(int port) {
    listen_ports_.insert(port);
}

double DrivenEngine::get_clock() {
    return clock_;
}

UniqueChannel DrivenEngine::new_outgoing_channel(const std::string &target) {
    int chid = find_unused_chid();
    new_outgoing_.insert(chid);
    return UniqueChannel(new Channel(this, chid, 0, target));
}

UniqueChannel DrivenEngine::new_incoming_channel() {
    if (accepted_channels_.empty()) {
        return nullptr;
    } else {
        UniqueChannel result = std::move(accepted_channels_.back());
        accepted_channels_.pop_back();
        return std::move(result);
    }
}

Channel *DrivenEngine::get_stdio_channel() {
    return stdio_channel_.get();
}

util::LuaSourcePtr DrivenEngine::get_lua_source() {
    return std::move(lua_source_);
}

void DrivenEngine::rescan_lua_source() {
    rescan_lua_source_ = true;
}

void DrivenEngine::stop_driver() {
    stop_driver_ = true;
}

void DrivenEngine::drv_get_listen_ports(std::set<int> &ports) {
    ports = listen_ports_;
}

void DrivenEngine::drv_get_new_closed(std::set<int> &channels) {
    channels = std::move(new_closed_);
    new_closed_.clear();
}

void DrivenEngine::drv_get_new_outgoing(std::set<int> &channels) {
    channels = std::move(new_outgoing_);
    new_outgoing_.clear();
}

const std::string &DrivenEngine::drv_get_target(int chid) {
    return get_chid(chid)->target_;
}

bool DrivenEngine::drv_outgoing_empty(int chid) {
    return get_chid(chid)->sb_out_->empty();
}

void DrivenEngine::drv_peek_outgoing(int chid, int *nbytes, const char **bytes) {
    Channel *ch = get_chid(chid);
    if (ch->echo_len_ > 0) {
        *nbytes = ch->echo_len_;
        *bytes = ch->echo_buf_;
    } else {
        *nbytes = ch->sb_out_->fill();
        *bytes = ch->sb_out_->data();
    }
}

void DrivenEngine::drv_sent_outgoing(int chid, int nbytes) {
    Channel *ch = get_chid(chid);
    if (nbytes > 0) {
        if (ch->echo_len_ > 0) {
            if (nbytes >= ch->echo_len_) {
                ch->sb_out_->read_bytes(nbytes - ch->echo_len_);
                ch->echo_len_ = 0;
            } else {
                ch->echo_len_ -= nbytes;
                memmove(ch->echo_buf_, ch->echo_buf_ + nbytes, ch->echo_len_);
            }
        } else {
            ch->sb_out_->read_bytes(nbytes);
        }
    }
}

void DrivenEngine::drv_recv_incoming(int chid, int nbytes, const char *bytes) {
    if (nbytes > 0) {
        Channel *ch = get_chid(chid);
        if (ch->readline_enabled_) {
            ch->feed_readline(nbytes, bytes);
        } else {
            ch->sb_in_->write_bytes(bytes, nbytes);
        }
    }
}

void DrivenEngine::drv_notify_close(int chid) {
    get_chid(chid)->closed_ = true;
}

int DrivenEngine::drv_notify_accept(int port) {
    int chid = find_unused_chid();
    accepted_channels_.emplace_back(new Channel(this, chid, port, ""));
    return chid;
}

void DrivenEngine::drv_set_clock(double t) {
    clock_ = t;
}

void DrivenEngine::drv_set_lua_source(util::LuaSourcePtr source) {
    lua_source_ = std::move(source);
    rescan_lua_source_ = false;
}

void DrivenEngine::drv_invoke_event_init() {
    event_init();
}

void DrivenEngine::drv_invoke_event_update() {
    event_update();
}

bool DrivenEngine::drv_get_rescan_lua_source() {
    return rescan_lua_source_;
}

bool DrivenEngine::drv_get_stop_driver() {
    return stop_driver_;
}

DrivenEngine::DrivenEngine() {
    for (int i = 0; i < MAX_CHAN; i++) {
        channels_[i] = nullptr;
    }
    next_unused_chid_ = 1;
    stdio_channel_.reset(new Channel(this, 0, 0, ""));
    rescan_lua_source_ = true;
    clock_ = 0.0;
    stop_driver_ = false;
}

DrivenEngine::~DrivenEngine() {
    // Delete the channels that we own.
    stdio_channel_.reset();
    accepted_channels_.clear();
    // At this point, all channels should be gone.
    for (int i = 0; i < MAX_CHAN; i++) {
        assert(channels_[i] == nullptr);
    }
}

static DrivenEngine *engine_;

void DrivenEngine::set(DrivenEngine *de) {
    engine_ = de;
}

DrivenEngine *DrivenEngine::get() {
    return engine_;
}