MessageSession -> MessageQueue

This commit is contained in:
Lucien Greathouse 2018-08-19 21:51:12 -07:00
parent 2a4704f5b0
commit 214c93dc95
4 changed files with 22 additions and 28 deletions

View file

@ -12,7 +12,7 @@ extern crate tempfile;
pub mod commands; pub mod commands;
pub mod id; pub mod id;
pub mod message_session; pub mod message_queue;
pub mod pathext; pub mod pathext;
pub mod project; pub mod project;
pub mod rbx; pub mod rbx;

View file

@ -11,14 +11,14 @@ pub enum Message {
}, },
} }
pub struct MessageSession { pub struct MessageQueue {
messages: RwLock<Vec<Message>>, messages: RwLock<Vec<Message>>,
message_listeners: Mutex<HashMap<Id, mpsc::Sender<()>>>, message_listeners: Mutex<HashMap<Id, mpsc::Sender<()>>>,
} }
impl MessageSession { impl MessageQueue {
pub fn new() -> MessageSession { pub fn new() -> MessageQueue {
MessageSession { MessageQueue {
messages: RwLock::new(Vec::new()), messages: RwLock::new(Vec::new()),
message_listeners: Mutex::new(HashMap::new()), message_listeners: Mutex::new(HashMap::new()),
} }
@ -66,13 +66,8 @@ impl MessageSession {
let current_cursor = messages.len() as u32; let current_cursor = messages.len() as u32;
// Cursor is out of bounds // Cursor is out of bounds or there are no new messages
if cursor > current_cursor { if cursor >= current_cursor {
return (current_cursor, Vec::new());
}
// No new messages
if cursor == current_cursor {
return (current_cursor, Vec::new()); return (current_cursor, Vec::new());
} }

View file

@ -3,14 +3,14 @@ use std::{
}; };
use ::{ use ::{
message_session::MessageSession, message_queue::MessageQueue,
rbx::RbxTree, rbx::RbxTree,
project::Project, project::Project,
}; };
pub struct ServeSession { pub struct ServeSession {
project: Project, project: Project,
messages: Arc<MessageSession>, message_queue: Arc<MessageQueue>,
tree: Arc<RbxTree>, tree: Arc<RbxTree>,
} }
@ -18,7 +18,7 @@ impl ServeSession {
pub fn new(project: Project) -> ServeSession { pub fn new(project: Project) -> ServeSession {
ServeSession { ServeSession {
project, project,
messages: Arc::new(MessageSession::new()), message_queue: Arc::new(MessageQueue::new()),
tree: Arc::new(RbxTree::new()), tree: Arc::new(RbxTree::new()),
} }
} }
@ -27,8 +27,8 @@ impl ServeSession {
&self.project &self.project
} }
pub fn get_messages(&self) -> &MessageSession { pub fn get_message_queue(&self) -> &MessageQueue {
&self.messages &self.message_queue
} }
pub fn get_tree(&self) -> &RbxTree { pub fn get_tree(&self) -> &RbxTree {

View file

@ -7,13 +7,12 @@ use rand;
use ::{ use ::{
id::Id, id::Id,
message_session::Message, message_queue::Message,
project::Project, project::Project,
rbx::RbxInstance, rbx::RbxInstance,
serve_session::ServeSession, serve_session::ServeSession,
}; };
#[derive(Debug, Serialize, Deserialize)] #[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
pub struct ServerInfoResponse<'a> { pub struct ServerInfoResponse<'a> {
@ -28,7 +27,7 @@ pub struct ServerInfoResponse<'a> {
pub struct ReadResponse<'a> { pub struct ReadResponse<'a> {
pub server_id: &'a str, pub server_id: &'a str,
pub message_cursor: u32, pub message_cursor: u32,
// pub instances: HashMap<Id, Cow<'a, RbxInstance>>, pub instances: HashMap<Id, Cow<'a, RbxInstance>>,
} }
#[derive(Debug, Serialize, Deserialize)] #[derive(Debug, Serialize, Deserialize)]
@ -80,7 +79,7 @@ impl Server {
// Did the client miss any messages since the last subscribe? // 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 { if new_messages.len() > 0 {
return Response::json(&SubscribeResponse { return Response::json(&SubscribeResponse {
@ -93,17 +92,17 @@ impl Server {
let (tx, rx) = mpsc::channel(); 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() { match rx.recv() {
Ok(_) => (), Ok(_) => (),
Err(_) => return Response::text("error!").with_status_code(500), 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 { return Response::json(&SubscribeResponse {
server_id: &self.server_id, server_id: &self.server_id,
@ -126,17 +125,17 @@ impl Server {
let rbx_tree = self.session.get_tree(); 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(); let mut instances = HashMap::new();
for &requested_id in &requested_ids { for &requested_id in &requested_ids {
match rbx_tree.get_instance(requested_id) { match rbx_tree.get_instance(requested_id) {
Some(instance) => { 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) { for descendant in rbx_tree.iter_descendants(requested_id) {
instances.insert(descendant.get_id(), descendant); instances.insert(descendant.get_id(), Cow::Borrowed(descendant));
} }
}, },
None => {}, None => {},
@ -146,7 +145,7 @@ impl Server {
Response::json(&ReadResponse { Response::json(&ReadResponse {
server_id: &self.server_id, server_id: &self.server_id,
message_cursor, message_cursor,
// instances, instances,
}) })
}, },