Make MessageQueue generic, collapse Message into a single struct

server/src/message_queue.rs

@@ -8,8 +8,6 @@ use std::{
     },
 };
 
-use rbx_tree::RbxId;
-
 /// A unique identifier, not guaranteed to be generated in any order.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct ListenerId(usize);
@@ -21,32 +19,21 @@ pub fn get_listener_id() -> ListenerId {
     ListenerId(LAST_ID.fetch_add(1, Ordering::SeqCst))
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(tag = "type")]
-pub enum Message {
-    InstancesRemoved {
-        ids: Vec<RbxId>,
-    },
-    InstancesUpdated {
-        ids: Vec<RbxId>,
-    },
-}
-
 #[derive(Default)]
-pub struct MessageQueue {
-   messages: RwLock<Vec<Message>>,
+pub struct MessageQueue<T> {
+   messages: RwLock<Vec<T>>,
    message_listeners: Mutex<HashMap<ListenerId, mpsc::Sender<()>>>,
 }
 
-impl MessageQueue {
-    pub fn new() -> MessageQueue {
+impl<T: Clone> MessageQueue<T> {
+    pub fn new() -> MessageQueue<T> {
         MessageQueue {
             messages: RwLock::new(Vec::new()),
             message_listeners: Mutex::new(HashMap::new()),
         }
     }
 
-    pub fn push_messages(&self, new_messages: &[Message]) {
+    pub fn push_messages(&self, new_messages: &[T]) {
         let message_listeners = self.message_listeners.lock().unwrap();
 
         {
@@ -77,7 +64,7 @@ impl MessageQueue {
         self.messages.read().unwrap().len() as u32
     }
 
-    pub fn get_messages_since(&self, cursor: u32) -> (u32, Vec<Message>) {
+    pub fn get_messages_since(&self, cursor: u32) -> (u32, Vec<T>) {
         let messages = self.messages.read().unwrap();
 
         let current_cursor = messages.len() as u32;

server/src/rbx_session.rs

@@ -1,6 +1,6 @@
 use std::{
     borrow::Cow,
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     path::Path,
     sync::{Arc, Mutex},
     str,
@@ -10,23 +10,34 @@ use rbx_tree::{RbxTree, RbxValue, RbxId};
 
 use crate::{
     project::{Project, ProjectNode, InstanceProjectNode},
-    message_queue::{Message, MessageQueue},
+    message_queue::MessageQueue,
     imfs::{Imfs, ImfsItem, ImfsFile},
     path_map::PathMap,
     rbx_snapshot::{RbxSnapshotInstance, reify_root, reconcile_subtree},
 };
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InstanceChanges {
+    pub added: HashSet<RbxId>,
+    pub removed: HashSet<RbxId>,
+    pub updated: HashSet<RbxId>,
+}
+
 pub struct RbxSession {
     tree: RbxTree,
     paths_to_node_ids: PathMap<RbxId>,
     ids_to_project_paths: HashMap<RbxId, String>,
-    message_queue: Arc<MessageQueue>,
+    message_queue: Arc<MessageQueue<InstanceChanges>>,
     imfs: Arc<Mutex<Imfs>>,
     project: Arc<Project>,
 }
 
 impl RbxSession {
-    pub fn new(project: Arc<Project>, imfs: Arc<Mutex<Imfs>>, message_queue: Arc<MessageQueue>) -> RbxSession {
+    pub fn new(
+        project: Arc<Project>,
+        imfs: Arc<Mutex<Imfs>>,
+        message_queue: Arc<MessageQueue<InstanceChanges>>,
+    ) -> RbxSession {
         let tree = {
             let temp_imfs = imfs.lock().unwrap();
             construct_initial_tree(&project, &temp_imfs)
@@ -97,12 +108,14 @@ impl RbxSession {
             },
         };
 
-        let removed_ids: Vec<RbxId> = removed_subtree.iter_all_ids().collect();
+        let removed_ids: HashSet<RbxId> = removed_subtree.iter_all_ids().collect();
 
         self.message_queue.push_messages(&[
-            Message::InstancesRemoved {
-                ids: removed_ids,
-            },
+            InstanceChanges {
+                added: HashSet::new(),
+                removed: removed_ids,
+                updated: HashSet::new(),
+            }
         ]);
     }
 

server/src/session.rs

@@ -8,14 +8,14 @@ use crate::{
     project::Project,
     imfs::Imfs,
     session_id::SessionId,
-    rbx_session::RbxSession,
+    rbx_session::{RbxSession, InstanceChanges},
     fs_watcher::FsWatcher,
 };
 
 pub struct Session {
     pub project: Arc<Project>,
     pub session_id: SessionId,
-    pub message_queue: Arc<MessageQueue>,
+    pub message_queue: Arc<MessageQueue<InstanceChanges>>,
     pub rbx_session: Arc<Mutex<RbxSession>>,
     _fs_watcher: FsWatcher,
 }

server/src/web.rs

@@ -13,10 +13,10 @@ use rouille::{
 use rbx_tree::{RbxId, RootedRbxInstance};
 
 use crate::{
-    message_queue::Message,
     session::Session,
     session_id::SessionId,
     project::Project,
+    rbx_session::{InstanceChanges},
 };
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -43,7 +43,7 @@ pub struct ReadResponse<'a> {
 pub struct SubscribeResponse<'a> {
     pub session_id: SessionId,
     pub message_cursor: u32,
-    pub messages: Cow<'a, [Message]>,
+    pub messages: Cow<'a, [InstanceChanges]>,
 }
 
 pub struct Server {