Implement long-polling message passing

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"

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.serverId = body.serverId
-			self.currentTime = response.currentTime
+			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
 

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)
 

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
 

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

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<u64>) {
     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(),

server/src/session.rs

@@ -26,7 +26,7 @@ pub struct Session {
     config: SessionConfig,
     vfs_session: Arc<RwLock<VfsSession>>,
     rbx_session: Arc<RwLock<RbxSession>>,
-    middlewares: Vec<Box<Middleware>>,
+    // middlewares: Vec<Box<Middleware>>,
     watchers: Vec<PartitionWatcher>,
     events: Arc<RwLock<Vec<SessionEvent>>>,
     event_listeners: Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>>,
@@ -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);
+                            }
+
+                            // 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,
                     }

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<RwLock<RbxSession>>,
     pub events: Arc<RwLock<Vec<SessionEvent>>>,
     pub event_listeners: Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>>,
@@ -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<Id, RbxInstance>,
+    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 event_cursor = {
-                    let elapsed = config.start_time.elapsed();
+                    let events = config.events.read().unwrap();
-
+                    events.len() as i32 - 1
-                    elapsed.as_secs() as f64 + elapsed.subsec_nanos() as f64 / 1_000_000_000.0
                 };
 
                 json_response(ReadAllResponse {
                     server_id: &server_id,
-                    current_time,
+                    event_cursor,
                     instances: &rbx_session.instances,
                 })
             },