From 170df4aefa714b12f6379d593e3e69433b58e48f Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Wed, 26 Apr 2023 23:56:58 +0200 Subject: [PATCH] deleting a path with an update did not work before, among other things --- .gitignore | 1 + js-client | 2 +- run | 2 +- src/server.rs | 5 +- src/server/itemstore.rs | 26 ++++- src/server/proto.rs | 243 +++++++++++++++++++++++++++++++++++++++- 6 files changed, 265 insertions(+), 14 deletions(-) diff --git a/.gitignore b/.gitignore index cf2c9c2..776ee59 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ /target +/mize-folder # I put testing stuff for this projekt that I don't want to push to github into a folder called gitignore /gitignore diff --git a/js-client b/js-client index 335db6d..53c3b6c 160000 --- a/js-client +++ b/js-client @@ -1 +1 @@ -Subproject commit 335db6d99e0060d3845ab29dccaeacd4545639b3 +Subproject commit 53c3b6cc38c13519ef6c6e50f3144e71192a9b82 diff --git a/run b/run index c8213d0..9dc6b07 100644 --- a/run +++ b/run @@ -1,3 +1,3 @@ #!/bin/bash -cargo run -- --folder=./gitignore/folder run +cargo run -- --folder=./mize-folder run diff --git a/src/server.rs b/src/server.rs index 9fc8b7b..0554a91 100644 --- a/src/server.rs +++ b/src/server.rs @@ -45,7 +45,7 @@ use axum::http::{StatusCode, Uri, Response, self}; use axum::http::header::{HeaderName, HeaderValue, HeaderMap}; use tower_http::services::{ServeDir, ServeFile}; - +static SERVER_PORT: u16 = 9432; //static API_PATH_STRING: &str = "$api"; static SOCKET_CHANNEL_SIZE: usize = 200; @@ -293,7 +293,7 @@ async fn axum_server(mize_folder: String, mutexes: Mutexes) { } drop(renders); - let addr = SocketAddr::from(([0, 0, 0, 0], 3000)); + let addr = SocketAddr::from(([0, 0, 0, 0], SERVER_PORT)); // tracing::debug!("listening on {}", addr); axum::Server::bind(&addr) .serve(app.into_make_service()) @@ -359,6 +359,7 @@ async fn handle_websocket_connection( match msg { proto::MizeMessage::Json(json_msg) => { if let Ok(msg) = serde_json::to_string(&json_msg) { + println!("SEINDING: {}", msg); socket_tx.send(Message::Text(msg)).await; } else { let err: MizeError = MizeError::new(11).extra_msg("error while serializing a json message").handle(); diff --git a/src/server/itemstore.rs b/src/server/itemstore.rs index ef2f6f5..8e362ca 100644 --- a/src/server/itemstore.rs +++ b/src/server/itemstore.rs @@ -5,7 +5,7 @@ use surrealdb::kvs::{Datastore, Key, Val}; use itertools::Itertools; use crate::error; use crate::error::MizeError; -use crate::server::proto; +use crate::server::proto::{self}; use crate::server::Mutexes; use crate::server::Peer; @@ -45,10 +45,12 @@ impl Item { } pub fn apply_delta(&mut self, delta: Delta) -> Result<(), MizeError> { - println!("DELTA: {:?}", delta); + //println!("DELTA: {:?}", delta); for (path, value) in delta.delta { let mut to_replace = &mut self.json; + let mut counter = path.len() as i32; for name in path { + counter -= 1; // if let Some(inner) = to_replace.get(&name) { // to_replace = inner @@ -62,15 +64,29 @@ impl Item { // .extra_msg("Error applying a Delta to an Item. The path could not be followed.")); // } + + //on last iteration + if counter == 0 { + if value.is_none() { + if to_replace.is_object() { + to_replace.as_object_mut().ok_or(MizeError::new(11))?.remove(&name); + return Ok(()) + } + } + } + to_replace = if to_replace.get_mut(&name).is_some() { to_replace.get_mut(&name).ok_or(MizeError::new(11))? + } else if to_replace.is_object() { + // when path does not exist let mut new_map = JsonValue::Object(serde_json::Map::new()); to_replace.as_object_mut().ok_or(MizeError::new(11))?.insert(name.clone(), new_map); to_replace.get_mut(&name).ok_or(MizeError::new(11))? } else { + return Err(MizeError::new(11) .extra_msg("Error applying a Delta to an Item. The path could not be followed.")); }; @@ -78,7 +94,7 @@ impl Item { } // replace with value from the delta - *to_replace = value; + *to_replace = value.into(); } println!("After applying a Delta, {}", self.json); @@ -94,7 +110,7 @@ impl Item { impl Itemstore { pub async fn new(path: String) -> Result { - println!("PATH of DB: {}", path); + //println!("PATH of DB: {}", path); let ds = Datastore::new(&(String::from("file://") + &path[..])[..]).await .map_err(|e| MizeError::new(30).extra_msg(format!("surrealdb Error: {}", e)).extra_msg(format!("Trying to create at Location: {}", path)))?; @@ -168,7 +184,7 @@ impl Itemstore { let mut item = Item::from_bytes(item_bytes)?; item.apply_delta(delta)?; - println!("after applying delta: {:?}", item); + //println!("after applying delta: {:?}", item); tr.set(id.to_be_bytes(), serde_json::to_string(&item)?).await?; diff --git a/src/server/proto.rs b/src/server/proto.rs index 4650f3c..82af32d 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -159,11 +159,46 @@ pub struct UpdateMessage { delta: Delta, } -#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(from = "SerdeDelta")] +#[serde(into = "SerdeDelta")] +#[derive(Clone, Debug, Serialize, Deserialize, Default)] pub struct Delta { - pub delta: Vec<(Path, JsonValue)>, + pub delta: Vec<(Path, Option)>, } +#[derive(Clone, Debug, Serialize, Deserialize, Default)] +pub struct SerdeDelta { + pub delta: Vec, +} + +impl From for Delta { + fn from(serde: SerdeDelta) -> Delta { + let mut delta = Delta {delta: Vec::new()}; + for SerdeDeltaEntry(path, json) in serde.delta { + delta.delta.push((path, json)); + } + return delta; + } +} + +impl From for SerdeDelta { + fn from(delta: Delta) -> SerdeDelta { + let mut serde = SerdeDelta {delta: Vec::new()}; + for (path, json) in delta.delta { + serde.delta.push(SerdeDeltaEntry(path, json)); + } + return serde; + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, Default)] +pub struct SerdeDeltaEntry ( + pub Path, + #[serde(default)] + #[serde(skip_serializing_if = "Option::is_none")] + pub Option, + ); + pub type Path = Vec; pub type Update = Vec<(MizeId, Delta)>; @@ -181,9 +216,194 @@ pub enum MizeIdKind<'a>{ Local(u64), } +//mod serde_delta { +// pub fn serialize(delta: &super::Delta, ser: S) -> Result +// where S: Serializer +// { +// let delta_seq = ser.serialize_seq(delta.len())?; +// for (path, value) in delta { +// let path_seq = ser.serialize_seq(path.len())?; +// for path_el in path { +// path_seq.serialize_element(path_el) +// } +// path_seq.end(); +// tuple_seq = ser.serialize_seq(); +// } +// delta_seq.end() +// } +// pub fn deserialize<'de>(deser: D) -> Result +// where D: Deserializer +// { +// } +//} + //###//=================================================== //impls +//impl Serialize for Delta { +// fn serialize(&self, serializer: S) -> Result +// where +// S: serde::Serializer +// { + +// let delta = self.delta; +// 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 Delta { +// fn deserialize(deserializer: D) -> Result +// where +// D: Deserializer<'de>, +// { +// #[derive(Deserialize)] +// #[serde(field_identifier, rename_all = "lowercase")] +// enum Field { Delta } + + // This part could also be generated independently by: + // + // #[derive(Deserialize)] + // #[serde(field_identifier, rename_all = "lowercase")] + // enum Field { Secs, Nanos } +// impl<'de> Deserialize<'de> for Field { +// fn deserialize(deserializer: D) -> Result +// where +// D: Deserializer<'de>, +// { +// struct FieldVisitor; + +// impl<'de> Visitor<'de> for FieldVisitor { +// type Value = Field; + +// fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { +// formatter.write_str("`secs` or `nanos`") +// } + +// fn visit_str(self, value: &str) -> Result +// where +// E: de::Error, +// { +// match value { +// "secs" => Ok(Field::Secs), +// "nanos" => Ok(Field::Nanos), +// _ => Err(de::Error::unknown_field(value, FIELDS)), +// } +// } +// } + +// deserializer.deserialize_identifier(FieldVisitor) +// } +// } + +// struct DeltaVisitor; + +// impl<'de> Visitor<'de> for DeltaVisitor { +// type Value = Delta; + +// fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { +// formatter.write_str("struct Delta") +// } + +// fn visit_seq(self, mut seq: V) -> Result +// where +// V: SeqAccess<'de>, +// { +// let secs = seq.next_element()? +// .ok_or_else(|| de::Error::invalid_length(0, &self))?; +// let nanos = seq.next_element()? +// .ok_or_else(|| de::Error::invalid_length(1, &self))?; +// Ok(Duration::new(secs, nanos)) +// } + +// fn visit_map(self, mut map: V) -> Result +// where +// V: MapAccess<'de>, +// { +// let mut delta = None; +// while let Some(key) = map.next_key()? { +// match key { +// Field::Delta => { +// if secs.is_some() { +// return Err(de::Error::duplicate_field("delta")); +// } +// secs = Some(map.next_value()?); + // do deserialisation of Vec<(Path, ...)> +// } +// } +// } +// let delta = delta.ok_or_else(|| de::Error::missing_field("delta"))?; + +// Ok(Delta::new(secs, nanos)) +// } +// } + +// const FIELDS: &'static [&'static str] = &["delta"]; +// deserializer.deserialize_struct("Delta", FIELDS, DeltaVisitor) +// } +//} + +//impl<'de> Deserialize<'de> for Delta { +// fn deserialize(deserializer: D) -> Result +// where +// D: serde::Deserializer<'de>, +// { +// struct MizeIdVisitor; + +// impl<'de> Visitor<'de> for Delta { +// type Value = Delta; + +// fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { +// formatter.write_str("a Delta. mize::server::proto::Delta") +// } + +// fn visit_map(self, map: &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_map(DeltaVisitor {}) +// } +//} + impl MizeId { pub fn new(main: String) -> MizeId { MizeId { main } @@ -236,7 +456,7 @@ impl Delta { pub fn append(&mut self, path: Vec<&str>, json: JsonValue){ let new_path: Vec = path.into_iter().map(|e| e.to_owned()).collect(); - self.delta.push((new_path, json)); + self.delta.push((new_path, Some(json))); } } @@ -393,18 +613,31 @@ pub async fn handle_update(update: Update, origin: Peer, mutexes: Mutexes) -> Re pub async fn update_thread(mut update_rx: mpsc::Receiver<(Update, Peer)>, mutexes: Mutexes) -> Result<(), MizeError> { while let Some((update, origin)) = update_rx.recv().await { println!("UPDATE: {:?}", update.iter().next().unwrap().0); - println!("DELTA: {:?}", update.iter().next().unwrap().1); //TODO: run update code from modules and types let itemstore = mutexes.itemstore.lock().await; + let subs = mutexes.subs.lock().await; for (id, delta) in update { + + //update the items in the itemstore if let MizeIdKind::Local(id) = id.kind()? { - itemstore.update(id, delta).await?; + itemstore.update(id.clone(), delta.clone()).await?; } else { return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet")); } + + //send a update msg to subbed clients + let peers = match subs.get(&id) { + Some(hi) => hi, + None => break, + }; + let msg = UpdateMessage{id, delta}; + for peer in peers { + peer.send(msg.clone()).await; + } } + } Ok(()) }