integration/luprex/core/cpp/lpxserver.cpp

#include "lpxserver.hpp"

#include "world.hpp"
#include "drivenengine.hpp"
#include "luaconsole.hpp"
#include "util.hpp"
#include "printbuffer.hpp"
#include <memory>

class Client {
public:
    int64_t actor_id_;
    SharedChannel channel_;
    UniqueWorld sync_;
};
using UniqueClient = std::unique_ptr<Client>;
using ClientVector = std::vector<UniqueClient>;

class LpxServer : public DrivenEngine {
public:
    UniqueWorld master_;
    LuaConsole console_;
    ClientVector clients_;
    PrintChanneler print_channeler_;
    int64_t admin_id_;

public:
    virtual void event_init(int argc, char *argv[]) {
        // Create the master world model.
        master_.reset(new World(util::WORLD_TYPE_MASTER));

        // Update the source code of the master model.
        master_->update_source(get_lua_source());

        // Create an actor for administrative commands.
        admin_id_ = master_->create_login_actor();

        // Print out admin ID for debugging purposes.
        stdostream() << "Admin actor id = " << admin_id_ << std::endl;

        // Enable listening on port 8085.
        listen_port(8085);

        // Set the console prompt.
        get_stdio_channel()->set_prompt(console_.get_prompt());
    }

    void do_luainvoke_command(const util::StringVec &words) {
        master_->invoke(Invocation(Invocation::KIND_LUA, admin_id_, admin_id_, words[1]));
    }

    void do_luaprobe_command(const util::StringVec &words) {
        master_->snapshot();
        stdostream() << master_->probe_lua(admin_id_, words[1]);;
        master_->rollback();
    }

    void do_syntax_command(const util::StringVec &words) {
        stdostream() << "Syntax Error: " << words[1] << std::endl;
    }

    void do_tick_command(const util::StringVec &words) {
        master_->invoke(Invocation(Invocation::KIND_TICK, admin_id_, admin_id_, ""));
    }

    void do_cpl_command(const util::StringVec &words) {
        rescan_lua_source();
    }

    void do_quit_command(const util::StringVec &words) {
        stop_driver();
    }

    void do_command(const util::StringVec &words) {
        if (words.empty()) return;
        else if (words[0] == "luainvoke") do_luainvoke_command(words);
        else if (words[0] == "luaprobe") do_luaprobe_command(words);
        else if (words[0] == "syntax") do_syntax_command(words);
        else if (words[0] == "tick") do_tick_command(words);
        else if (words[0] == "cpl") do_cpl_command(words);
        else if (words[0] == "quit") do_quit_command(words);
        else {
            stdostream() << "Unsupported command: " << words[0] << std::endl;
        }
    }

    void delete_client(UniqueClient &client) {
        stdostream() << "Client closed: actor id=" << client->actor_id_ << std::endl;
        client.reset();
    }

    void send_diffs(UniqueClient &client) {
        StreamBuffer *sb = client->channel_->out();
        sb->write_uint8(util::MSG_DIFF);
        sb->write_uint32(0);
        int64_t tw_1 = sb->total_writes();
        stdostream() << "Sending diffs to client " << client->actor_id_ << std::endl;
        client->sync_->diff_everything(client->actor_id_, master_.get(), sb);
        int64_t tw_2 = sb->total_writes();
        sb->overwrite_int32(tw_1, tw_2 - tw_1);
    }

    bool handle_invocation(UniqueClient &client) {
        StreamBuffer *sb = client->channel_->in();
        int64_t tr_before = sb->total_reads();
        Invocation inv;
        try {
            uint8_t msg_type = sb->read_uint8();
            if (msg_type != util::MSG_INVOKE) {
                delete_client(client);
                return false;
            }
            inv.deserialize(sb);
        } catch (const StreamEof &seof) {
            sb->unread_to(tr_before);
            return false;
        } catch (const StreamCorruption &scorr) {
            delete_client(client);
            return false;
        }
        if (inv.actor() != client->actor_id_) {
            stdostream() << "Ignoring invoke with wrong actor ID " << inv.actor() << std::endl;
            return true;
        }
        stdostream() << "Invoking: " << inv.debug_string() << std::endl;
        master_->invoke(inv);
        client->channel_->out()->write_uint8(util::MSG_ACK);
        client->channel_->out()->write_uint32(0);
        client->sync_->invoke(inv);
        send_diffs(client);
        return true;
    }
    
    virtual void event_update() {
        // If the driver has reloaded the source, put it into master model.
        master_->update_source(get_lua_source());

        // Check for keyboard input on stdin.
        while (true) {
            std::string line = get_stdio_channel()->in()->readline();
            if (line == "") break;
            console_.add(line);
            get_stdio_channel()->set_prompt(console_.get_prompt());
            do_command(console_.get_command());
        }

        // Anything in the administrator printbuffer should go to stdostream.
        if (print_channeler_.channel(master_->get_printbuffer(admin_id_), stdostream())) {
            master_->invoke(print_channeler_.invocation(admin_id_));
        }

        // Check for new incoming channels, set up client structures.
        while (true) {
            SharedChannel chan = new_incoming_channel();
            if (chan == nullptr) break;
            Client *client = new Client;
            client->actor_id_ = master_->create_login_actor();
            client->channel_ = std::move(chan);
            client->sync_.reset(new World(util::WORLD_TYPE_S_SYNC));
            client->sync_->create_login_actor();
            clients_.emplace_back(client);
            stdostream() << "New client: actor id=" << client->actor_id_ << std::endl;
            send_diffs(clients_.back());
        }

        // Traverse all existing channels, process any communication.
        for (UniqueClient &client : clients_) {
            if (client->channel_->closed()) {
                delete_client(client);
                continue;
            }
            
            // Check for received invocations.
            while (handle_invocation(client));
        }
        util::remove_nullptrs(clients_);
    }
};

UniqueDrivenEngine make_LpxServer() {
    return UniqueDrivenEngine(new LpxServer);
}