From d85383cfd49f6947115fee9082794ffc2445e052 Mon Sep 17 00:00:00 2001 From: Lucien Greathouse Date: Sat, 26 May 2018 01:11:05 -0700 Subject: [PATCH] Introduce MessageSession, for keeping track of messages to send to clients --- server/src/commands/serve.rs | 3 +- server/src/lib.rs | 1 + server/src/message_session.rs | 57 +++++++++++++++++++++++++++++++++++ server/src/rbx_session.rs | 9 ++++-- server/src/session.rs | 54 +++++++++++---------------------- server/src/vfs_session.rs | 2 +- server/src/web.rs | 56 +++++++++++++++------------------- 7 files changed, 108 insertions(+), 74 deletions(-) create mode 100644 server/src/message_session.rs diff --git a/server/src/commands/serve.rs b/server/src/commands/serve.rs index 03af6050..b5e0eb4c 100644 --- a/server/src/commands/serve.rs +++ b/server/src/commands/serve.rs @@ -50,8 +50,7 @@ pub fn serve(project_path: &PathBuf, port: Option) { 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); diff --git a/server/src/lib.rs b/server/src/lib.rs index baccab96..5ea4b8f0 100644 --- a/server/src/lib.rs +++ b/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; diff --git a/server/src/message_session.rs b/server/src/message_session.rs new file mode 100644 index 00000000..a1426101 --- /dev/null +++ b/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>>, + pub message_listeners: Arc>>>, +} + +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); + } + } +} diff --git a/server/src/rbx_session.rs b/server/src/rbx_session.rs index fadc3ab2..3bbaaa67 100644 --- a/server/src/rbx_session.rs +++ b/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, instances_by_route: &mut HashMap) -> Id { match file_item { @@ -71,7 +72,7 @@ pub struct RbxSession { } impl RbxSession { - pub fn new(config: SessionConfig, vfs_session: Arc>) -> RbxSession { + pub fn new(config: SessionConfig, vfs_session: Arc>, 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); + } } diff --git a/server/src/session.rs b/server/src/session.rs index b4ee4aad..37dd3311 100644 --- a/server/src/session.rs +++ b/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>, rbx_session: Arc>, - // middlewares: Vec>, + message_session: MessageSession, watchers: Vec, - events: Arc>>, - event_listeners: Arc>>>, } 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>> { - self.events.clone() - } - - pub fn get_event_listeners(&self) -> Arc>>> { - self.event_listeners.clone() + pub fn get_message_session(&self) -> MessageSession { + self.message_session.clone() } } diff --git a/server/src/vfs_session.rs b/server/src/vfs_session.rs index 1ae5da86..17ef13a1 100644 --- a/server/src/vfs_session.rs +++ b/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 { diff --git a/server/src/web.rs b/server/src/web.rs index ee4139e2..3246db02 100644 --- a/server/src/web.rs +++ b/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>, - pub events: Arc>>, - pub event_listeners: Arc>>>, + pub message_session: MessageSession, } #[derive(Debug, Serialize)] @@ -31,15 +30,15 @@ struct ServerInfoResponse<'a> { struct ReadAllResponse<'a> { server_id: &'a str, instances: &'a HashMap, - 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, }) },