324 lines
11 KiB
C++
324 lines
11 KiB
C++
#include "wrap-string.hpp"
|
|
#include "wrap-vector.hpp"
|
|
|
|
#include "world.hpp"
|
|
#include "drivenengine.hpp"
|
|
#include "luaconsole.hpp"
|
|
#include "util.hpp"
|
|
#include "printbuffer.hpp"
|
|
|
|
#include <memory>
|
|
|
|
class Client : public eng::opnew {
|
|
public:
|
|
int64_t actor_id_;
|
|
SharedChannel channel_;
|
|
UniqueWorld sync_;
|
|
double last_diff_;
|
|
bool async_diff_;
|
|
};
|
|
using UniqueClient = std::unique_ptr<Client>;
|
|
using ClientVector = eng::vector<UniqueClient>;
|
|
|
|
class LpxServer : public DrivenEngine, public CommonCommands {
|
|
public:
|
|
using StringVec = LuaConsole::StringVec;
|
|
UniqueWorld master_;
|
|
LuaConsole console_;
|
|
ClientVector clients_;
|
|
PrintChanneler print_channeler_;
|
|
HttpChannelMap http_client_channels_;
|
|
HttpChannelVec http_server_channels_;
|
|
eng::vector<Invocation> delayed_invocations_;
|
|
int64_t admin_id_ = 0;
|
|
double next_tick_ = 0;
|
|
|
|
public:
|
|
LpxServer()
|
|
{
|
|
// Create the master world model.
|
|
master_.reset(new World(WORLD_TYPE_MASTER));
|
|
|
|
// Create the admin actor. Note: there isn't any 'init' function yet.
|
|
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 (client connections)
|
|
listen_port(8085);
|
|
|
|
// Enable listening on port 8080 (http server connections)
|
|
listen_port(8080);
|
|
|
|
// Set the console prompt.
|
|
set_console_prompt(console_.get_prompt());
|
|
|
|
// Export stuff to the graphics engine.
|
|
set_visible_world_and_actor(master_.get(), admin_id_);
|
|
|
|
// Trigger the loading of the lua source.
|
|
rescan_lua_source();
|
|
}
|
|
|
|
virtual void do_syntax_error(std::string_view error) override {
|
|
stdostream() << "Syntax error: " << error << std::endl;
|
|
}
|
|
|
|
virtual void do_unknown_command(std::string_view name) override {
|
|
stdostream() << "Unknown command: " << name << std::endl;
|
|
}
|
|
|
|
virtual void do_view_command() override {
|
|
stdostream() << master_->tangibles_near_debug_string(admin_id_, 1000);
|
|
}
|
|
|
|
virtual void do_moveto_command(int x, int y) override {
|
|
do_unknown_command("moveto");
|
|
}
|
|
|
|
virtual void do_quit_command() override {
|
|
stop_driver();
|
|
}
|
|
|
|
virtual void do_cpl_command() override {
|
|
rescan_lua_source();
|
|
}
|
|
|
|
virtual void do_work_command() override {
|
|
do_unknown_command("work");
|
|
}
|
|
|
|
virtual void do_display_command() override {
|
|
do_unknown_command("display");
|
|
}
|
|
|
|
virtual void do_aborthttp_command() override {
|
|
master_->abort_all_http_requests(425, "http requests aborted from the server command line");
|
|
}
|
|
|
|
virtual void do_connect_command(std::string_view hostname) override {
|
|
do_unknown_command("connect");
|
|
}
|
|
|
|
virtual void do_luainvoke_command(std::string_view cmd) override {
|
|
master_->invoke(Invocation(AccessKind::INVOKE_LUA_EXPR, admin_id_, admin_id_, cmd));
|
|
}
|
|
|
|
virtual void do_luaprobe_command(std::string_view cmd) override {
|
|
master_->snapshot();
|
|
stdostream() << master_->probe_lua_expr(admin_id_, cmd);
|
|
master_->rollback();
|
|
}
|
|
|
|
void delete_client(UniqueClient &client) {
|
|
stdostream() << "Client closed: actor id=" << client->actor_id_ << std::endl;
|
|
master_->disconnected(client->actor_id_);
|
|
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 StreamEofOnRead &seof) {
|
|
sb->unread_to(tr_before);
|
|
return false;
|
|
} catch (const StreamException &sexcept) {
|
|
delete_client(client);
|
|
return false;
|
|
}
|
|
// Security check.
|
|
if (inv.actor() != client->actor_id_) {
|
|
delete_client(client);
|
|
return false;
|
|
}
|
|
//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);
|
|
client->async_diff_ = true;
|
|
return true;
|
|
}
|
|
|
|
virtual void event_access(AccessKind kind, int64_t place_id, std::string_view datapk, StreamBuffer *retpk) override {
|
|
if (place_id == 0) place_id = admin_id_;
|
|
switch (kind) {
|
|
case AccessKind::INVOKE_LUA_CALL:
|
|
case AccessKind::INVOKE_LUA_EXPR:
|
|
case AccessKind::INVOKE_FLUSH_PRINTS:
|
|
case AccessKind::INVOKE_TICK:
|
|
case AccessKind::INVOKE_LUA_SOURCE: {
|
|
delayed_invocations_.emplace_back(kind, admin_id_, place_id, datapk);
|
|
break;
|
|
}
|
|
case AccessKind::PROBE_LUA_CALL: {
|
|
master_->snapshot();
|
|
master_->probe_lua_call(admin_id_, place_id, datapk, retpk);
|
|
master_->rollback();
|
|
break;
|
|
}
|
|
default: {
|
|
stdostream() << "Invalid event_access: " << int(kind) << std::endl;
|
|
}
|
|
}
|
|
}
|
|
|
|
virtual void event_update() override {
|
|
// Get the clock.
|
|
double clock = get_clock();
|
|
|
|
// Execute any queued invocations.
|
|
// We just feed these directly into the master model.
|
|
for (const Invocation &inv : delayed_invocations_) {
|
|
master_->invoke(inv);
|
|
}
|
|
delayed_invocations_.clear();
|
|
|
|
// Check for keyboard input on stdin.
|
|
while (true) {
|
|
eng::string line = get_stdio_channel()->in()->readline();
|
|
if (line == "") break;
|
|
console_.add(line);
|
|
set_console_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;
|
|
if (chan->port() == 8085) {
|
|
Client *client = new Client;
|
|
client->actor_id_ = master_->create_login_actor();
|
|
// TODO: initialize the login actor on the master.
|
|
client->channel_ = std::move(chan);
|
|
client->async_diff_ = true;
|
|
client->last_diff_ = -100.0;
|
|
client->sync_.reset(new World(WORLD_TYPE_PREDICTIVE));
|
|
// This login actor is never used, it is just to preserve the invariant that
|
|
// the client model and the server synchronous model are identical.
|
|
client->sync_->create_login_actor();
|
|
clients_.emplace_back(client);
|
|
stdostream() << "New client: actor id=" << client->actor_id_ << std::endl;
|
|
} else if (chan->port() == 8080) {
|
|
HttpChannel htchan;
|
|
htchan.channel_ = chan;
|
|
http_server_channels_.push_back(htchan);
|
|
}
|
|
}
|
|
|
|
// If the clock has advanced far enough, tick the master model.
|
|
if (clock >= next_tick_) {
|
|
master_->invoke(Invocation(AccessKind::INVOKE_TICK, 0, 0, ""));
|
|
for (UniqueClient &client : clients_) {
|
|
client->async_diff_ = true;
|
|
}
|
|
next_tick_ += 1.0;
|
|
if (next_tick_ < clock + 0.3) next_tick_ = clock + 0.3;
|
|
}
|
|
|
|
// 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));
|
|
if (client == nullptr) continue;
|
|
|
|
// Possibly send a diff.
|
|
double diffdelay = 5.0;
|
|
if (client->async_diff_) diffdelay = 0.1;
|
|
if (clock >= client->last_diff_ + diffdelay) {
|
|
send_diffs(client);
|
|
client->last_diff_ = clock;
|
|
client->async_diff_ = false;
|
|
}
|
|
}
|
|
util::remove_nullptrs(clients_);
|
|
|
|
// Look for new outgoing HTTP client requests.
|
|
for (const auto &pair : master_->http_requests()) {
|
|
const HttpClientRequest &request = pair.second;
|
|
HttpChannel &channel = http_client_channels_[request.request_id()];
|
|
if (channel.channel_ == nullptr) {
|
|
channel.channel_ = new_outgoing_channel(request.target());
|
|
channel.method_ = request.method();
|
|
channel.parsed_bytes_ = 0;
|
|
request.send(channel.channel_->out());
|
|
}
|
|
}
|
|
|
|
// Maintain existing outgoing HTTP client requests.
|
|
HttpParserVec http_responses;
|
|
for (auto &pair : http_client_channels_) {
|
|
HttpChannel &htchan = pair.second;
|
|
Channel &channel = *htchan.channel_;
|
|
if (channel.closed() || (channel.in()->fill() > htchan.parsed_bytes_)) {
|
|
HttpParser response;
|
|
if (!channel.error().empty()) {
|
|
response.fail(503, util::ss("Service Unavailable: ", channel.error()));
|
|
} else {
|
|
htchan.parsed_bytes_ = channel.in()->fill();
|
|
response.parse_response(channel.in()->view(), channel.closed(), htchan.method_);
|
|
}
|
|
if (response.complete()) {
|
|
response.set_request_id(pair.first);
|
|
http_responses.push_back(response);
|
|
}
|
|
}
|
|
}
|
|
for (const HttpParser &response : http_responses) {
|
|
http_client_channels_.erase(response.request_id());
|
|
}
|
|
master_->http_responses(http_responses);
|
|
|
|
// Maintain incoming HTTP server channels.
|
|
for (HttpChannel &htchan : http_server_channels_) {
|
|
SharedChannel &chan = htchan.channel_;
|
|
if (chan->in()->fill() > htchan.parsed_bytes_) {
|
|
HttpParser parser;
|
|
parser.parse_request(chan->in()->view(), chan->closed());
|
|
htchan.parsed_bytes_ = chan->in()->fill();
|
|
if (parser.complete()) {
|
|
StreamBuffer *sb = chan->out();
|
|
HttpServerResponse resp = master_->http_serve(parser);
|
|
resp.send(sb);
|
|
htchan.channel_ = nullptr;
|
|
}
|
|
}
|
|
}
|
|
util::remove_marked_items(http_server_channels_);
|
|
}
|
|
};
|
|
|
|
DrivenEngineDefine("lpxserver", LpxServer);
|
|
|