From 8729de1312c1634279ab637ad3987fb274f93653 Mon Sep 17 00:00:00 2001 From: Lucien Greathouse Date: Sat, 26 May 2018 02:06:10 -0700 Subject: [PATCH] Basic synchronization of edited objects --- plugin/src/ApiContext.lua | 31 +++++++++++--- plugin/src/Session.lua | 56 +++++++++++++++++++++--- server/src/message_session.rs | 8 +++- server/src/rbx_session.rs | 80 +++++++++++++++++++++++++++-------- server/src/session.rs | 5 +-- server/src/vfs_session.rs | 2 - 6 files changed, 148 insertions(+), 34 deletions(-) diff --git a/plugin/src/ApiContext.lua b/plugin/src/ApiContext.lua index 981b286d..d32a5612 100644 --- a/plugin/src/ApiContext.lua +++ b/plugin/src/ApiContext.lua @@ -57,14 +57,35 @@ function ApiContext:connect() self.serverId = body.serverId self.connected = true - - return self:_retrieveMessages() end) end -function ApiContext:_retrieveMessages() +function ApiContext:readAll() if not self.connected then - return Promise.resolve() + return Promise.reject() + end + + return Http.get(self.url .. "/read_all") + :andThen(function(response) + local body = response:json() + + if body.serverId ~= self.serverId then + return Promise.reject("server changed ID") + end + + self.messageCursor = body.messageCursor + + return body.instances + end, function(err) + self.connected = false + + return Promise.reject(err) + end) +end + +function ApiContext:retrieveMessages() + if not self.connected then + return Promise.reject() end return Http.get(self.url .. "/subscribe/" .. self.messageCursor) @@ -81,7 +102,7 @@ function ApiContext:_retrieveMessages() self.messageCursor = body.messageCursor - return self:_retrieveMessages() + return self:retrieveMessages() end, function(err) self.connected = false diff --git a/plugin/src/Session.lua b/plugin/src/Session.lua index f918f898..2570d9f0 100644 --- a/plugin/src/Session.lua +++ b/plugin/src/Session.lua @@ -1,7 +1,4 @@ -local Promise = require(script.Parent.Parent.modules.Promise) - local Config = require(script.Parent.Config) -local Http = require(script.Parent.Http) local ApiContext = require(script.Parent.ApiContext) local REMOTE_URL = ("http://localhost:%d"):format(Config.port) @@ -14,10 +11,59 @@ function Session.new() setmetatable(self, Session) - local api = ApiContext.new(REMOTE_URL, function(message) - print("Got message:", message) + local created = {} + + created["0"] = game:GetService("ReplicatedFirst") + + local api + local function readAll() + print("Reading all...") + + return api:readAll() + :andThen(function(instances) + for id, instance in pairs(instances) do + if id ~= "0" then + local existing = created[id] + if existing ~= nil then + pcall(existing.Destroy, existing) + end + + local real = Instance.new(instance.className) + real.Name = instance.name + + for key, value in pairs(instance.properties) do + real[key] = value + end + + created[id] = real + end + end + + for id, instance in pairs(instances) do + if id ~= "0" then + print("parent", created[id], created[tostring(instance.parent)]) + created[id].Parent = created[tostring(instance.parent)] + end + end + end) + end + + api = ApiContext.new(REMOTE_URL, function(message) + print("got message", message) + + if message.type == "InstanceChanged" then + print("Instance", message.id, "changed!") + readAll() + else + warn("Unknown message type " .. message.type) + end end) + api:connect() + :andThen(readAll) + :andThen(function() + return api:retrieveMessages() + end) return self end diff --git a/server/src/message_session.rs b/server/src/message_session.rs index a1426101..7f2c840f 100644 --- a/server/src/message_session.rs +++ b/server/src/message_session.rs @@ -4,8 +4,14 @@ use std::sync::{mpsc, Arc, RwLock, Mutex}; use id::{Id, get_id}; #[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] pub enum Message { - Something, + InstanceChanged { + id: Id, + }, + Error { + message: String + }, } #[derive(Clone)] diff --git a/server/src/rbx_session.rs b/server/src/rbx_session.rs index 3bbaaa67..d1d766d6 100644 --- a/server/src/rbx_session.rs +++ b/server/src/rbx_session.rs @@ -7,51 +7,74 @@ use partition::Partition; use rbx::RbxInstance; use session::SessionConfig; use vfs_session::{VfsSession, FileItem, FileChange}; -use message_session::MessageSession; +use message_session::{Message, MessageSession}; -fn file_to_instances(file_item: &FileItem, partition: &Partition, output: &mut HashMap, instances_by_route: &mut HashMap) -> Id { +fn file_to_instances( + file_item: &FileItem, + partition: &Partition, + output: &mut HashMap, + instances_by_route: &mut HashMap, +) -> (Id, Vec) { match file_item { FileItem::File { contents, route } => { let mut properties = HashMap::new(); properties.insert("Value".to_string(), contents.clone()); - let primary_id = get_id(); + let (primary_id, parent) = match instances_by_route.get(&file_item.get_route()) { + Some(&id) => { + (id, output.get(&id).unwrap().parent) + }, + None => { + let id = get_id(); + instances_by_route.insert(route.clone(), id); - instances_by_route.insert(route.clone(), primary_id); + (id, None) + }, + }; output.insert(primary_id, RbxInstance { name: route.name(partition).to_string(), class_name: "StringValue".to_string(), - parent: None, + parent, properties, }); - primary_id + (primary_id, vec![primary_id]) }, FileItem::Directory { children, route } => { - let primary_id = get_id(); + let (primary_id, parent) = match instances_by_route.get(&file_item.get_route()) { + Some(&id) => { + (id, output.get(&id).unwrap().parent) + }, + None => { + let id = get_id(); + instances_by_route.insert(route.clone(), id); - instances_by_route.insert(route.clone(), primary_id); + (id, None) + }, + }; output.insert(primary_id, RbxInstance { name: route.name(partition).to_string(), class_name: "Folder".to_string(), - parent: None, + parent, properties: HashMap::new(), }); + let mut changed_ids = vec![primary_id]; + for child_file_item in children.values() { - let mut child_instances = HashMap::new(); - let child_id = file_to_instances(child_file_item, partition, &mut child_instances, instances_by_route); + let (child_id, mut child_changed_ids) = file_to_instances(child_file_item, partition, output, instances_by_route); - child_instances.get_mut(&child_id).unwrap().parent = Some(primary_id); + output.get_mut(&child_id).unwrap().parent = Some(primary_id); - for (instance_id, instance) in child_instances.drain() { - output.insert(instance_id, instance); + changed_ids.push(child_id); + for id in child_changed_ids.drain(..) { + changed_ids.push(id); } } - primary_id + (primary_id, changed_ids) }, } } @@ -61,6 +84,8 @@ pub struct RbxSession { vfs_session: Arc>, + message_session: MessageSession, + /// The RbxInstance that represents each partition. partition_instances: HashMap, @@ -76,6 +101,7 @@ impl RbxSession { RbxSession { config, vfs_session, + message_session, partition_instances: HashMap::new(), instances: HashMap::new(), instances_by_route: HashMap::new(), @@ -92,12 +118,32 @@ impl RbxSession { route: Vec::new(), }; let file_item = vfs_session.get_by_route(&route).unwrap(); - let root_id = file_to_instances(file_item, partition, &mut self.instances, &mut self.instances_by_route); + let (root_id, _) = file_to_instances(file_item, partition, &mut self.instances, &mut self.instances_by_route); self.partition_instances.insert(partition.name.clone(), root_id); } } pub fn handle_change(&mut self, change: &FileChange) { - println!("Handle RBX change {:?}", change); + let vfs_session_arc = self.vfs_session.clone(); + let vfs_session = vfs_session_arc.read().unwrap(); + + match change { + FileChange::Created(route) | FileChange::Updated(route) => { + let file_item = vfs_session.get_by_route(route).unwrap(); + let partition = self.config.partitions.get(&route.partition).unwrap(); + let (_, changed_ids) = file_to_instances(file_item, partition, &mut self.instances, &mut self.instances_by_route); + + let messages = changed_ids + .iter() + .map(|&id| Message::InstanceChanged { id }) + .collect::>(); + + self.message_session.push_messages(&messages); + }, + FileChange::Deleted(route) => { + }, + FileChange::Moved(from_route, to_route) => { + }, + } } } diff --git a/server/src/session.rs b/server/src/session.rs index 37dd3311..6f493260 100644 --- a/server/src/session.rs +++ b/server/src/session.rs @@ -5,7 +5,7 @@ use std::thread; use partition::Partition; use partition_watcher::PartitionWatcher; use rbx_session::RbxSession; -use message_session::{MessageSession, Message}; +use message_session::MessageSession; use vfs_session::VfsSession; #[derive(Debug, Clone, PartialEq)] @@ -62,7 +62,6 @@ impl Session { { let vfs_session = self.vfs_session.clone(); let rbx_session = self.rbx_session.clone(); - let message_session = self.message_session.clone(); thread::spawn(move || { loop { @@ -77,8 +76,6 @@ impl Session { let mut rbx_session = rbx_session.write().unwrap(); rbx_session.handle_change(&change); } - - message_session.push_messages(&[Message::Something]); }, Err(_) => break, } diff --git a/server/src/vfs_session.rs b/server/src/vfs_session.rs index 17ef13a1..3a60cfbe 100644 --- a/server/src/vfs_session.rs +++ b/server/src/vfs_session.rs @@ -65,8 +65,6 @@ impl VfsSession { } pub fn handle_change(&mut self, change: &FileChange) -> Option<()> { - println!("Got file change {:?}", change); - match change { FileChange::Created(route) | FileChange::Updated(route) => { let new_item = self.read(&route).ok()?;