diff --git a/src/error.rs b/src/error.rs index 96419f3..b1dd554 100644 --- a/src/error.rs +++ b/src/error.rs @@ -216,20 +216,35 @@ impl MizeResultExtension for Result { } impl From for MizeError { + #[track_caller] fn from(utf_err: FromUtf8Error) -> MizeError{ - MizeError::new(109).extra_msg("std::string::FromUtf8Error while parsing something from the message") + let caller_location = std::panic::Location::caller(); + + let mut err = MizeError::new(109).extra_msg(format!("From std::string::FromUtf8Error: {}", utf_err)); + err.code_location = Some(caller_location.into()); + return err; } } impl From for MizeError { + #[track_caller] fn from(sur_err: surrealdb::error::Db) -> MizeError { - MizeError::new(129).extra_msg(format!("surrealdb Error: {}", sur_err)) + let caller_location = std::panic::Location::caller(); + + let mut err = MizeError::new(129).extra_msg(format!("From surrealdb Error: {}", sur_err)); + err.code_location = Some(caller_location.into()); + return err; } } impl From for MizeError { + #[track_caller] fn from(serde_err: serde_json::Error) -> MizeError { - MizeError::new(11).extra_msg("From an serde_json::Error. TODO: include actual error msg.") + let caller_location = std::panic::Location::caller(); + + let mut err = MizeError::new(11).extra_msg(format!("From an serde_json::Error: {}", serde_err)); + err.code_location = Some(caller_location.into()); + return err; } } diff --git a/src/server.rs b/src/server.rs index 574ab8c..d12b21f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -15,6 +15,8 @@ use crate::error::MizeError; use crate::error::ERRORS; use crate::server::proto::MizeMessage; use crate::server::proto::MizeId; +use crate::server::proto::Update; +use crate::server::proto::update_thread; use crate::error::MizeResultExtension; use serde_json::Value as JsonValue; @@ -47,6 +49,7 @@ use tower_http::services::{ServeDir, ServeFile}; //static API_PATH_STRING: &str = "$api"; static SOCKET_CHANNEL_SIZE: usize = 200; +static UPDATE_CHANNEL_SIZE: usize = 200; // max file sizez in the mize/data folder (in bytes) static MAX_INDEX_FILE_SIZE: usize = 1_800_000; @@ -178,6 +181,7 @@ pub struct Mutexes { itemstore: Arc>, mize_folder: String, server_uuid: Uuid, + update_tx: mpsc::Sender<(Update, Peer)>, } impl Mutexes { @@ -191,6 +195,7 @@ impl Mutexes { mize_folder: mutexes.mize_folder.clone(), itemstore: Arc::clone(&mutexes.itemstore), server_uuid: mutexes.server_uuid, + update_tx: mutexes.update_tx.clone(), } } } @@ -217,6 +222,8 @@ pub async fn run_server(args: Vec) { let (server_uuid, itemstore, renders) = init_server(mize_folder.clone()).await.extra_msg("Could not init server").handle().is_critical(); + // the channel to send updates to, that are then handeled in another thread + let (update_tx, update_rx) = mpsc::channel::<(Update, Peer)>(UPDATE_CHANNEL_SIZE); // Collection of all the things, that most functions need let mutexes: Mutexes = Mutexes { @@ -228,8 +235,15 @@ pub async fn run_server(args: Vec) { itemstore: Arc::new(Mutex::new(itemstore)), mize_folder: mize_folder.clone(), server_uuid, + update_tx, }; + // the update handling task + let update_mutexes = mutexes.clone(); + tokio::spawn(async move { + proto::update_thread(update_rx, update_mutexes); + }); + //listen on the local unix socket //local_socket_server(mize_folder); @@ -250,8 +264,6 @@ async fn axum_server(mize_folder: String, mutexes: Mutexes) { //## $api/socket/id //## $api/rest/ - - let mutexes_clone = Mutexes::clone(&mutexes); let serve_client = get_service(ServeDir::new("js-client/src")).handle_error(handle_error); @@ -341,7 +353,7 @@ async fn handle_websocket_connection( let mut msg_rx = ReceiverStream::new(msg_rx); - //my own forward + // my own forward tokio::spawn(async move { while let Some(msg) = msg_rx.next().await { match msg { @@ -364,7 +376,8 @@ async fn handle_websocket_connection( } }); - //check if mize-module header exists + + // check if mize-module header exists let origin = if let Some(mod_header_val) = headers.get("mize-module"){ let mut mods = mutexes.modules.lock().await; @@ -410,7 +423,6 @@ async fn handle_websocket_connection( msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await; continue; }; - match msg { Message::Binary(b) => { println!("Recieved a Binary Message. Those are not implemented yet.") @@ -418,12 +430,13 @@ async fn handle_websocket_connection( Message::Text(text) => { + println!("GOT: {}", text); let json_msg: proto::JsonMessage = serde_json::from_str(&text).expect("error parsing json"); - if let Err(err) = proto::handle_json_msg(json_msg, origin.clone(), mutexes.clone()).await { - let err_msg: MizeMessage = err.handle().into(); - origin.send(err_msg).await; - }; - } + if let Err(err) = proto::handle_json_msg(json_msg, origin.clone(), mutexes.clone()).await { + let err_msg: MizeMessage = err.handle().into(); + origin.send(err_msg).await; + }; + }, Message::Close(_) => { match origin { diff --git a/src/server/itemstore.rs b/src/server/itemstore.rs index 54ac3f7..ef2f6f5 100644 --- a/src/server/itemstore.rs +++ b/src/server/itemstore.rs @@ -7,6 +7,7 @@ use crate::error; use crate::error::MizeError; use crate::server::proto; use crate::server::Mutexes; +use crate::server::Peer; use serde_json::Value as JsonValue; use serde::{Serialize, Deserialize}; @@ -29,7 +30,7 @@ pub struct Item { impl Item { pub fn get_id(&self) -> Result { if let JsonValue::Object(ob) = &self.json{ - if let Some(id_val) = ob.get("mize-id") { + if let Some(id_val) = ob.get("__item__") { if let JsonValue::String(id) = id_val { return Ok(id.to_string()) } else { @@ -119,7 +120,7 @@ impl Itemstore { return Ok(Itemstore {db:ds}); } - pub async fn create(&self, mut item: Item, mutexes: Mutexes) -> Result { + pub async fn create(&self, mut item: Item, mutexes: Mutexes, origin: Peer) -> Result { let mut tr = self.db.transaction(true, false).await?; //get next_free_id @@ -135,7 +136,9 @@ impl Itemstore { let new_num = num_of_items + 1; //add mize-id field to item - let delta: Delta = serde_json::from_str(&format!("[[[\"mize-id\"], \"{}\"]]", new_id))?; + //let delta: Delta = serde_json::from_str(&format!("[[[\"__item__\"], \"{}\"]]", new_id))?; + let mut delta = Delta::new(); + delta.append(vec!["__item__"], serde_json::json!(new_id)); item.apply_delta(delta)?; tr.set(new_id.to_be_bytes(), item.json.to_string()).await?; @@ -143,8 +146,11 @@ impl Itemstore { tr.commit().await?; //handle update to item0 - let delta_item0: Delta = serde_json::from_str(&format!("[[[\"num_of_items\"], {}], [[\"next_free_id\"], {}]]", new_num, new_next_id))?; - proto::handle_update(vec![(proto::MizeId::new(format!("0")), delta_item0)], mutexes.clone(), None).await?; + let mut delta_item0: Delta = Delta::new(); + delta_item0.append(vec!["num_of_items"], serde_json::json!(new_num)); + delta_item0.append(vec!["next_free_id"], serde_json::json!(new_next_id)); + + proto::handle_update(vec![(proto::MizeId::new(format!("0")), delta_item0)], origin, mutexes.clone()).await?; return Ok(new_id); } diff --git a/src/server/proto.rs b/src/server/proto.rs index 7c2c7af..1379953 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -160,12 +160,13 @@ pub struct UpdateMessage { #[derive(Clone, Debug, Serialize, Deserialize)] pub struct Delta { + #[serde(flatten)] pub delta: Vec<(Path, JsonValue)>, } -type Path = Vec; +pub type Path = Vec; -type Update = Vec<(MizeId, Delta)>; +pub type Update = Vec<(MizeId, Delta)>; #[derive(Debug, Default, Clone, Serialize, Deserialize, Eq, Hash, PartialEq)] @@ -228,6 +229,17 @@ impl MizeId { } +impl Delta { + pub fn new() -> Delta{ + Delta { delta: Vec::new() } + } + + 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)); + } +} + //###//=================================================== //From trait implementations, to be able to treat everything as a MizeMessage that can be @@ -324,29 +336,29 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) - JsonMessage::UpdateRequest(mut msg) => { let update: Update = vec![(msg.id, msg.delta)]; - let new_update = handle_update(update , mutexes.clone(), Some(origin)).await?; + handle_update(update, origin, mutexes.clone()).await?; - let subs = mutexes.subs.lock().await; +// let subs = mutexes.subs.lock().await; - for (id, delta) in new_update { +// for (id, delta) in new_update { //send that delta to every origin that is subbed the item - let peers = match subs.get(&id) { - Some(vec) => vec, - None => continue, - }; - let message = UpdateMessage{id, delta}; +// let peers = match subs.get(&id) { +// Some(vec) => vec, +// None => continue, +// }; +// let message = UpdateMessage{id, delta}; - for peer in peers { +// for peer in peers { //TODO: send should only take a reference - peer.send(message.clone()).await - } - } +// peer.send(message.clone()).await +// } +// } }, JsonMessage::Create(msg) => { let itemstore = mutexes.itemstore.lock().await; - let id = itemstore.create(msg.item, mutexes.clone()).await?; + let id = itemstore.create(msg.item, mutexes.clone(), origin.clone()).await?; let response = CreatedIdMessage {id: MizeId::new(format!("{}", id))}; origin.send(response).await; @@ -373,20 +385,25 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) - return Ok(()) } - -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; - - for (id, delta) in update.clone() { - if let MizeIdKind::Local(id) = id.kind()? { - itemstore.update(id, delta).await?; - } else { - return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet")); - } - } - - Ok(update) +pub async fn handle_update(update: Update, origin: Peer, mutexes: Mutexes) -> Result<(), MizeError> { + mutexes.update_tx.send((update, origin)); + Ok(()) +} + +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 { + //TODO: run update code from modules and types + + let itemstore = mutexes.itemstore.lock().await; + + for (id, delta) in update { + if let MizeIdKind::Local(id) = id.kind()? { + itemstore.update(id, delta).await?; + } else { + return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet")); + } + } + } + Ok(()) }