Introduce MessageSession, for keeping track of messages to send to clients

server/src/commands/serve.rs

@@ -50,8 +50,7 @@ pub fn serve(project_path: &PathBuf, port: Option<u64>) {
         port: port.unwrap_or(project.serve_port),
         server_id,
         rbx_session: session.get_rbx_session(),
-        events: session.get_events(),
-        event_listeners: session.get_event_listeners(),
+        message_session: session.get_message_session(),
     };
 
     println!("Server listening on port {}", web_config.port);

server/src/lib.rs

@@ -13,6 +13,7 @@ extern crate tempfile;
 pub mod commands;
 pub mod file_route;
 pub mod id;
+pub mod message_session;
 pub mod partition;
 pub mod partition_watcher;
 pub mod pathext;

server/src/message_session.rs

@@ -0,0 +1,57 @@
+use std::collections::HashMap;
+use std::sync::{mpsc, Arc, RwLock, Mutex};
+
+use id::{Id, get_id};
+
+#[derive(Debug, Clone, Serialize)]
+pub enum Message {
+    Something,
+}
+
+#[derive(Clone)]
+pub struct MessageSession {
+   pub messages: Arc<RwLock<Vec<Message>>>,
+   pub message_listeners: Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>>,
+}
+
+impl MessageSession {
+    pub fn new() -> MessageSession {
+        MessageSession {
+            messages: Arc::new(RwLock::new(Vec::new())),
+            message_listeners: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub fn push_messages(&self, new_messages: &[Message]) {
+        {
+            let mut messages = self.messages.write().unwrap();
+            messages.extend_from_slice(new_messages);
+        }
+
+        {
+            let message_listeners = self.message_listeners.lock().unwrap();
+
+            for listener in message_listeners.values() {
+                listener.send(()).unwrap();
+            }
+        }
+    }
+
+    pub fn subscribe(&self, sender: mpsc::Sender<()>) -> Id {
+        let id = get_id();
+
+        {
+            let mut message_listeners = self.message_listeners.lock().unwrap();
+            message_listeners.insert(id, sender);
+        }
+
+        id
+    }
+
+    pub fn unsubscribe(&self, id: Id) {
+        {
+            let mut message_listeners = self.message_listeners.lock().unwrap();
+            message_listeners.remove(&id);
+        }
+    }
+}

server/src/rbx_session.rs

@@ -6,7 +6,8 @@ use id::{Id, get_id};
 use partition::Partition;
 use rbx::RbxInstance;
 use session::SessionConfig;
-use vfs_session::{VfsSession, FileItem};
+use vfs_session::{VfsSession, FileItem, FileChange};
+use message_session::MessageSession;
 
 fn file_to_instances(file_item: &FileItem, partition: &Partition, output: &mut HashMap<Id, RbxInstance>, instances_by_route: &mut HashMap<FileRoute, Id>) -> Id {
     match file_item {
@@ -71,7 +72,7 @@ pub struct RbxSession {
 }
 
 impl RbxSession {
-    pub fn new(config: SessionConfig, vfs_session: Arc<RwLock<VfsSession>>) -> RbxSession {
+    pub fn new(config: SessionConfig, vfs_session: Arc<RwLock<VfsSession>>, message_session: MessageSession) -> RbxSession {
         RbxSession {
             config,
             vfs_session,
@@ -95,4 +96,8 @@ impl RbxSession {
             self.partition_instances.insert(partition.name.clone(), root_id);
         }
     }
+
+    pub fn handle_change(&mut self, change: &FileChange) {
+        println!("Handle RBX change {:?}", change);
+    }
 }

server/src/session.rs

@@ -1,12 +1,12 @@
 use std::collections::HashMap;
-use std::sync::{mpsc, Arc, RwLock, Mutex};
+use std::sync::{mpsc, Arc, RwLock};
 use std::thread;
 
 use partition::Partition;
-use rbx_session::RbxSession;
-use vfs_session::VfsSession;
 use partition_watcher::PartitionWatcher;
-use id::Id;
+use rbx_session::RbxSession;
+use message_session::{MessageSession, Message};
+use vfs_session::VfsSession;
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct SessionConfig {
@@ -17,33 +17,25 @@ 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>>,
+    message_session: MessageSession,
     watchers: Vec<PartitionWatcher>,
-    events: Arc<RwLock<Vec<SessionEvent>>>,
-    event_listeners: Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>>,
 }
 
 impl Session {
     pub fn new(config: SessionConfig) -> Session {
+        let message_session = MessageSession::new();
         let vfs_session = Arc::new(RwLock::new(VfsSession::new(config.clone())));
-        let rbx_session = Arc::new(RwLock::new(RbxSession::new(config.clone(), vfs_session.clone())));
+        let rbx_session = Arc::new(RwLock::new(RbxSession::new(config.clone(), vfs_session.clone(), message_session.clone())));
 
         Session {
             vfs_session,
             rbx_session,
-            // middlewares: Vec::new(),
             watchers: Vec::new(),
-            events: Arc::new(RwLock::new(Vec::new())),
-            event_listeners: Arc::new(Mutex::new(HashMap::new())),
+            message_session,
             config,
         }
     }
@@ -69,8 +61,8 @@ impl Session {
 
         {
             let vfs_session = self.vfs_session.clone();
-            let event_listeners = self.event_listeners.clone();
-            let events = self.events.clone();
+            let rbx_session = self.rbx_session.clone();
+            let message_session = self.message_session.clone();
 
             thread::spawn(move || {
                 loop {
@@ -78,23 +70,15 @@ impl Session {
                         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);
+                                vfs_session.handle_change(&change);
                             }
 
                             {
-                                let listeners = event_listeners.lock().unwrap();
-
-                                for listener in listeners.values() {
-                                    listener.send(()).unwrap();
-                                }
+                                let mut rbx_session = rbx_session.write().unwrap();
+                                rbx_session.handle_change(&change);
                             }
+
+                            message_session.push_messages(&[Message::Something]);
                         },
                         Err(_) => break,
                     }
@@ -114,11 +98,7 @@ impl Session {
         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()
+    pub fn get_message_session(&self) -> MessageSession {
+        self.message_session.clone()
     }
 }

server/src/vfs_session.rs

@@ -64,7 +64,7 @@ impl VfsSession {
         }
     }
 
-    pub fn handle_change(&mut self, change: FileChange) -> Option<()> {
+    pub fn handle_change(&mut self, change: &FileChange) -> Option<()> {
         println!("Got file change {:?}", change);
 
         match change {

server/src/web.rs

@@ -1,21 +1,20 @@
 use std::collections::HashMap;
-use std::sync::{mpsc, RwLock, Mutex, Arc};
+use std::sync::{mpsc, RwLock, Arc};
 
 use rouille;
 
 use web_util::json_response;
-use id::{Id, get_id};
+use id::Id;
 use rbx::RbxInstance;
 use rbx_session::RbxSession;
-use session::SessionEvent;
+use message_session::{MessageSession, Message};
 
 /// The set of configuration the web server needs to start.
 pub struct WebConfig {
     pub port: u64,
     pub server_id: u64,
     pub rbx_session: Arc<RwLock<RbxSession>>,
-    pub events: Arc<RwLock<Vec<SessionEvent>>>,
-    pub event_listeners: Arc<Mutex<HashMap<Id, mpsc::Sender<()>>>>,
+    pub message_session: MessageSession,
 }
 
 #[derive(Debug, Serialize)]
@@ -31,15 +30,15 @@ struct ServerInfoResponse<'a> {
 struct ReadAllResponse<'a> {
     server_id: &'a str,
     instances: &'a HashMap<Id, RbxInstance>,
-    event_cursor: i32,
+    message_cursor: i32,
 }
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct SubscribeResponse<'a> {
     server_id: &'a str,
-    events: &'a [SessionEvent],
-    event_cursor: i32,
+    messages: &'a [Message],
+    message_cursor: i32,
 }
 
 /// Start the Rojo web server and park our current thread.
@@ -68,46 +67,39 @@ pub fn start(config: WebConfig) {
 
                 // 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 + 1) as usize..];
-                        let new_cursor = cursor + new_events.len() as i32;
+                    let messages = config.message_session.messages.read().unwrap();
+                    if cursor < messages.len() as i32 - 1 {
+                        let new_messages = &messages[(cursor + 1) as usize..];
+                        let new_cursor = cursor + new_messages.len() as i32;
 
                         return json_response(SubscribeResponse {
                             server_id: &server_id,
-                            events: new_events,
-                            event_cursor: new_cursor,
+                            messages: new_messages,
+                            message_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);
-                }
+                let sender_id = config.message_session.subscribe(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);
-                }
+                config.message_session.unsubscribe(sender_id);
 
                 {
-                    let events = config.events.read().unwrap();
-                    let new_events = &events[(cursor + 1) as usize..];
-                    let new_cursor = cursor + new_events.len() as i32;
+                    let messages = config.message_session.messages.read().unwrap();
+                    let new_messages = &messages[(cursor + 1) as usize..];
+                    let new_cursor = cursor + new_messages.len() as i32;
 
                     return json_response(SubscribeResponse {
                         server_id: &server_id,
-                        events: new_events,
-                        event_cursor: new_cursor,
+                        messages: new_messages,
+                        message_cursor: new_cursor,
                     });
                 }
             },
@@ -115,14 +107,14 @@ pub fn start(config: WebConfig) {
             (GET) (/read_all) => {
                 let rbx_session = config.rbx_session.read().unwrap();
 
-                let event_cursor = {
-                    let events = config.events.read().unwrap();
-                    events.len() as i32 - 1
+                let message_cursor = {
+                    let messages = config.message_session.messages.read().unwrap();
+                    messages.len() as i32 - 1
                 };
 
                 json_response(ReadAllResponse {
                     server_id: &server_id,
-                    event_cursor,
+                    message_cursor,
                     instances: &rbx_session.instances,
                 })
             },