From 9825c5242b7a71bbd119e83af95b2f77af16571f Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Sun, 12 Feb 2023 23:25:50 +0100 Subject: [PATCH] basic JSON message handling. not tested yet at all --- Cargo.lock | 57 +- Cargo.toml | 5 +- errors.toml | 51 +- src/error.rs | 173 ++++-- src/server.rs | 159 ++++-- src/server/itemstore.rs | 272 ++++----- src/server/proto.rs | 1154 ++++++++++++--------------------------- 7 files changed, 790 insertions(+), 1081 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 174250b..4b627f6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -664,6 +664,17 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "colored" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3616f750b84d8f0de8a58bda93e08e2a81ad3f523089b05f1dffecab48c6cbd" +dependencies = [ + "atty", + "lazy_static", + "winapi", +] + [[package]] name = "concurrent-queue" version = "1.2.4" @@ -679,6 +690,12 @@ version = "0.4.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fbdcdcb6d86f71c5e97409ad45898af11cbc995b4ee8112d59095a28d376c935" +[[package]] +name = "convert_case" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e" + [[package]] name = "cookie" version = "0.14.4" @@ -899,6 +916,19 @@ dependencies = [ "syn", ] +[[package]] +name = "derive_more" +version = "0.99.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fb810d30a7c1953f91334de7244731fc3f3c10d7fe163338a35b9f640960321" +dependencies = [ + "convert_case", + "proc-macro2", + "quote", + "rustc_version 0.4.0", + "syn", +] + [[package]] name = "deunicode" version = "1.3.2" @@ -1617,12 +1647,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "json" -version = "0.12.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "078e285eafdfb6c4b434e0d31e8cfcb5115b651496faca5749b88fafd4f23bfd" - [[package]] name = "kv-log-macro" version = "1.0.7" @@ -1803,10 +1827,13 @@ dependencies = [ "axum", "axum-extra", "axum-macros", + "colored", + "derive_more", "futures-util", "itertools", - "json", "lazy_static", + "serde", + "serde_json", "surrealdb", "tokio", "tokio-stream", @@ -2476,18 +2503,18 @@ checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3" [[package]] name = "serde" -version = "1.0.145" +version = "1.0.152" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b" +checksum = "bb7d1f0d3021d347a83e556fc4683dea2ea09d87bccdf88ff5c12545d89d5efb" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.145" +version = "1.0.152" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81fa1584d3d1bcacd84c277a0dfe21f5b0f6accf4a23d04d4c6d61f1af522b4c" +checksum = "af487d118eecd09402d70a5d72551860e788df87b464af30e5ea6a38c75c541e" dependencies = [ "proc-macro2", "quote", @@ -2496,9 +2523,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.85" +version = "1.0.91" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44" +checksum = "877c235533714907a8c2464236f5c4b2a17262ef1bd71f38f35ea592c8da6883" dependencies = [ "itoa", "ryu", @@ -2845,9 +2872,9 @@ dependencies = [ [[package]] name = "syn" -version = "1.0.102" +version = "1.0.107" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3fcd952facd492f9be3ef0d0b7032a6e442ee9b361d4acc2b1d0c4aaa5f613a1" +checksum = "1f4064b5b16e03ae50984a5a8ed5d4f8803e6bc1fd170a3cda91a1be4b18e3f5" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 93eb04f..7b78492 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,10 +9,13 @@ edition = "2021" axum = { version = "0.6.1", features = ["ws", "headers"] } axum-extra = { version = "0.4.2", features = ["spa"] } axum-macros = "0.3.0" +colored = "2.0.0" +derive_more = "0.99.17" futures-util = "0.3.24" itertools = "0.10.5" -json = "0.12.4" lazy_static = "1.4.0" +serde = "1.0.152" +serde_json = "1.0.91" surrealdb = "1.0.0-beta.7" tokio = {version = "1.21.0", features = ["full"]} tokio-stream = "0.1.9" diff --git a/errors.toml b/errors.toml index ed54f65..66fcdcf 100644 --- a/errors.toml +++ b/errors.toml @@ -2,27 +2,42 @@ [[error]] code = 11 category = "don't know yet" -message = "Something went wrong. Someone was too lazy to define an exact error for that." +message = "Something went wrong. //TODO: give this error it's own category and code." + +[[error]] +category = "data_storage" +code = 28 +message = "Something went wrong in the data storage." + +[[error]] +code = 29 +category = "data_storage:surrealdb_error" +message = "Something went wrong internally with the data-storage (surrealdb/rocksdb): {}" + +[[error]] +code = 30 +category = "data_storage:surrealdb_error" +message = "Could not create data_storeage at {}" + +[[error]] +code = 31 +category = "data_storage:surrealdb_error" +message = "No item with id 0. There definetly should be one" + +[[error]] +code = 32 +category = "data_storage" +message = "item {} was not found in storage" [[error]] code = 100 category = "don't know yet" message = "You can't delete item number 0. This item contains mandatory Config for the Server." -[[error]] -category = "data_storage" -code = 128 -message = "Something went wrong in the data storage." - -[[error]] -code = 129 -category = "data_storage:surrealdb_error" -message = "Something went wrong internally with the data-storage (surrealdb): {}" - [[error]] code = 101 category = "key_missing::item0::num_of_items" -messate = "num_of_items missing in item0" +message = "num_of_items missing in item0" [[error]] code = 102 @@ -30,12 +45,12 @@ category = "key_missing::item0::next_free_id" message = "next_free_id missing in item0" [[error]] -_commit missing in item0: +message = "_commit missing in item0" code = 103 category = "key_missing::item0::_commit" [[error]] -_commit missing: +message = "_commit missing" code = 103 category = "key_missing::_commit" @@ -80,7 +95,13 @@ category = "don't know yet" message = "module header not a String" [[error]] -code 114 +code = 114 category = "error with error system" message = "the error with code {} was not found in the errors.toml file" +[[error]] +code = 115 +category = "WebSocket Error" +message = "Error reviecing a message from the WebSocket" + + diff --git a/src/error.rs b/src/error.rs index 61a99d2..4f49154 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,72 +1,171 @@ use std::collections::HashMap; use std::string::FromUtf8Error; +use toml::Value; +use colored::Colorize; + +use serde::{Serialize, Deserialize}; +use serde_json::Value as JsonValue; +use toml; + +use crate::server::proto::{self, JsonMessage}; +use crate::error::proto::ErrMessage; +use crate::server::proto::MizeMessage; +use crate::server::Peer; lazy_static::lazy_static! { - static ref ERRORS: HashMap = parse_errors(); + pub static ref ERRORS: HashMap = parse_errors(); } fn parse_errors() -> HashMap{ - let toml_string = include_str!("../errors.toml"); + let mut map = HashMap::new(); + let binding = include_str!("../errors.toml") + .parse::() + .expect("Error pasting errors.toml file to Toml."); + + let toml_error = binding.get("error") + .expect("Error getting error table from errors.toml file for Error handling"); + + if let Value::Array(error_arr) = toml_error { + for err in error_arr { + + let code: u32 = if let Value::Integer(code) = err.get("code") + .expect("some error object in errors.toml has no code field.") { + u32::try_from(*code).expect(&format!("Error converting the error code {} &i64 to u32", code)) + } else { + panic!("code value exists in errors.toml, but is not an Integer") + }; + + let message = if let Value::String(message) = err.get("message") + .unwrap_or(&Value::String("".to_string())) {message.to_string()} else { + panic!("message value exists in errors.toml on error obect with code {}, but is not a String", code) + }; + + let category = if let Value::String(category) = err.get("category") + .expect(&format!("the error object in errors.toml with code {} has no category field", code)) {category.to_string()} else { + panic!("category value exists in errors.toml on error obect with code {}, but is not a String", code) + }; + + let error = MizeError{code, message, category, extra_msg: None, caused_by_msg: None, code_location: None}; + + map.insert(code, error); + } + } else { + panic!("errors.toml File wrong. error path is not an array") + } + + return map; } -use json; -use toml; -use crate::server::proto; - -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct MizeError { pub category: String, pub code: u32, pub message: String, pub extra_msg: Option, + pub caused_by_msg: Option, + pub code_location: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CausedByMessage { + client_id: u64, + msg: Box, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MizeCodeLocation { + file: String, + line: u32, + column: u32, +} + +impl From<&std::panic::Location <'_>> for MizeCodeLocation { + fn from(panic_location: &std::panic::Location) -> MizeCodeLocation { + let file = panic_location.file().to_string(); + let line = panic_location.line(); + let column = panic_location.column(); + MizeCodeLocation {file, line, column} + } } impl MizeError { + #[track_caller] pub fn new(code: u32) -> MizeError { - //let error = ERRORS. + let caller_location = std::panic::Location::caller(); MizeError { category: "error with error system".to_string(), code: 114, - message: "The error with code {} was not found in the errors.toml file".to_string(), - extra_msg: None + message: "The error with code {} was not found in the Errors that where imported from the errors.toml file at build time.".to_string(), + extra_msg: None, + caused_by_msg: None, + code_location: Some(caller_location.into()), } } -// pub fn format(&self, values: [str]){ -// } - - pub fn extra_msg(mut self, msg: &str) -> MizeError{ + pub fn extra_msg(mut self, msg: &str) -> MizeError { self.extra_msg = Some(msg.to_string()); return self; } + + pub fn format(mut self, values: Vec<&str>) -> MizeError { + let count = 0; + for val in values { + self.message += &(" FORMAT_".to_string() + &format!("{}", count) + ": " + val) + } + return self; + } - pub fn to_json(self: MizeError) -> json::JsonValue{ - if let Some(extra_msg) = self.extra_msg { - json::object!{ - category: self.category, - code: self.code, - message: self.message, - extra_msg: extra_msg, + fn set_msg(self, msg: MizeMessage, origin: Peer) -> MizeError { + MizeError::new(11).extra_msg("in set_msg TODO!!") + } + + fn send(self) -> MizeError{ + self + } + + fn add_to_err_item(self) -> MizeError { + self + } + + fn write_to_stderr(self) -> MizeError { + eprintln!("[{}] {}", "ERROR".red(), self.to_json()); + self + } + + pub fn handle(self) -> MizeError { + return self.send().add_to_err_item().write_to_stderr(); + } + + pub fn to_json(&self) -> JsonValue{ + if let Ok(string) = serde_json::to_string(self) { + if let Ok(json) = serde_json::from_str(&string) { + return json; + } else { + return serde_json::json!({ + "category": "error with error system", + "code": 114, + "message": "error converting a String to a JsonValue for the ErrMsg", + }) } } else { - json::object!{ - category: self.category, - code: self.code, - message: self.message, - } + return serde_json::json!({ + "category": "error with error system", + "code": 114, + "message": "error converting an MizeError to String", + }) } } - pub fn to_message(self: MizeError, origin: proto::Origin) -> proto::Message { - let data = self.to_json(); - let mut msg = vec![crate::PROTO_VERSION, proto::MSG_ERROR]; - msg.extend(json::stringify(data).into_bytes()); - return proto::Message::from_bytes(msg, origin); + pub fn to_json_message(self) -> JsonMessage { + proto::JsonMessage::ErrMsg( + ErrMessage{ + err: self, + } + ) } - } impl From for MizeError { @@ -76,7 +175,15 @@ impl From for MizeError { } impl From for MizeError { - fn from(sur_err: surrealdb::Error) -> MizeError{ + fn from(sur_err: surrealdb::Error) -> MizeError { MizeError::new(129) } } + +impl From for MizeError { + fn from(serde_err: serde_json::Error) -> MizeError { + MizeError::new(11).extra_msg("From an serde_json::Error. TODO: include actual error msg.") + } +} + + diff --git a/src/server.rs b/src/server.rs index 1c632e3..f774f4b 100644 --- a/src/server.rs +++ b/src/server.rs @@ -2,6 +2,7 @@ pub mod server_utils; pub mod itemstore; + pub mod proto; //use futures_util::lock::Mutex; @@ -11,6 +12,11 @@ use std::boxed::Box; use std::path::Path; use crate::error; use crate::error::MizeError; +use crate::error::ERRORS; +use crate::server::proto::MizeMessage; + +use serde_json::Value as JsonValue; +use serde::{Serialize, Deserialize}; use tokio::sync::mpsc::{Sender, channel, Receiver}; use tokio_stream::wrappers::ReceiverStream; @@ -19,7 +25,6 @@ use std::fs; use std::sync::Arc; use tokio::sync::Mutex; use tokio::sync::mpsc; -use crate::server::proto::handle_mize_message; use self::itemstore::Itemstore; @@ -69,17 +74,42 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000; //Version: 0 //"; +#[derive(Clone, Debug)] +pub enum Peer { + Client(Client), //the client id + Module(Module), //Module_name + Upstream(String), //a hostname Type, but for now just a string +} + +impl Peer { + pub async fn send (&self, message: T) where T: Into{ + let message: proto::MizeMessage = message.into(); + match self { + Peer::Client(client) => {client.tx.send(message).await;}, + Peer::Module(module) => {module.tx.send(message).await;}, + Peer::Upstream(_) => {}, + } + } + pub fn get_id(&self) -> u64{ + match self { + Peer::Client(client) => client.id, + Peer::Module(module) => module.client_id, + Peer::Upstream(_) => 0, + } + } +} + #[derive(Clone, Debug)] pub struct Client { // rx: UnboundedReceiverStream>, - tx: Sender, + tx: Sender, id: u64, } #[derive(Clone, Debug)] pub struct Module { - tx: Sender, + tx: Sender, name: String, client_id: u64, kind: ModuleKind, @@ -88,9 +118,10 @@ pub struct Module { #[derive(Clone, Debug)] pub enum ModuleKind { - Binary(), + Rust(), Python(), - Node(), + JS(), + Lua(), //connects to ws server Extern(), @@ -104,6 +135,16 @@ pub struct Render { folder: String, } +trait SendMize where T: Into { + fn send(&self, msg: T); +} + +impl SendMize for Sender where MizeMessage: From { + fn send(&self, msg: T){ + self.send(msg.into()); + } +} + //pub struct Upstream maybe???? ... I'd say later.... @@ -113,7 +154,7 @@ 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>, @@ -134,10 +175,6 @@ impl Mutexes { } } -fn test(){ - println!("test func"); -} - #[tokio::main] pub async fn run_server(args: Vec) { /* @@ -157,18 +194,11 @@ pub async fn run_server(args: Vec) { } } - //load the errors.toml file and init the error system - //let error_toml_file = include_str!("../errors.toml"); - //println!("ERRORS: {:?}", crate::error::ERRORS); - error::init(); - println!("hello in run_server"); - test(); - //create the itemstore let itemstore = crate::server::itemstore::Itemstore::new(mize_folder.clone() + "/db").await.expect("error creating itemstore"); //load modules and renders - let ren_mods = std::fs::read_dir(mize_folder.clone() + "/modules-renders") + let ren_mods = std::fs::read_dir(mize_folder.clone() + "/mr") .expect("error reading the modules-renders dir in the mize-folder") .filter(|entry| entry.as_ref().unwrap().file_type().unwrap().is_dir()); @@ -318,14 +348,30 @@ async fn handle_websocket_connection( headers: HeaderMap, ){ let (mut socket_tx, mut socket_rx) = socket.split(); - let (msg_tx, msg_rx): (Sender, Receiver) = mpsc::channel(SOCKET_CHANNEL_SIZE); + let (msg_tx, msg_rx): (Sender, Receiver) = mpsc::channel(SOCKET_CHANNEL_SIZE); let mut msg_rx = ReceiverStream::new(msg_rx); //my own forward tokio::spawn(async move { while let Some(msg) = msg_rx.next().await { - socket_tx.send(Message::Binary(msg.raw)).await; + match msg { + proto::MizeMessage::Json(json_msg) => { + if let Ok(msg) = serde_json::to_string(&json_msg) { + socket_tx.send(Message::Text(msg)).await; + } else { + let err: MizeError = MizeError::new(11).extra_msg("error while serializing a json message").handle(); + if let Ok(text) = serde_json::to_string(&err.to_json()){ + socket_tx.send(Message::Text(text)); + } else { + MizeError::new(11).extra_msg("not even being able to serialize an error, while failing to serializing a message").handle(); + } + } + } + proto::MizeMessage::Bin(bin_msg) => { + socket_tx.send(Message::Binary(bin_msg.raw)).await; + } + } } }); @@ -350,13 +396,13 @@ async fn handle_websocket_connection( let err = MizeError::new(113) .extra_msg("The mize-module Header could not be decoded into a String."); - msg_tx.send(err.to_message(proto::Origin::Module(module.clone()))).await; + msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await; return; }; println!("module {} connected", module.name); mods.push(module.clone()); drop(mods); - proto::Origin::Module(module) + Peer::Module(module) } else { let mut cli = mutexes.clients.lock().await; @@ -365,53 +411,82 @@ async fn handle_websocket_connection( cli.push(client.clone()); *client_id += 1; drop(cli); - proto::Origin::Client(client) + Peer::Client(client) }; - // Reading and broadcasting messages + // Reading messages while let Some(result) = socket_rx.next().await { - let msg = result.expect("Error when getting message from WebSocket"); - println!("got message: {:?}", msg); + let msg = if let Ok(msg) = result {msg} else { + let err = MizeError::new(115).handle(); + 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.") + }, + + + Message::Text(text) => { + let json_msg: proto::JsonMessage = serde_json::from_str(&text).expect("error parsing json"); + match json_msg { + proto::JsonMessage::ForItem(item_msg) => { + if let Err(err) = proto::handle_item_msg(item_msg, origin.clone(), mutexes.clone()).await { + let err_msg: MizeMessage = err.handle().into(); + origin.send(err_msg).await; + }; + }, + _ => {println!("Got a non ItemMessage")}, + } + + } + - let bytes = match msg { - Message::Binary(b) => b, Message::Close(_) => { match origin { - proto::Origin::Client(client) => { + Peer::Client(ref client) => { //remove client from client list let mut clients = mutexes.clients.lock().await; - let index = clients.iter().position(|client_iter| client_iter.id == client.id).expect("A Client disconected, that had never connected......"); + let index = match clients.iter() + .position(|client_iter| client_iter.id == client.id) + .ok_or(MizeError::new(11).extra_msg("A Client disconected, that had never connected......")){ + Ok(index) => index, + Err(err) => { + let msg: MizeMessage = err.into(); + origin.send(msg); + return; + }, + }; clients.remove(index); println!("Close from Client"); return; }, - proto::Origin::Module(module) => { + Peer::Module(module) => { //remove module from module list let mut modules = mutexes.modules.lock().await; - let index = modules.iter().position(|module_iter| module_iter.client_id == module.client_id).expect("A Module disconected, that had never connected....."); + let index = modules.iter() + .position(|module_iter| module_iter.client_id == module.client_id) + .expect("A Module disconected, that had never connected....."); modules.remove(index); println!("Close from Module"); return; } - proto::Origin::Upstream(_) => { + Peer::Upstream(_) => { return; } } } + + _ => { let err = MizeError::new(11) - .extra_msg("the message type was not Binary"); + .extra_msg("unhandeld WebSocket-Message type") + .handle(); - msg_tx.send(err.to_message(origin.clone())).await; - vec![0] + msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await; }, }; - - if let Err(err) = handle_mize_message( - proto::Message::from_bytes(bytes, origin.clone()), mutexes.clone(), - ).await { - msg_tx.send(err.to_message(origin.clone())).await; - }; }; } diff --git a/src/server/itemstore.rs b/src/server/itemstore.rs index 66b5923..c88f2b0 100644 --- a/src/server/itemstore.rs +++ b/src/server/itemstore.rs @@ -6,6 +6,10 @@ use itertools::Itertools; use crate::error; use crate::error::MizeError; use crate::server::proto; +use crate::server::Mutexes; + +use serde_json::Value as JsonValue; +use serde::{Serialize, Deserialize}; use super::proto::Delta; @@ -16,37 +20,72 @@ pub struct Itemstore { db: surrealdb::Datastore, } -pub type Item = Vec<[Vec; 2]>; +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Item { + #[serde(flatten)] + pub json: JsonValue, +} + +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 JsonValue::String(id) = id_val { + return Ok(id.to_string()) + } else { + return Err(MizeError::new(11).extra_msg("mize-id field in an item was not a String")); + } + } else { + return Err(MizeError::new(11).extra_msg("There was no mize-id field in an item")); + } + } else { + return Err(MizeError::new(11).extra_msg("An Item was not a JsonValue::Object")); + } + } + + pub fn apply_delta(&mut self, delta: Delta) -> Result<(), MizeError> { + for (path, value) in delta.raw { + let mut to_replace = &mut self.json; + for name in path { + to_replace = if to_replace.get(&name).is_none() { + return Err(MizeError::new(11) + .extra_msg("Error applying a Delta to an Item. The path could not be followed.")); + } else { + to_replace.get_mut(&name).ok_or(MizeError::new(11))? + }; + } + *to_replace = value; + } + + return Ok(()) + } + + pub fn from_bytes(vec: Vec) -> Result { + let string = String::from_utf8(vec)?; + let item = serde_json::from_str(&string)?; + return Ok(item); + } +} impl Itemstore { pub async fn new(path: String) -> Result { - let ds = Datastore::new(&(String::from("file://") + &path[..] + "/db")[..]).await.expect("Could not open database"); + let ds = Datastore::new(&(String::from("file://") + &path[..] + "/db")[..]).await + .map_err(|_| MizeError::new(30).format(vec![&path]))?; let mut tr = ds.transaction(true, false).await?; - match tr.get(vec!('0' as u8)).await { + match tr.get(0u64.to_le_bytes()).await { Ok(Some(val)) => println!("item 0 already created"), Ok(None) => { - let keys = vec![ - "num_of_items".as_bytes().to_vec(), - "next_free_id".as_bytes().to_vec(), - "_commit".as_bytes().to_vec(), - "_type".as_bytes().to_vec(), - //"renders".as_bytes().to_vec(), - //"modules".as_bytes().to_vec(), - ]; - - tr.set("0".to_string().into_bytes(), encode(keys)).await?; - let one: u64 = 1; - let zero: u64 = 0; - - tr.set("0:num_of_items".to_string().into_bytes(), one.to_be_bytes()).await?; - tr.set("0:next_free_id".to_string().into_bytes(), one.to_be_bytes()).await?; - tr.set("0:_commit".to_string().into_bytes(), zero.to_be_bytes()).await?; - tr.set("0:_type".to_string().into_bytes(), "mize-main".as_bytes().to_vec()).await?; - //tr.set("0:renders".to_string().into_bytes(), Vec::new()); - //tr.set("0:modules".to_string().into_bytes(), Vec::new()); + let data = serde_json::json!({ + "num_of_items": 0, + "next_free_id": 0, + "_commit": 0, + "mize-type": "mize-main", + "mize-id": "0" + }); + tr.set(0u64.to_be_bytes(), data.to_string()).await?; tr.commit().await?; }, @@ -55,83 +94,50 @@ impl Itemstore { return Ok(Itemstore {db:ds}); } - pub async fn create(&self, item: Item) -> Result { - let mut new_id:u64 = 0; - + pub async fn create(&self, mut item: Item, mutexes: Mutexes) -> Result { let mut tr = self.db.transaction(true, false).await?; - let id = tr.get("0:next_free_id").await - .expect("no item with id 0 when there definetly should be one") - .expect("error reading item 0 from db"); - let id_u64 = u64::from_be_bytes(id.clone().try_into().unwrap()); - let mut keys: Vec> = Vec::new(); - //set id:_commit - let key = format!("{}:_commit", id_u64).into_bytes(); - tr.set(key.clone(), vec![0,0,0,0,0,0,0,0]).await?; - keys.push("_commit".to_owned().into_bytes()); + //get next_free_id + let json: JsonValue = serde_json::from_str(&String::from_utf8(tr.get(0u64.to_be_bytes()).await?.ok_or(MizeError::new(31))?) + .map_err(|_| MizeError::new(11))?) + .map_err(|_| MizeError::new(11))?; + let new_id: u64 = json.get("next_free_id").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?; - for field in item { - let mut key = format!("{}:", id_u64).into_bytes(); - key.extend(field[0].clone()); - keys.push(field[0].clone()); + let num_of_items: u64 = json.get("num_of_items").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?; - let val = field[1].clone(); - tr.set(key, val).await?; - } - tr.set(format!("{}", id_u64), encode(keys)).await?; + let new_next_id = new_id +1; + 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))?; + item.apply_delta(delta)?; + + tr.set(new_id.to_be_bytes(), item.json.to_string()).await?; tr.commit().await?; - return Ok(new_id) + + //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(proto::MizeId::Local(0), delta_item0, mutexes.clone(), None).await?; + + return Ok(new_id); } - pub async fn update(&self, update: proto::Update) -> Result<(), MizeError>{ - println!("itemstore.update func"); + + pub async fn update(&self, id: u64, delta: Delta) -> Result<(), MizeError>{ let mut tr = self.db.transaction(true, true).await?; - //read keys - let mut old_keys = match tr.get(format!("{}", update.id).into_bytes()).await? { - Some(keys) => decode(keys), - None => { - return Err(MizeError::new(104) - .extra_msg("Internal Datastorage Error: the Index of the Item was not found")); - }, - }; + let item_bytes: Vec = tr.get(id.to_le_bytes()).await? + .ok_or(MizeError::new(11) + .extra_msg("The Item an update should be applied to does not exist."))?; - //write new values - for (mut key, delta) in update.raw { + let mut item = Item::from_bytes(item_bytes)?; - let mut store_key = (update.id.clone() + ":").into_bytes(); - store_key.extend(key.clone()); + item.apply_delta(delta)?; - let mut new_val = Vec::new(); - - if let Some(old_val) = tr.get(store_key.clone()).await? { - new_val = proto::apply_delta(old_val, delta)?; - tr.set(store_key.clone(), new_val.clone()).await?; - } else { - old_keys.push(key.clone()); - new_val = proto::apply_delta(Vec::new(), delta)?; - tr.set(store_key.clone(), new_val.clone()).await?; - } - - //remove field if new_val is empty - if (new_val.len() == 0) { - if let Some(key_index) = old_keys.iter().position(|x| *x == key.clone()){ - old_keys.remove(key_index); - } - let mut store_key = update.id.clone().into_bytes(); - store_key.push(':' as u8); - store_key.extend(key); - tr.del(store_key).await?; - } else { - tr.set(store_key.clone(), new_val).await?; - } - } - - //write keys again - tr.set(update.id.into_bytes(), encode(old_keys)).await?; + tr.set(id.to_be_bytes(), serde_json::to_string(&item)?).await?; tr.commit().await?; return Ok(()); @@ -139,107 +145,29 @@ impl Itemstore { - pub async fn delete(&self, id_str: String) -> Result<(), error::MizeError>{ - let id_u64: u64 = match id_str.parse() { - Ok(id) => id, - Err(err) => { - return Err(MizeError::new(110) - .extra_msg("std::num::ParseIntError while parsing id \"{}\" into an Integer")); - } - }; + pub async fn delete(&self, id: u64) -> Result<(), error::MizeError>{ let mut tr = self.db.transaction(true, true).await?; - let keys_res = tr.get(format!("{}", id_u64).into_bytes()).await?; - if let Some(keys) = keys_res { - for key in decode(keys) { - tr.del(key).await?; - } - } else { - return Err(MizeError::new(104) - .extra_msg("Internal Datastorage Error: the Index of the Item was not found, \ - This means most likely that this item does not exist.")); - } - - tr.del(id_str.into_bytes()).await?; + tr.del(id.to_be_bytes()).await?; tr.commit().await?; + return Ok(()); } - pub async fn get(&self, id: u64) -> Result { let mut tr = self.db.transaction(false, false).await?; - let mut item: Item = Vec::new(); - let keys_res = tr.get(format!("{}", id).into_bytes()).await?; - if let Some(keys) = keys_res { - for key in decode(keys) { - let mut field: [Vec; 2] = [Vec::new(), Vec::new()]; - field[0] = key.clone(); - let mut ke = format!("{}:", id).into_bytes(); - ke.extend(key.clone()); - field[1] = match tr.get(ke).await { - Ok(Some(val)) => val, - Ok(None) => panic!("a field that should be there wasn't"), - Err(e) => panic!("Error while querying the db: {}", e), - }; - item.push(field); - }; - return Ok(item); + let item_res = tr.get(id.to_le_bytes()).await?; + if let Some(item_vec) = item_res { + let item_str: String = String::from_utf8(item_vec).map_err(|_|MizeError::new(11))?; + let item: JsonValue = serde_json::from_str(&item_str).map_err(|_|MizeError::new(11))?; + return Ok(Item { json: item }) } else { - return Err(MizeError::new(104) - .extra_msg("Internal Datastorage Error: the Index of the Item was not found")); + return Err(MizeError::new(32).format(vec![&format!("{}", id)])); } - return Ok(item); + return Err(MizeError::new(11)); } } -pub fn decode(bytes: Vec) -> Vec> { - let mut count = 0; - let mut index: Vec> = Vec::new(); - let mut key: Vec = Vec::new(); - loop { - if bytes[count] == '$' as u8 { - if key.len() != 0 { - index.push(key.clone()); - key.clear(); - } - } else if bytes[count] == '!' as u8 { - count += 1; - key.push(bytes[count]); - } else { - key.push(bytes[count]); - } - - if count == bytes.len() -1 { - index.push(key.clone()); - } - - count += 1; - if count >= bytes.len() {break}; - } - return index; -} - -pub fn encode(index: Vec>) -> Vec { - let mut bytes: Vec = Vec::new(); - for key in index { - let mut count = 0; - - loop { - if key[count] == '!' as u8 || key[count] == '$' as u8 { - bytes.insert(count, '!' as u8); - count += 1; - } - - count += 1; - if count >= key.len() {break}; - } - bytes.push('$' as u8); - bytes.extend(key); - } - - return bytes; -} - diff --git a/src/server/proto.rs b/src/server/proto.rs index 7de692c..1e48cca 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -1,50 +1,8 @@ - - // 1: get - // 2: give (by default only give the first 1k bytes, and the first 100 fields) - // 3: get_val (to get more than 1k bytes) - // 4: give_val - // 5: unauthorized - // 6: sub - // 7: unsub - // 8: update_request - // 9: update_deny - // 10: update - // 11: delete - // 12: create - // 13: created_id - // 14: unsupported_version - // 15: get_and_sub - // 16: get_fields - // 17: error - // 18: give_fields - - - // ID - // just a number: localy storeWEK_K2JNQqkd - // #hello.mize.works#034.312 upstream in the internet ... domain specified - // !youtube!WEK_K2JNQqk upstream in a module ... if module is not on local server - // forwarded to configured upstream server (would happen, if we are running in daemon mode) - -pub static VERSION: u8 = 1; - -pub static MSG_GET: u8 = 1; -pub static MSG_GIVE: u8 = 2; -pub static MSG_GET_VAL: u8 = 3; -pub static MSG_GIVE_VAL: u8 = 4; -pub static MSG_UNAUTHORIZED: u8 = 5; -pub static MSG_SUB: u8 = 6; -pub static MSG_UNSUB: u8 = 7; -pub static MSG_UPDATE_REQUEST: u8 = 8; -pub static MSG_UPDATE_DENY: u8 = 9; -pub static MSG_UPDATE: u8 = 10; -pub static MSG_DELETE: u8 = 11; -pub static MSG_CREATE: u8 = 12; -pub static MSG_CREATED_ID: u8 = 13; -pub static MSG_UNSUPPORTED_VERSION: u8 = 14; -pub static MSG_GET_AND_SUB: u8 = 15; -pub static MSG_GET_FIELDS: u8 = 16; -pub static MSG_ERROR: u8 = 17; -pub static MSG_GIVE_FIELDS: u8 = 18; +use derive_more::From; +use serde::de::Error; +use serde_json::json; +use serde::de::Visitor; +use serde::de; use std::time::Duration; //use std::panic::update_hook; @@ -53,7 +11,7 @@ use std::collections::HashMap; use crate::server::itemstore; use crate::server::Mutexes; -use crate::server::itemstore::encode; +use crate::server::Peer; use crate::error; use crate::server::Client; use std::sync::Arc; @@ -61,799 +19,389 @@ use futures_util::stream::Collect; use tokio::sync::{Mutex, mpsc}; use crate::server; use crate::error::MizeError; +use crate::server::itemstore::Item; +use crate::server::proto; -#[derive(Clone, Debug)] -pub struct Message { +use serde_json::Value as JsonValue; +use serde::{Serialize, Deserialize}; + +//###//=================================================== +//all the structs and enums +#[derive(From, Debug, Clone)] +pub enum MizeMessage { + Json(JsonMessage), + Bin(BinMessage), +} + + +#[derive(Debug, Clone)] +pub struct BinMessage { pub raw: Vec, - version: u8, - cmd: u8, - origin: Origin, - index: usize, - meta_gotten: bool, - has_meta: bool, - id: Option, } -//pub enum Response { -// One(Vec), -// All(Vec), -// AllSubbed(String, Vec), -// None, -//} - -#[derive(Clone, Debug)] -pub enum Origin { - Client(server::Client), //the client id - Module(server::Module), //Module_name - Upstream(String), //a hostname Type, but for now just a string +#[derive(From, Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "cat")] +pub enum JsonMessage { + #[serde(rename="err")] + ErrMsg(ErrMessage), + #[serde(rename="item")] + ForItem(ItemMessage), + #[serde(rename="arr")] + ForArr(ArrayMessage), } -impl Origin { - pub async fn send(&self, message: Message){ - match self { - Origin::Client(client) => {client.tx.send(message).await;}, - Origin::Module(module) => {module.tx.send(message).await;}, - Origin::Upstream(_) => {}, - } - } - pub fn get_id(&self) -> u64{ - match self { - Origin::Client(client) => client.id, - Origin::Module(module) => module.client_id, - Origin::Upstream(_) => 0, - } - } +#[derive(From, Debug, Clone, Serialize, Deserialize)] +pub struct ErrMessage { + //pub cat: String, + pub err: MizeError, } -#[derive(Clone)] -pub struct Update { - pub id: String, - origin: Origin, - pub raw: Vec<(Vec, Delta)>, +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ArrayMessage { + //for interacting with __arr__ category + //TODO } -#[derive(Clone, Debug)] +//enum variant and a struct for every cmd a good idea??? +#[derive(From, Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "cmd")] +pub enum ItemMessage { + #[serde(rename="get")] + Get(GetItemMessage), + #[serde(rename="get-sub")] + GetSub(GetSubMessage), + #[serde(rename="give")] + Give(GiveItemMessage), + #[serde(rename="create")] + Create(CreateItemMessage), + #[serde(rename="created-id")] + CreatedId(CreatedIdMessage), + #[serde(rename="delete")] + Delete(DeleteItemMessage), + #[serde(rename="sub")] + Sub(SubItemMessage), + #[serde(rename="unsub")] + Unsub(UnsubItemMessage), + #[serde(rename="update-req")] + UpdateRequest(UpdateRequestMessage), + #[serde(rename="update")] + Update(UpdateMessage), +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GetItemMessage { + //pub cat: String, + //cmd: String, + id: MizeId, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GetSubMessage { + //pub cat: String, + //cmd: String, + id: MizeId, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GiveItemMessage { + //pub cat: String, + //cmd: String, + id: MizeId, + item: Item, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CreateItemMessage { + //pub cat: String, + //cmd: String, + id: MizeId, + item: Item, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CreatedIdMessage { + //pub cat: String, + //cmd: String, + id: MizeId, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DeleteItemMessage { + //pub cat: String, + //cmd: String, + id: MizeId, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SubItemMessage { + //pub cat: String, + //cmd: String, + id: MizeId, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UnsubItemMessage { + //pub cat: String, + //cmd: String, + id: MizeId, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UpdateRequestMessage { + //pub cat: String, + //cmd: String, + id: MizeId, + delta: Delta, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct UpdateMessage { + //pub cat: String, + //cmd: String, + id: MizeId, + deltas: Delta, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct Delta { - raw: Vec, + #[serde(flatten)] + pub raw: Vec<(Path, JsonValue)>, } -#[derive(Clone, Debug)] -pub enum DeltaCmd { - Replace(u32, u32, Vec), - Insert(u32, u32, Vec), - Delete(u32, u32), -} +type Path = Vec; -impl Delta { - pub fn from_bytes(bytes: Vec) -> Result { - let out_of_bytes = Err(MizeError::new(107).extra_msg("There were not eneough Bytes in the message to complete a Delta.")); - let mut bytes = bytes.iter(); - let mut delta = Delta{raw: Vec::new()}; - - while true { - let cmd = if let Some(cmd) = bytes.next() { - cmd - } else { - //end of bytes - return Ok(delta); - }; - - match cmd { - 0 => { - let mut start_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; - let start = u32::from_be_bytes(start_bytes); - - let mut stop_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; - let stop = u32::from_be_bytes(stop_bytes); - - let mut val: Vec = Vec::new(); - for i in 0..stop-start {val.push(if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;})}; - - delta.raw.push(DeltaCmd::Replace(start, stop, val)); - }, - 1 => { - let mut start_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; - let start = u32::from_be_bytes(start_bytes); - - let mut stop_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; - let stop = u32::from_be_bytes(stop_bytes); - - let mut val: Vec = Vec::new(); - for i in 0..stop-start {val.push(if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;})}; - - delta.raw.push(DeltaCmd::Insert(start, stop, val)); - }, - 2 => { - let mut start_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; - let start = u32::from_be_bytes(start_bytes); - - let mut stop_bytes: [u8; 4] = [0,0,0,0]; - for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}}; - let stop = u32::from_be_bytes(stop_bytes); - - delta.raw.push(DeltaCmd::Delete(start, stop)); - }, - _ => { - return Err(MizeError::new(107) - .extra_msg(&format!("The update message included a DeltaCmd of {}.\ - Only 0,1,2 (for: replace, insert, delete) are allowed.", cmd))); - }, - } - }; - return Ok(delta); - } -} - -impl Update { - pub async fn change_whole_val(id: String, key: Vec, new_val: Vec, origin: Origin, mutexes: Mutexes) -> Result { - let mut old_val_len: u32 = 0; - let itemstore = mutexes.itemstore.lock().await; - - let id_u64 = match id.parse::(){ - Ok(id) => id, - Err(err) => { - return Err(MizeError::new(11) - .extra_msg("Error while parsing id to u64 (std::num::ParseIntError) in order to get it from the itemstore.\ - (get or get_sub message)")); - } - }; - - let item = itemstore.get(id_u64).await?; - - for field in item { - if (field[0] == key){ - old_val_len = field[1].len() as u32; - } - } - - let delete_all = DeltaCmd::Delete(0, old_val_len); - let replace_all = DeltaCmd::Replace(0, new_val.len() as u32, new_val); - let deltas = Delta{raw: vec![delete_all, replace_all]}; - let update: Update = Update { id, raw: vec![(key, deltas)], origin}; - return Ok(update); - } - - pub fn to_message(self, origin: Origin) -> Message{ - return Message::from_update(self, origin); - } - - pub fn from_message(mut message: Message) -> Result { - - message.index = 2; - message.id = None; - - let id = message.get_id()?; - - let mut update = Update{id, raw: Vec::new(), origin: message.clone().origin}; - - let num_of_updates = message.get_u32(); - - for i in 0..num_of_updates as usize { - let key_len = message.get_u32(); - let key = message.get_bytes(key_len as usize); - - let update_len = message.get_u32(); - let update_bytes = message.get_bytes(update_len as usize); - - //get DeltaCmds - let delta = Delta::from_bytes(update_bytes)?; - - update.raw.push((key, delta)); - } - return Ok(update); - - } -} - -impl Message { - pub fn from_bytes(vec: Vec, origin: Origin) -> Message { - - // this weired version does not panic, while others do for some unknown reason - let mut count = 0; - - let mut cmd: u8 = 0; - let mut version: u8 = 0; - - for i in vec.clone() { - if count == 0 { - version = i; - } - if count == 1 { - cmd = i - } - - count += 1; - - if count == 2 {break}; - }; - - let mut has_meta = false; - if (cmd >= 128) {cmd = cmd - 128; has_meta = true} - - Message { raw: vec, version, cmd, index: 2 , origin, meta_gotten: false, id: None, has_meta} - } - pub async fn send(self, origin: &Origin) { - origin.send(self).await; - } - - pub async fn forward_module(mut self, id: String, mutexes: Mutexes) -> Result<(), MizeError>{ - let mut meta: HashMap = HashMap::new(); - meta.insert("c".to_string(), format!("{}", self.origin.get_id())); - self.set_metadata(meta)?; - - //add 128 to the cmd - self.raw[1] = self.raw[1] + 128; - - - let mut id_iter = id.chars(); - //scip the first "!" - id_iter.next(); - let mod_name = id_iter.take_while(|&ch| ch != '!').collect::(); - - let modules = mutexes.modules.lock().await; - for module in &*modules { - if (module.name == mod_name) { - self.send(&Origin::Module(module.clone())).await; - return Ok(()); - } - } - return Ok(()); - } - - pub fn from_update(update: Update, origin: Origin) -> Message{ - let mut message: Vec = vec![VERSION, MSG_UPDATE]; - - // id - message.extend(update.id.into_bytes()); - message.push("/".to_string().into_bytes()[0]); - - // update_len - let update_len = update.raw.len() as u32; - message.extend(update_len.to_be_bytes()); - - for update in update.raw { - //key_len - let key_len = update.0.len() as u32; - message.extend(key_len.to_be_bytes()); - - // key - message.extend(update.0); - - //delta_len - let mut delta_len: u32 = 0; - - // deltas - let mut delta_part: Vec = Vec::new(); - - // loop - for delta in update.1.raw { - match delta { - DeltaCmd::Replace(start, stop, data) => { - delta_len += 9; //for len of start, stop and cmd - delta_len += data.len() as u32; - delta_part.push(0); - delta_part.extend(start.to_be_bytes()); - delta_part.extend(stop.to_be_bytes()); - delta_part.extend(data) - }, - DeltaCmd::Insert(start, stop, data) => { - delta_len += 9; //for len of start, stop and cmd - delta_len += data.len() as u32; - delta_part.push(1); - delta_part.extend(start.to_be_bytes()); - delta_part.extend(stop.to_be_bytes()); - delta_part.extend(data) - }, - DeltaCmd::Delete(start, stop) => { - delta_len += 9; //for len of start, stop and cmd - // - delta_part.push(2); - delta_part.extend(start.to_be_bytes()); - delta_part.extend(stop.to_be_bytes()); - }, - } - }; - - message.extend(delta_len.to_be_bytes()); - message.extend(delta_part); - - } - return Message::from_bytes(message, origin); - } - - pub fn set_metadata(&mut self, metadata: HashMap) -> Result<(), MizeError> { - let mut meta_vec: Vec = Vec::new(); - meta_vec.push('{' as u8); - - for (mut key, mut val) in metadata.iter(){ - meta_vec.append(&mut key.clone().into_bytes()); - meta_vec.push('=' as u8); - meta_vec.append(&mut val.clone().into_bytes()); - meta_vec.push(';' as u8); - } - //meta_vec.remove(meta_vec.len()); - meta_vec.push('}' as u8); - - self.raw.splice(2..2, meta_vec); - return Ok(()); - } - - pub fn get_metadata(&mut self) -> Result, MizeError> { - if !self.has_meta { - return Err(MizeError::new(108).extra_msg("The message should have medatata, but it hasn't. (cmd <= 128)")) - }; - - let meta_err = MizeError::new(107).extra_msg("Error while parsing the metadata of a message."); - - let mut meta_vec:Vec = Vec::new(); - let mut msg_iter = self.raw[self.index..].iter(); - - if (msg_iter.nth(0).unwrap() != &('{' as u8)){ - return Err(meta_err); - }; - - for ch in msg_iter { - - if (ch == &('}' as u8)){ - let mut map: HashMap = HashMap::new(); - let string = String::from_utf8(meta_vec)?; - let sp: Vec<&str> = string.split(";").collect(); - - for st in sp.clone() { - if (st == ""){continue;} - let mut split = st.split("=").collect::>(); - let key = split.clone().into_iter().nth(0).ok_or(meta_err.clone())?; - let val = split.into_iter().nth(1).ok_or(meta_err.clone())?; - map.insert(key.to_string(), val.to_string()); - } - self.meta_gotten = true; - return Ok(map); - } - meta_vec.push(*ch); - }; - - return Err(MizeError::new(107).extra_msg("Error while parsing the metadata of a message.")) - - } - - pub fn get_id(&mut self) -> Result { - //get the metadata just to increment the index - if let Some(id) = &self.id { - return Ok(id.clone()); - } - if (self.has_meta && !self.meta_gotten){ - self.get_metadata(); - } - let mut ch: u8 = 0; - let mut id = String::new(); - for ch in &self.raw[self.index..] { - - if (*ch == 47 as u8){ - self.index += 1; - self.id = Some(id.clone()); - return Ok(id); - } else { - id.push(*ch as char); - self.index += 1; - } - } - - return Err(MizeError::new(112).extra_msg("There is no / to indicate where the id ends")); - -// let len = self.raw.len(); -// while self.index < 10000 { -// if (self.index == len){ -// return Err(MizeError { -// code: 112, -// kind: "faulty_message".to_string(), -// message: "There is no / to indicate where the id ends".to_string(), -// }) -// } -// let ch: u8 = self.raw[self.index]; -// if (ch == 47) {break}; - -// self.index += 1; -// } - //skip "/" -// self.index += 1; - -// self.id_gotten = true; -// return Ok(id); - } - - pub fn get_u32(&mut self) -> u32 { - let tmp = &self.raw[self.index..self.index + 4]; - let num: u32 = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]); - self.index += 4; - - return num; - } - - pub fn get_u64(&mut self) -> u64 { - let tmp = &self.raw[self.index..self.index + 8]; - let num: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]); - self.index += 8; - - return num; - } - - pub fn get_bytes(&mut self, n: usize) -> Vec{ - let tmp = &self.raw[self.index..self.index + n]; - self.index += n; - - return tmp.to_vec(); - } +#[derive(Debug, Default, Clone)] +pub enum MizeId { + Module{mod_name: String, id: String}, + Upstream(String), + Local(u64), + #[default] + None, } -pub async fn handle_mize_message( - mut message: Message, - mutexes: Mutexes, - ) -> Result<(), MizeError> { +//###//=================================================== +//From trait implementations, to be able to treat everything as a MizeMessage that can be +impl From for MizeMessage { + fn from(give: GiveItemMessage) -> MizeMessage { + let hi: ItemMessage = give.into(); + let two: JsonMessage = hi.into(); + return two.into(); + } +} - //###########################################################// - //get and get_and_sub cmd from client or module - if ((message.cmd == MSG_GET || message.cmd == MSG_GET_AND_SUB) && - (matches!(&message.origin, Origin::Client(_)) || matches!(&message.origin, Origin::Module(_)))) +impl From for MizeMessage { + fn from(err: MizeError) -> MizeMessage { + let hi: ErrMessage = err.into(); + let two: JsonMessage = hi.into(); + return two.into(); + } +} + +impl From for MizeMessage { + fn from(give: CreatedIdMessage) -> MizeMessage { + let hi: ItemMessage = give.into(); + let two: JsonMessage = hi.into(); + return two.into(); + } +} + +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 = message.get_id()?; - let client_id: u64 = match message.origin { - Origin::Client(ref Client) => Client.id, - Origin::Module(ref Module) => Module.client_id, - Origin::Upstream(_) => 0, + 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()}, }; - - //sub to item in case of MSG_GET_AND_SUB - if (message.cmd == MSG_GET_AND_SUB) { - let mut subs = mutexes.subs.lock().await; - if let Some(mut vec_ids) = subs.get_mut(&id) { - vec_ids.push(message.origin.clone()); - } else { - subs.insert(id.clone(), vec![message.origin.clone()]); - }; - drop(subs); - }; - - //forward to module in case the item is from a Module - //append medatata - let first_char = id.chars().nth(0) - .ok_or(MizeError::new(11).extra_msg("Somehow the id is empty. while handling a get or get_sub message."))?; - - if (first_char == '!'){ - message.clone().forward_module(id.clone(), mutexes.clone()).await; - return Ok(()); - } - - //forward to upstream server in case the item is from another Server - if (first_char == '#'){ - return Ok(()); - } - - - // answer: - // u8: version - // u8: cmd (2 for give) - // id: terminated by "/" - // u32: num_of_fields - // as often as num_of_fields: - // u64: key_len - // key_len: key - // u64: val_len - // val_len: val - let mut answer: Vec = vec![VERSION, MSG_GIVE]; - answer.extend(format!("{}/", id).into_bytes()); - - let itemstore = mutexes.itemstore.lock().await; - - let id_u64 = match id.parse::(){ - Ok(id) => id, - Err(err) => { - return Err(MizeError::new(11) - .extra_msg("Error while parsing id to u64 in order to get it from the itemstore.\ - (get or get_sub message)")); - } - }; - - let mut item = match itemstore.get(id_u64).await { - Ok(item) => item, - Err(err) => {return Err(err)}, - }; - - let num_of_fields = item.len() as u32; - answer.extend(num_of_fields.to_be_bytes()); - - for field in item { - let key_len = field[0].len() as u32; - answer.extend(key_len.to_be_bytes()); - answer.extend(field[0].clone()); - - let val_len = field[1].len() as u32; - answer.extend(val_len.to_be_bytes()); - answer.extend(field[1].clone()); - } - - let msg = Message::from_bytes(answer, message.origin.clone()); - message.origin.send(msg).await; - return Ok(()); - }; - - - //###########################################################// - //give from module - if (message.cmd == MSG_GIVE && matches!(&message.origin, Origin::Module(_))){ - let meta = message.get_metadata()?; - let id: String = message.get_id()?; - let client_id_str = meta.get("c") - .ok_or(MizeError::new(11) - .extra_msg("there is no \"c\" key in the metadata of a message gotten from a module"))?; - - let client_id: u64 = client_id_str.parse() - .map_err(|_| MizeError::new(11) - .extra_msg("Error (std::num::ParseIntError) while parsing the \"c\" key \ - in the Message metadata from a give Message from a Module"))?; - - let clients = mutexes.clients.lock().await; - let modules = mutexes.modules.lock().await; - let module: Vec<&server::Module> = modules.iter().filter(|&module| module.client_id == client_id).collect(); - let client: Vec<&Client>= clients.iter().filter(|&client| client.id == client_id).collect(); - - //let origin = if let Some(module) = modules.get(module_position) - - - let origin = if (module.len() == 1) { - Origin::Module(module[0].to_owned()) - } else if (client.len() == 1) { - Origin::Client(client[0].to_owned()) - } else { - return Err(MizeError::new(11)) - }; - - drop(clients); - drop(modules); - - while message.raw[2] != '}' as u8 { - println!("CHAR: {:?}", message.raw[2]); - message.raw.remove(2); - } - message.raw.remove(2); - println!("message raw: {:?}", message.raw); - - message.has_meta = false; - message.raw[1] = message.raw[1] - 128; - - origin.send(message).await; - - - return Ok(()); - }; - - - //###########################################################// - //update from module - if (message.cmd == MSG_UPDATE && matches!(&message.origin, Origin::Module(_))){ - let id: String = message.get_id()?; - - let subs = mutexes.subs.lock().await; - - let empty_vec = &Vec::new(); - let subbed_origins = subs.get(&id).unwrap_or(empty_vec); - for origin in subbed_origins { - message.clone().send(origin).await; - } - return Ok(()); - }; - - - //###########################################################// - //update_request from module and client - if (message.cmd == MSG_UPDATE_REQUEST && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ - - let id: String = message.get_id()?; - let first_char = id.chars().nth(0) - .ok_or(MizeError::new(11) - .extra_msg("Somehow the id is empty. while handling a get or get_sub message."))?; - - if (first_char == '!'){ - message.clone().forward_module(id, mutexes.clone()).await; - } - - let update = Update::from_message(message.clone())?; - handle_update(update, mutexes, message.origin.clone()).await?; - return Ok(()); - }; - - - //###########################################################// - //delete from Client or Module - if (message.cmd == MSG_DELETE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ - let id: String = message.get_id()?; - if id == "0".to_string() { - let err = MizeError::new(100); - return Err(err); - } - return Ok(()); + serializer.serialize_str(&id_string) } - - - //###########################################################// - //create from module and Client - if (message.cmd == MSG_CREATE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ - - let num_of_fields = message.get_u32(); - let mut item: Vec<[Vec; 2]> = Vec::new(); - - let mut index = 6; - - for i in 0..num_of_fields { - let key_len = message.get_u32(); - let key = message.get_bytes(key_len as usize); - - let val_len = message.get_u32(); - let val = message.get_bytes(val_len as usize); - - item.push([key, val]); - } - - let itemstore = mutexes.itemstore.lock().await; - let new_id = itemstore.create(item).await?; - - let mut item0: Vec<[Vec; 2]> = Vec::new(); - match itemstore.get(0).await { - Err(err) => {return Err(err);}, - Ok(item) => {item0 = item;}, - } - println!("got create message"); - - //update item0 - let mut next_free_id: Vec = Vec::new(); - let mut num_of_items: Vec = Vec::new(); - for field in item0 { - if field[0] == "next_free_id".to_string().into_bytes() { - // soooooo ugly - let mut tmp = u64::from_be_bytes(field[1].clone().try_into().unwrap()); - tmp += 1; - next_free_id = tmp.to_be_bytes().to_vec(); - - } - if field[0] == "num_of_items".to_string().into_bytes() { - // soooooo ugly - let mut tmp = u64::from_be_bytes(field[1].clone().try_into().unwrap()); - tmp += 1; - num_of_items = tmp.to_be_bytes().to_vec(); - } - } - - //because Update::change_whole_val trys to lock the itemstore as well - drop(itemstore); - - let update_next_id: Update = Update::change_whole_val("0".to_string(), "next_free_id".to_string().into_bytes(), next_free_id.clone(), message.origin.clone(), mutexes.clone()).await?; - let mut update: Update = Update::change_whole_val("0".to_string(), "num_of_items".to_string().into_bytes(), num_of_items.clone(), message.origin.clone(), mutexes.clone()).await?; - - update.raw.extend(update_next_id.raw); - - handle_update(update, mutexes.clone(), message.origin.clone()).await?; - - //send back MSG_CREATED_ID - let mut created_id_message = vec![VERSION, MSG_CREATED_ID]; - created_id_message.extend(new_id.to_be_bytes()); - message.origin.send(Message::from_bytes(created_id_message, message.origin.clone())).await; - - return Ok(()); - } - - //###########################################################// - //for every other message - println!("Got a Message That did not Trigger any if Statement....: {:?}", message.raw); - return Ok(()); } -pub async fn handle_update(mut update: Update, mutexes: Mutexes, origin: Origin) -> Result<(), MizeError>{ - //handle type-code - //and either call itemstore.update(), send the update to the module or upstream, or spawn another - //update by calling handle_update() - +impl<'de> Deserialize<'de> for MizeId { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + struct MizeIdVisitor {} - ///////////////////////////////// TYPE CODE //////////////////////////////////////// - //can change the update and spawn new updates (to a different item) (which would call the handle_update func again) + impl<'de> Visitor<'de> for MizeIdVisitor { + type Value = MizeId; - //the "type code" for the Null type (code that effects every item) - //eg. increment _commit - let itemstore = mutexes.itemstore.lock().await; + fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + formatter.write_str("Could not deserialize element") + } - let id_u64 = match update.id.clone().parse::(){ - Ok(id) => id, - Err(err) => { - return Err(MizeError::new(11) - .extra_msg("Error while parsing id to u64 in order to get it from the itemstore.\ - (get or get_sub message)")); - } - }; - let item = itemstore.get(id_u64).await?; - let commit_pos = item.iter() - .position(|field| field[0] == "_commit".to_string().into_bytes()) - .ok_or(MizeError::new(103) - .extra_msg(&format!("Item {} has not _commit", id_u64)))?; + 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")); + }, + }; - let cur_commit = u64::from_be_bytes(item[commit_pos][1].clone().try_into() - .map_err(|_|MizeError::new(105))? - ); + if (first_char == '#'){ + let iter = v.chars(); - //because Update::change_whole_val trys to lock the itemstore as well - drop(itemstore); + //scip the # + let mod_name: String = iter.clone().take_while(|ch| ch != &'#' || ch != &'/').collect(); + let id: String = iter.collect(); + println!("in deserialize..... mod_name: --{}-- id: --{}--", mod_name.clone(), id.clone()); + 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>, +// { - update.raw.extend(Update::change_whole_val(update.id.clone(), "_commit".to_string().into_bytes(), (cur_commit +1).to_be_bytes().to_vec(), update.origin.clone(), mutexes.clone()).await?.raw); - - - //run some type-code here -// if let Some(index) = item.iter().position(|&field| field[0] == "_type".to_string().into_bytes()) { -// let type_string = String::from_utf8(item[index][0])?; -// let types = type_string.split(" ").collect::Vec(); -// for &typ[..] in types { -// match typ { -// "test-same" => crate::integrated_types::test_same::main(update, item); // } -// }; -// }; + } + + deserializer.deserialize_map(MizeIdVisitor {}) + } +} + +//###//=================================================== +// functions + +pub async fn handle_item_msg(msg: ItemMessage, origin: Peer, mutexes: Mutexes) -> Result<(), MizeError>{ + + match msg { + + ItemMessage::Get(msg) => { + match msg.id { + //MizeId::Module { mod_name, id } => { + //TODO: get type of module with mod_name .... and then either forward msg or call + //module_get_code + //}, + + MizeId::Local(id) => { + let itemstore = mutexes.itemstore.lock().await; + let response = GiveItemMessage { + id: msg.id, + item: itemstore.get(id).await?, + }; + origin.send(response).await; + }, + + _ => {return Err(MizeError::new(11).extra_msg("id types not implemented"));}, + } + return Ok(()); + }, + + + ItemMessage::GetSub(msg) => { + + //sub to item + let mut subs = mutexes.subs.lock().await; + if let Some(mut vec_ids) = subs.get_mut(&msg.id.as_string()) { + vec_ids.push(origin.clone()); + } else { + subs.insert(msg.id.as_string(), vec![origin.clone()]); + }; + + match msg.id { + //MizeId::Module { mod_name, id } => { + //TODO: get type of module with mod_name .... and then either forward msg or call + //module_get_code + //}, + + MizeId::Local(id) => { + let itemstore = mutexes.itemstore.lock().await; + let response = GiveItemMessage { + id: msg.id, + item: itemstore.get(id).await?, + }; + origin.send(response).await; + }, + + _ => {return Err(MizeError::new(11).extra_msg("id types not implemented"));}, + } + + return Ok(()); + } + + + ItemMessage::UpdateRequest(msg) => { + handle_update(msg.id, msg.delta, mutexes.clone(), Some(origin)).await?; + }, + + + ItemMessage::Create(msg) => { + let itemstore = mutexes.itemstore.lock().await; + let id = itemstore.create(msg.item, mutexes.clone()).await?; + let response = CreatedIdMessage {id: MizeId::Local(id)}; + + origin.send(response).await; + return Ok(()); + }, + + + ItemMessage::Delete(msg) => { + match msg.id { + MizeId::Local(id) => { + let itemstore = mutexes.itemstore.lock().await; + itemstore.delete(id).await?; + return Ok(()); + }, + _ => {return Err(MizeError::new(11).extra_msg("deleteng non Local items not implemented yet"));} + } + }, + + + _ => { + return Err(MizeError::new(11).extra_msg("ItemMessage cmd not handeld")); + }, + }; + return Ok(()) +} + + +pub async fn handle_update(id: MizeId, delta: Delta, mutexes: Mutexes, origin: Option) -> Result<(), MizeError>{ + //TODO: run update code from modules and types - //if update is changing _type, then run type::create(), or type::destroy(), for types that got added or deleted - //apply item in itemstore let itemstore = mutexes.itemstore.lock().await; - itemstore.update(update.clone()).await?; - - //send update to clients/modules - let message = update.clone().to_message(origin.clone()); - let subs = mutexes.subs.lock().await; - - let empty_vec = &Vec::new(); - let subbed_origins = subs.get(&update.id).unwrap_or(empty_vec); - for origin in subbed_origins { - message.clone().send(origin).await; + if let MizeId::Local(id) = id { + itemstore.update(id, delta); + } else { + return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet")); } Ok(()) } -// just like all of this crate, definetly could be done with less clones().... and better error -// handling - -pub fn apply_delta(mut val: Vec, delta: Delta) -> Result, MizeError>{ - - let mut update_iter = delta.raw.iter(); - for delta_cmd in delta.raw.iter() { - match delta_cmd { - //r,start:u32,stop:u32,bytes start..stop - DeltaCmd::Replace(start, stop, bytes) => { - let bytes = (*bytes).clone(); - let start = *start as usize; - let stop = *stop as usize; - let val_len = val.len(); - if (stop > val_len) { - val.extend(vec![0; stop-val_len]) - } - val.splice(start..stop, bytes); - }, - //i,start:u32, stop:u32, bytes stop-start - DeltaCmd::Insert(start, stop, bytes) => { - val.splice((*start as usize)..(*start as usize), (*bytes).clone()); - }, - //d,start:u32,stop:u32 - DeltaCmd::Delete(start, stop) => { - val.splice((*start as usize)..(*stop as usize), Vec::new()); - }, - _ => {panic!("unknown update command")} - } - } - - return Ok(val); -} -