From 92fa52c56f9e0ae819e78ac2fb5034c4553d28c3 Mon Sep 17 00:00:00 2001 From: Lucien Greathouse Date: Sat, 26 May 2018 00:16:27 -0700 Subject: [PATCH] Implement long-polling message passing --- plugin/rojo.json | 4 ++++ plugin/src/ApiContext.lua | 45 ++++++++++++++++++++++++++++++------ plugin/src/Http.lua | 3 ++- plugin/src/Main.server.lua | 11 +++++++-- plugin/src/Session.lua | 7 ++++-- server/src/commands/serve.rs | 2 -- server/src/session.rs | 28 ++++++++++++++++++---- server/src/web.rs | 32 +++++++++---------------- 8 files changed, 93 insertions(+), 39 deletions(-) diff --git a/plugin/rojo.json b/plugin/rojo.json index c1cbdb8c..e184e52f 100644 --- a/plugin/rojo.json +++ b/plugin/rojo.json @@ -22,6 +22,10 @@ "path": "modules/testez/lib", "target": "ReplicatedStorage.TestEZ" }, + "modules/promise": { + "path": "modules/promise/lib", + "target": "ReplicatedStorage.Rojo.modules.Promise" + }, "tests": { "path": "tests", "target": "TestService" diff --git a/plugin/src/ApiContext.lua b/plugin/src/ApiContext.lua index 3c281ed0..5be7bae9 100644 --- a/plugin/src/ApiContext.lua +++ b/plugin/src/ApiContext.lua @@ -1,5 +1,6 @@ +local Promise = require(script.Parent.Parent.modules.Promise) + local Config = require(script.Parent.Config) -local Promise = require(script.Parent.Promise) local Version = require(script.Parent.Version) local Http = require(script.Parent.Http) @@ -24,7 +25,8 @@ function ApiContext.new(url, onMessage) url = url, onMessage = onMessage, serverId = nil, - currentTime = 0, + connected = false, + eventCursor = -1, } setmetatable(context, ApiContext) @@ -35,9 +37,9 @@ end function ApiContext:connect() return Http.get(self.url) :andThen(function(response) - response = response:json() + local body = response:json() - if response.protocolVersion ~= Config.protocolVersion then + if body.protocolVersion ~= Config.protocolVersion then local message = ( "Found a Rojo dev server, but it's using a different protocol version, and is incompatible." .. "\nMake sure you have matching versions of both the Rojo plugin and server!" .. @@ -47,14 +49,43 @@ function ApiContext:connect() ):format( Version.display(Config.version), Config.protocolVersion, Config.expectedApiContextVersionString, - response.serverVersion, response.protocolVersion + body.serverVersion, body.protocolVersion ) return Promise.reject(message) end - self.serverId = response.serverId - self.currentTime = response.currentTime + self.serverId = body.serverId + self.connected = true + + return self:_retrieveMessages() + end) +end + +function ApiContext:_retrieveMessages() + if not self.connected then + return Promise.resolve() + end + + return Http.get(self.url .. "/subscribe/" .. self.eventCursor) + :andThen(function(response) + local body = response:json() + + if body.serverId ~= self.serverId then + return Promise.reject("server changed ID") + end + + for _, event in ipairs(body.events) do + self.onMessage(event) + end + + self.eventCursor = body.eventCursor + + return self:_retrieveMessages() + end, function(err) + self.connected = false + + return Promise.reject(err) end) end diff --git a/plugin/src/Http.lua b/plugin/src/Http.lua index 8afb111d..5323ee3e 100644 --- a/plugin/src/Http.lua +++ b/plugin/src/Http.lua @@ -2,7 +2,8 @@ local HttpService = game:GetService("HttpService") local HTTP_DEBUG = false -local Promise = require(script.Parent.Promise) +local Promise = require(script.Parent.Parent.modules.Promise) + local HttpError = require(script.Parent.HttpError) local HttpResponse = require(script.Parent.HttpResponse) diff --git a/plugin/src/Main.server.lua b/plugin/src/Main.server.lua index 2f46c8b1..6cdd9725 100644 --- a/plugin/src/Main.server.lua +++ b/plugin/src/Main.server.lua @@ -43,12 +43,19 @@ local function main() local toolbar = plugin:CreateToolbar("Rojo " .. displayedVersion) + local currentSession + toolbar:CreateButton("Connect", "Connect to Rojo Session", "") .Click:Connect(function() checkUpgrade() - Session.new() - error("NYI") + if currentSession ~= nil then + warn("Rojo: A session is already running!") + return + end + + print("Rojo: Started session.") + currentSession = Session.new() end) end diff --git a/plugin/src/Session.lua b/plugin/src/Session.lua index d69daf5a..f918f898 100644 --- a/plugin/src/Session.lua +++ b/plugin/src/Session.lua @@ -1,7 +1,8 @@ +local Promise = require(script.Parent.Parent.modules.Promise) + local Config = require(script.Parent.Config) local Http = require(script.Parent.Http) local ApiContext = require(script.Parent.ApiContext) -local Promise = require(script.Parent.Promise) local REMOTE_URL = ("http://localhost:%d"):format(Config.port) @@ -13,7 +14,9 @@ function Session.new() setmetatable(self, Session) - local api = ApiContext.new(REMOTE_URL) + local api = ApiContext.new(REMOTE_URL, function(message) + print("Got message:", message) + end) api:connect() return self diff --git a/server/src/commands/serve.rs b/server/src/commands/serve.rs index c36fb11c..03af6050 100644 --- a/server/src/commands/serve.rs +++ b/server/src/commands/serve.rs @@ -1,7 +1,6 @@ use std::collections::HashMap; use std::path::PathBuf; use std::process; -use std::time::Instant; use std::fs; use rand; @@ -50,7 +49,6 @@ pub fn serve(project_path: &PathBuf, port: Option) { let web_config = web::WebConfig { port: port.unwrap_or(project.serve_port), server_id, - start_time: Instant::now(), rbx_session: session.get_rbx_session(), events: session.get_events(), event_listeners: session.get_event_listeners(), diff --git a/server/src/session.rs b/server/src/session.rs index cfeecb34..b4ee4aad 100644 --- a/server/src/session.rs +++ b/server/src/session.rs @@ -26,7 +26,7 @@ pub struct Session { config: SessionConfig, vfs_session: Arc>, rbx_session: Arc>, - middlewares: Vec>, + // middlewares: Vec>, watchers: Vec, events: Arc>>, event_listeners: Arc>>>, @@ -40,7 +40,7 @@ impl Session { Session { vfs_session, rbx_session, - middlewares: Vec::new(), + // middlewares: Vec::new(), watchers: Vec::new(), events: Arc::new(RwLock::new(Vec::new())), event_listeners: Arc::new(Mutex::new(HashMap::new())), @@ -69,12 +69,32 @@ impl Session { { let vfs_session = self.vfs_session.clone(); + let event_listeners = self.event_listeners.clone(); + let events = self.events.clone(); + thread::spawn(move || { loop { match rx.recv() { Ok(change) => { - let mut vfs_session = vfs_session.write().unwrap(); - vfs_session.handle_change(change); + { + let mut vfs_session = vfs_session.write().unwrap(); + vfs_session.handle_change(change); + } + + // TODO: Handle change in RbxSession + + { + let mut events = events.write().unwrap(); + events.push(SessionEvent::Something); + } + + { + let listeners = event_listeners.lock().unwrap(); + + for listener in listeners.values() { + listener.send(()).unwrap(); + } + } }, Err(_) => break, } diff --git a/server/src/web.rs b/server/src/web.rs index 10a07876..ee4139e2 100644 --- a/server/src/web.rs +++ b/server/src/web.rs @@ -1,5 +1,4 @@ use std::collections::HashMap; -use std::time::Instant; use std::sync::{mpsc, RwLock, Mutex, Arc}; use rouille; @@ -14,7 +13,6 @@ use session::SessionEvent; pub struct WebConfig { pub port: u64, pub server_id: u64, - pub start_time: Instant, pub rbx_session: Arc>, pub events: Arc>>, pub event_listeners: Arc>>>, @@ -26,22 +24,22 @@ struct ServerInfoResponse<'a> { server_version: &'static str, protocol_version: u64, server_id: &'a str, - current_time: f64, } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct ReadAllResponse<'a> { server_id: &'a str, - current_time: f64, instances: &'a HashMap, + event_cursor: i32, } #[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] struct SubscribeResponse<'a> { server_id: &'a str, events: &'a [SessionEvent], - cursor: i32, + event_cursor: i32, } /// Start the Rojo web server and park our current thread. @@ -57,17 +55,10 @@ pub fn start(config: WebConfig) { (GET) (/) => { // Get a summary of information about the server. - let current_time = { - let elapsed = config.start_time.elapsed(); - - elapsed.as_secs() as f64 + elapsed.subsec_nanos() as f64 / 1_000_000_000.0 - }; - json_response(ServerInfoResponse { server_version, protocol_version: 2, server_id: &server_id, - current_time, }) }, @@ -79,13 +70,13 @@ pub fn start(config: WebConfig) { { let events = config.events.read().unwrap(); if cursor < events.len() as i32 - 1 { - let new_events = &events[cursor as usize + 1..]; + let new_events = &events[(cursor + 1) as usize..]; let new_cursor = cursor + new_events.len() as i32; return json_response(SubscribeResponse { server_id: &server_id, events: new_events, - cursor: new_cursor, + event_cursor: new_cursor, }); } } @@ -110,13 +101,13 @@ pub fn start(config: WebConfig) { { let events = config.events.read().unwrap(); - let new_events = &events[cursor as usize + 1..]; + let new_events = &events[(cursor + 1) as usize..]; let new_cursor = cursor + new_events.len() as i32; return json_response(SubscribeResponse { server_id: &server_id, events: new_events, - cursor: new_cursor, + event_cursor: new_cursor, }); } }, @@ -124,15 +115,14 @@ pub fn start(config: WebConfig) { (GET) (/read_all) => { let rbx_session = config.rbx_session.read().unwrap(); - let current_time = { - let elapsed = config.start_time.elapsed(); - - elapsed.as_secs() as f64 + elapsed.subsec_nanos() as f64 / 1_000_000_000.0 + let event_cursor = { + let events = config.events.read().unwrap(); + events.len() as i32 - 1 }; json_response(ReadAllResponse { server_id: &server_id, - current_time, + event_cursor, instances: &rbx_session.instances, }) },