From de02966db6358a009d2d058c6ef218a81281331f Mon Sep 17 00:00:00 2001
From: Lucien Greathouse <me@lpghatguy.com>
Date: Fri, 18 May 2018 00:43:12 -0700
Subject: [PATCH] Implement portion of long-polling message interface, untested

---
 server/src/commands/serve.rs |  7 ++--
 server/src/session.rs        | 20 ++++++++++-
 server/src/web.rs            | 65 ++++++++++++++++++++++++++++++++++--
 3 files changed, 84 insertions(+), 8 deletions(-)

diff --git a/server/src/commands/serve.rs b/server/src/commands/serve.rs
index 788de68c..c36fb11c 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::thread;
 use std::time::Instant;
 use std::fs;
 
@@ -53,13 +52,11 @@ pub fn serve(project_path: &PathBuf, port: Option<u64>) {
         server_id,
         start_time: Instant::now(),
         rbx_session: session.get_rbx_session(),
+        events: session.get_events(),
+        event_listeners: session.get_event_listeners(),
     };
 
     println!("Server listening on port {}", web_config.port);
 
     web::start(web_config);
-
-    // loop {
-    //     thread::park();
-    // }
 }
diff --git a/server/src/session.rs b/server/src/session.rs
index bc443c1e..cfeecb34 100644
--- a/server/src/session.rs
+++ b/server/src/session.rs
@@ -1,11 +1,12 @@
 use std::collections::HashMap;
-use std::sync::{mpsc, Arc, RwLock};
+use std::sync::{mpsc, Arc, RwLock, Mutex};
 use std::thread;
 
 use partition::Partition;
 use rbx_session::RbxSession;
 use vfs_session::VfsSession;
 use partition_watcher::PartitionWatcher;
+use id::Id;
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct SessionConfig {
@@ -16,12 +17,19 @@ pub struct SessionConfig {
 trait Middleware {
 }
 
+#[derive(Debug, Clone, Serialize)]
+pub enum SessionEvent {
+    Something,
+}
+
 pub struct Session {
     config: SessionConfig,
     vfs_session: Arc<RwLock<VfsSession>>,
     rbx_session: Arc<RwLock<RbxSession>>,
     middlewares: Vec<Box<Middleware>>,
     watchers: Vec<PartitionWatcher>,
+    events: Arc<RwLock<Vec<SessionEvent>>>,
+    event_listeners: Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>>,
 }
 
 impl Session {
@@ -34,6 +42,8 @@ impl Session {
             rbx_session,
             middlewares: Vec::new(),
             watchers: Vec::new(),
+            events: Arc::new(RwLock::new(Vec::new())),
+            event_listeners: Arc::new(Mutex::new(HashMap::new())),
             config,
         }
     }
@@ -83,4 +93,12 @@ impl Session {
     pub fn get_rbx_session(&self) -> Arc<RwLock<RbxSession>> {
         self.rbx_session.clone()
     }
+
+    pub fn get_events(&self) -> Arc<RwLock<Vec<SessionEvent>>> {
+        self.events.clone()
+    }
+
+    pub fn get_event_listeners(&self) -> Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>> {
+        self.event_listeners.clone()
+    }
 }
diff --git a/server/src/web.rs b/server/src/web.rs
index 38cb4237..10a07876 100644
--- a/server/src/web.rs
+++ b/server/src/web.rs
@@ -1,13 +1,14 @@
 use std::collections::HashMap;
 use std::time::Instant;
-use std::sync::{RwLock, Arc};
+use std::sync::{mpsc, RwLock, Mutex, Arc};
 
 use rouille;
 
 use web_util::json_response;
-use id::Id;
+use id::{Id, get_id};
 use rbx::RbxInstance;
 use rbx_session::RbxSession;
+use session::SessionEvent;
 
 /// The set of configuration the web server needs to start.
 pub struct WebConfig {
@@ -15,6 +16,8 @@ pub struct WebConfig {
     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<()>>>>,
 }
 
 #[derive(Debug, Serialize)]
@@ -34,7 +37,15 @@ struct ReadAllResponse<'a> {
     instances: &'a HashMap<Id, RbxInstance>,
 }
 
+#[derive(Debug, Serialize)]
+struct SubscribeResponse<'a> {
+    server_id: &'a str,
+    events: &'a [SessionEvent],
+    cursor: i32,
+}
+
 /// Start the Rojo web server and park our current thread.
+#[allow(unreachable_code)]
 pub fn start(config: WebConfig) {
     let address = format!("localhost:{}", config.port);
     let server_version = env!("CARGO_PKG_VERSION");
@@ -60,6 +71,56 @@ pub fn start(config: WebConfig) {
                 })
             },
 
+            (GET) (/subscribe/{ cursor: i32 }) => {
+                // Retrieve any messages past the given cursor index, and if
+                // there weren't any, subscribe to receive any new messages.
+
+                // Did the client miss any messages since the last subscribe?
+                {
+                    let events = config.events.read().unwrap();
+                    if cursor < events.len() as i32 - 1 {
+                        let new_events = &events[cursor as usize + 1..];
+                        let new_cursor = cursor + new_events.len() as i32;
+
+                        return json_response(SubscribeResponse {
+                            server_id: &server_id,
+                            events: new_events,
+                            cursor: new_cursor,
+                        });
+                    }
+                }
+
+                let sender_id = get_id();
+                let (tx, rx) = mpsc::channel();
+
+                {
+                    let mut event_listeners = config.event_listeners.lock().unwrap();
+                    event_listeners.insert(sender_id, tx);
+                }
+
+                match rx.recv() {
+                    Ok(_) => (),
+                    Err(_) => return rouille::Response::text("error!").with_status_code(500),
+                }
+
+                {
+                    let mut event_listeners = config.event_listeners.lock().unwrap();
+                    event_listeners.remove(&sender_id);
+                }
+
+                {
+                    let events = config.events.read().unwrap();
+                    let new_events = &events[cursor as usize + 1..];
+                    let new_cursor = cursor + new_events.len() as i32;
+
+                    return json_response(SubscribeResponse {
+                        server_id: &server_id,
+                        events: new_events,
+                        cursor: new_cursor,
+                    });
+                }
+            },
+
             (GET) (/read_all) => {
                 let rbx_session = config.rbx_session.read().unwrap();