From 214c93dc95aee85e313b2ba6e09ebb458c521fbe Mon Sep 17 00:00:00 2001 From: Lucien Greathouse Date: Sun, 19 Aug 2018 21:51:12 -0700 Subject: [PATCH] MessageSession -> MessageQueue --- server/src/lib.rs | 2 +- .../{message_session.rs => message_queue.rs} | 17 ++++++--------- server/src/serve_session.rs | 10 ++++----- server/src/web.rs | 21 +++++++++---------- 4 files changed, 22 insertions(+), 28 deletions(-) rename server/src/{message_session.rs => message_queue.rs} (84%) diff --git a/server/src/lib.rs b/server/src/lib.rs index 31f9aef0..8026c919 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -12,7 +12,7 @@ extern crate tempfile; pub mod commands; pub mod id; -pub mod message_session; +pub mod message_queue; pub mod pathext; pub mod project; pub mod rbx; diff --git a/server/src/message_session.rs b/server/src/message_queue.rs similarity index 84% rename from server/src/message_session.rs rename to server/src/message_queue.rs index 840b592a..1029c7e1 100644 --- a/server/src/message_session.rs +++ b/server/src/message_queue.rs @@ -11,14 +11,14 @@ pub enum Message { }, } -pub struct MessageSession { +pub struct MessageQueue { messages: RwLock>, message_listeners: Mutex>>, } -impl MessageSession { - pub fn new() -> MessageSession { - MessageSession { +impl MessageQueue { + pub fn new() -> MessageQueue { + MessageQueue { messages: RwLock::new(Vec::new()), message_listeners: Mutex::new(HashMap::new()), } @@ -66,13 +66,8 @@ impl MessageSession { let current_cursor = messages.len() as u32; - // Cursor is out of bounds - if cursor > current_cursor { - return (current_cursor, Vec::new()); - } - - // No new messages - if cursor == current_cursor { + // Cursor is out of bounds or there are no new messages + if cursor >= current_cursor { return (current_cursor, Vec::new()); } diff --git a/server/src/serve_session.rs b/server/src/serve_session.rs index 30036b5c..dfd90216 100644 --- a/server/src/serve_session.rs +++ b/server/src/serve_session.rs @@ -3,14 +3,14 @@ use std::{ }; use ::{ - message_session::MessageSession, + message_queue::MessageQueue, rbx::RbxTree, project::Project, }; pub struct ServeSession { project: Project, - messages: Arc, + message_queue: Arc, tree: Arc, } @@ -18,7 +18,7 @@ impl ServeSession { pub fn new(project: Project) -> ServeSession { ServeSession { project, - messages: Arc::new(MessageSession::new()), + message_queue: Arc::new(MessageQueue::new()), tree: Arc::new(RbxTree::new()), } } @@ -27,8 +27,8 @@ impl ServeSession { &self.project } - pub fn get_messages(&self) -> &MessageSession { - &self.messages + pub fn get_message_queue(&self) -> &MessageQueue { + &self.message_queue } pub fn get_tree(&self) -> &RbxTree { diff --git a/server/src/web.rs b/server/src/web.rs index ac638a61..d53c53f3 100644 --- a/server/src/web.rs +++ b/server/src/web.rs @@ -7,13 +7,12 @@ use rand; use ::{ id::Id, - message_session::Message, + message_queue::Message, project::Project, rbx::RbxInstance, serve_session::ServeSession, }; - #[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ServerInfoResponse<'a> { @@ -28,7 +27,7 @@ pub struct ServerInfoResponse<'a> { pub struct ReadResponse<'a> { pub server_id: &'a str, pub message_cursor: u32, - // pub instances: HashMap>, + pub instances: HashMap>, } #[derive(Debug, Serialize, Deserialize)] @@ -80,7 +79,7 @@ impl Server { // Did the client miss any messages since the last subscribe? { - let (new_cursor, new_messages) = self.session.get_messages().get_messages_since(cursor); + let (new_cursor, new_messages) = self.session.get_message_queue().get_messages_since(cursor); if new_messages.len() > 0 { return Response::json(&SubscribeResponse { @@ -93,17 +92,17 @@ impl Server { let (tx, rx) = mpsc::channel(); - let sender_id = self.session.get_messages().subscribe(tx); + let sender_id = self.session.get_message_queue().subscribe(tx); match rx.recv() { Ok(_) => (), Err(_) => return Response::text("error!").with_status_code(500), } - self.session.get_messages().unsubscribe(sender_id); + self.session.get_message_queue().unsubscribe(sender_id); { - let (new_cursor, new_messages) = self.session.get_messages().get_messages_since(cursor); + let (new_cursor, new_messages) = self.session.get_message_queue().get_messages_since(cursor); return Response::json(&SubscribeResponse { server_id: &self.server_id, @@ -126,17 +125,17 @@ impl Server { let rbx_tree = self.session.get_tree(); - let message_cursor = self.session.get_messages().get_message_cursor(); + let message_cursor = self.session.get_message_queue().get_message_cursor(); let mut instances = HashMap::new(); for &requested_id in &requested_ids { match rbx_tree.get_instance(requested_id) { Some(instance) => { - instances.insert(instance.get_id(), instance); + instances.insert(instance.get_id(), Cow::Borrowed(instance)); for descendant in rbx_tree.iter_descendants(requested_id) { - instances.insert(descendant.get_id(), descendant); + instances.insert(descendant.get_id(), Cow::Borrowed(descendant)); } }, None => {}, @@ -146,7 +145,7 @@ impl Server { Response::json(&ReadResponse { server_id: &self.server_id, message_cursor, - // instances, + instances, }) },