diff --git a/Cargo.lock b/Cargo.lock index 4b627f6..519edc4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1839,6 +1839,7 @@ dependencies = [ "tokio-stream", "toml", "tower-http", + "uuid", ] [[package]] @@ -3281,9 +3282,9 @@ checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" [[package]] name = "uuid" -version = "1.2.1" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "feb41e78f93363bb2df8b0e86a2ca30eed7806ea16ea0c790d757cf93f79be83" +checksum = "1674845326ee10d37ca60470760d4288a6f80f304007d92e5c53bab78c9cfd79" dependencies = [ "getrandom 0.2.7", "serde", diff --git a/Cargo.toml b/Cargo.toml index 7b78492..05e5e83 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,3 +21,4 @@ tokio = {version = "1.21.0", features = ["full"]} tokio-stream = "0.1.9" toml = "0.5.10" tower-http = "0.3.5" +uuid = "1.3.0" diff --git a/js-client b/js-client index df13e2a..c2f331c 160000 --- a/js-client +++ b/js-client @@ -1 +1 @@ -Subproject commit df13e2a4acd920eea7d8b7efc63ae4766da367d6 +Subproject commit c2f331ceb89d07d0019b993b2c0983c27750ed94 diff --git a/src/server.rs b/src/server.rs index 161d2df..470c105 100644 --- a/src/server.rs +++ b/src/server.rs @@ -14,9 +14,11 @@ use crate::error; use crate::error::MizeError; use crate::error::ERRORS; use crate::server::proto::MizeMessage; +use crate::server::proto::MizeId; use serde_json::Value as JsonValue; use serde::{Serialize, Deserialize}; +use uuid; use tokio::sync::mpsc::{Sender, channel, Receiver}; use tokio_stream::wrappers::ReceiverStream; @@ -78,7 +80,7 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000; pub enum Peer { Client(Client), //the client id Module(Module), //Module_name - Upstream(String), //a hostname Type, but for now just a string + Upstream(Upstream), //a hostname Type, but for now just a string } impl Peer { @@ -99,6 +101,9 @@ impl Peer { } } +pub struct Upstream { + id: uuid::uuid, +} #[derive(Clone, Debug)] pub struct Client { @@ -164,11 +169,12 @@ pub struct Mutexes { next_free_client_id: Arc>, clients: Arc>>, modules: Arc>>, - subs: Arc>>>, + subs: Arc>>>, renders: Arc>>, //maybe a list of upstream servers?? itemstore: Arc>, mize_folder: String, + server_uuid: uuid::Uuid, } impl Mutexes { @@ -181,6 +187,7 @@ impl Mutexes { subs: Arc::clone(&mutexes.subs), mize_folder: mutexes.mize_folder.clone(), itemstore: Arc::clone(&mutexes.itemstore), + server_uuid: mutexes.server_uuid, } } } @@ -207,6 +214,9 @@ pub async fn run_server(args: Vec) { //create the itemstore let itemstore = crate::server::itemstore::Itemstore::new(mize_folder.clone() + "/db").await.expect("error creating itemstore"); + //create mize.toml with id inside in folder if net yet done + let server_uuid = uuid::Uuid::new_4(); + let renders = load_mr(mize_folder.clone()).expect("Error loading Modules and Renders"); // Collection of all the things, that most functions need @@ -218,8 +228,10 @@ pub async fn run_server(args: Vec) { renders: Arc::new(Mutex::new(renders)), itemstore: Arc::new(Mutex::new(itemstore)), mize_folder: mize_folder.clone(), + server_uuid, }; + //listen on the local unix socket //local_socket_server(mize_folder); diff --git a/src/server/proto.rs b/src/server/proto.rs index 24c02b5..2be7435 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -77,6 +77,7 @@ pub enum JsonMessage { pub struct GetItemMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, } @@ -84,6 +85,7 @@ pub struct GetItemMessage { pub struct GetSubMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, } @@ -91,6 +93,7 @@ pub struct GetSubMessage { pub struct GiveItemMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, item: Item, } @@ -99,6 +102,7 @@ pub struct GiveItemMessage { pub struct CreateItemMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, item: Item, } @@ -107,6 +111,7 @@ pub struct CreateItemMessage { pub struct CreatedIdMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, } @@ -114,6 +119,7 @@ pub struct CreatedIdMessage { pub struct DeleteItemMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, } @@ -121,6 +127,7 @@ pub struct DeleteItemMessage { pub struct SubItemMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, } @@ -128,6 +135,7 @@ pub struct SubItemMessage { pub struct UnsubItemMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, } @@ -135,6 +143,7 @@ pub struct UnsubItemMessage { pub struct UpdateRequestMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, #[serde(flatten)] delta: Delta, @@ -144,6 +153,7 @@ pub struct UpdateRequestMessage { pub struct UpdateMessage { //pub cat: String, //cmd: String, + #[serde(flatten)] id: MizeId, #[serde(flatten)] delta: Delta, @@ -156,14 +166,43 @@ pub struct Delta { type Path = Vec; +type Update = Vec<(MizeId, Delta)>; + #[derive(Debug, Default, Clone)] -pub enum MizeId { - Module{mod_name: String, id: String}, - Upstream(String), +pub struct MizeId { + #[serde(rename="id")] + main: String, +} + +pub enum MizeIdKind<'a>{ + Module{mod_name: &'a str, id: &'a str}, + Upstream(&'a server::Peer), Local(u64), - #[default] - None, +} + +//###//=================================================== +//impls + +impl core::hash::Hash for MizeId { + fn hash(&self, state: &mut H) { + self.main.hash(state); + } +} + +impl MizeId { + pub fn as_string(&self) -> String { + match self { + MizeId::Module { mod_name, id } => { + String::from(mod_name) + id + }, + MizeId::Local(num) => { + format!("{}", num) + }, + MizeId::Upstream(string) => string.to_owned(), + MizeId::None => {"".to_owned()}, + } + } } @@ -192,92 +231,6 @@ impl From for MizeMessage { } } -impl MizeId { - pub fn as_string(&self) -> String { - match self { - MizeId::Module { mod_name, id } => { - String::from(mod_name) + id - }, - MizeId::Local(num) => { - format!("{}", num) - }, - MizeId::Upstream(string) => string.to_owned(), - MizeId::None => {"".to_owned()}, - } - } -} - -impl Serialize for MizeId { - fn serialize(&self, serializer: S) -> Result - where - S: serde::Serializer - { - let id_string = match self { - MizeId::Module { mod_name, id } => {String::from("#") + mod_name + id}, - MizeId::Upstream(id) => id.to_owned(), - MizeId::Local(num) => {format!("{}", num)}, - MizeId::None => {"".to_owned()}, - }; - serializer.serialize_str(&id_string) - } -} - -impl<'de> Deserialize<'de> for MizeId { - fn deserialize(deserializer: D) -> Result - where - D: serde::Deserializer<'de>, - { - struct MizeIdVisitor; - - impl<'de> Visitor<'de> for MizeIdVisitor { - type Value = MizeId; - - fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { - formatter.write_str("a MizeId id, which is a string that my start with @ or # ") - } - - fn visit_str(self, v: &str) -> Result - where - E: Error, - { - let first_char = match v.chars().nth(0) { - Some(ch) => ch, - None => { - return Err(de::Error::invalid_value( - de::Unexpected::Other("hi, what is this Unexpected for????"), &"The id was empty")); - }, - }; - - if (first_char == '#'){ - let iter = v.chars(); - - //scip the # - let mod_name: String = iter.clone().take_while(|ch| ch != &'#' || ch != &'/').collect(); - let id: String = iter.collect(); - return Ok(MizeId::Module { mod_name, id}); - } else { - let id: u64 = match v.parse(){ - Ok(num) => num, - Err(_) => { - return Err(de::Error::invalid_value(de::Unexpected::Other("hi, what is this Unexpected for????"), &"Could not parse the id to an u64")); - }, - }; - return Ok(MizeId::Local(id)); - } - } - -// fn visit_map(self, mut map: A) -> Result -// where -// A: serde::de::MapAccess<'de>, -// { - -// } - } - - deserializer.deserialize_str(MizeIdVisitor {}) - } -} - //###//=================================================== // functions @@ -339,8 +292,22 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) - } - JsonMessage::UpdateRequest(msg) => { - handle_update(msg.id, msg.delta, mutexes.clone(), Some(origin)).await?; + JsonMessage::UpdateRequest(mut msg) => { + let update: Update = vec![(msg.id, msg.delta)]; + let new_update = handle_update(update , mutexes.clone(), Some(origin)).await?; + + let subs = mutexes.subs.lock().await; + + for (id, delta) in new_update { + //send that delta to every origin that is subbed the item + let peers = subs.get(&id).ok_or(MizeError::new(11))?; + let message = UpdateMessage{id, delta}; + + for peer in peers { + //TODO: send should only take a reference + peer.send(message.clone()) + } + } }, @@ -374,17 +341,19 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) - } -pub async fn handle_update(id: MizeId, delta: Delta, mutexes: Mutexes, origin: Option) -> Result<(), MizeError>{ +pub async fn handle_update(update: Update, mutexes: Mutexes, origin: Option) -> Result { //TODO: run update code from modules and types - let itemstore = mutexes.itemstore.lock().await; - if let MizeId::Local(id) = id { - itemstore.update(id, delta).await?; - } else { - return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet")); + + for (id, delta) in update { + if let MizeId::Local(id) = id { + itemstore.update(id, delta).await?; + } else { + return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet")); + } } - Ok(()) + Ok(update) }