From 188a7b2a7eb41f53d97efe98d172137e13e96035 Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Sat, 14 Jan 2023 23:14:37 +0100 Subject: [PATCH] fields are now deleted when the val is empty .... among other changes --- Cargo.lock | 39 ++++++++++- Cargo.toml | 2 +- errorcodes.yml | 3 + js-client | 2 +- src/error.rs | 2 +- src/server.rs | 92 +++++++++++++++++++++----- src/server/itemstore.rs | 16 ++--- src/server/proto.rs | 140 +++++++++++++++++++++++++++++++--------- 8 files changed, 237 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cdd4400..18aed31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -288,6 +288,7 @@ dependencies = [ "bitflags", "bytes 1.2.1", "futures-util", + "headers", "http", "http-body", "hyper", @@ -1280,6 +1281,31 @@ version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" +[[package]] +name = "headers" +version = "0.3.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3e372db8e5c0d213e0cd0b9be18be2aca3d44cf2fe30a9d46a65581cd454584" +dependencies = [ + "base64", + "bitflags", + "bytes 1.2.1", + "headers-core", + "http", + "httpdate", + "mime", + "sha1 0.10.5", +] + +[[package]] +name = "headers-core" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7f66481bfee273957b1f20485a4ff3362987f85b2c236580d81b4eb7a326429" +dependencies = [ + "http", +] + [[package]] name = "heapless" version = "0.7.16" @@ -2530,6 +2556,17 @@ dependencies = [ "sha1_smol", ] +[[package]] +name = "sha1" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f04293dc80c3993519f2d7f6f511707ee7094fe0c6d3406feb330cdb3540eba3" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest 0.10.5", +] + [[package]] name = "sha1_smol" version = "1.0.0" @@ -2693,7 +2730,7 @@ dependencies = [ "serde", "serde_derive", "serde_json", - "sha1", + "sha1 0.6.1", "syn", ] diff --git a/Cargo.toml b/Cargo.toml index 503cedb..96bdbaf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,7 +6,7 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -axum = { version = "0.6.1", features = ["ws"] } +axum = { version = "0.6.1", features = ["ws", "headers"] } axum-extra = { version = "0.4.2", features = ["spa"] } axum-macros = "0.3.0" futures-util = "0.3.24" diff --git a/errorcodes.yml b/errorcodes.yml index d8507f2..73adefd 100644 --- a/errorcodes.yml +++ b/errorcodes.yml @@ -63,3 +63,6 @@ no / on the end of id: code: 112 category: "faulty_message" +module header not a String: + code: 113 + category: "don't know yet" diff --git a/js-client b/js-client index c9012a1..3287d3b 160000 --- a/js-client +++ b/js-client @@ -1 +1 @@ -Subproject commit c9012a1c11ab99bd8b2a0a62aa8c713b151c17ea +Subproject commit 3287d3badecdb838b4cb10fddca9d5ad0ad81d4b diff --git a/src/error.rs b/src/error.rs index 13b46ac..ffe74f2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -4,7 +4,7 @@ use std::string::FromUtf8Error; use json; use crate::server::proto; -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct MizeError { pub kind: String, pub code: u32, diff --git a/src/server.rs b/src/server.rs index 4928dce..3b8c1dd 100644 --- a/src/server.rs +++ b/src/server.rs @@ -30,6 +30,7 @@ use axum::response::{IntoResponse, Html}; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; use axum::extract::{State, self}; use axum::http::{StatusCode, Uri, Response, self}; +use axum::http::header::{HeaderName, HeaderValue, HeaderMap}; use tower_http::services::{ServeDir, ServeFile}; @@ -68,14 +69,14 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000; //"; -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct Client { // rx: UnboundedReceiverStream>, tx: Sender, id: u64, } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct Module { tx: Sender, name: String, @@ -84,11 +85,14 @@ pub struct Module { registered_types: Vec, } -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum ModuleKind { Binary(), Python(), Node(), + + //connects to ws server + Extern(), } #[derive(Clone, Debug)] @@ -290,14 +294,16 @@ async fn get_render_main(extract::Path(id): extract::Path, State(mutexes async fn websocket_handler( ws: WebSocketUpgrade, - State(mutexes): State, + State(mutexes): State, + headers: HeaderMap, ) -> impl IntoResponse { - ws.on_upgrade(|socket| handle_websocket_connection(socket, mutexes)) + ws.on_upgrade(|socket| handle_websocket_connection(socket, mutexes, headers)) } async fn handle_websocket_connection( socket: WebSocket, mutexes: Mutexes, + headers: HeaderMap, ){ let (mut socket_tx, mut socket_rx) = socket.split(); let (msg_tx, msg_rx): (Sender, Receiver) = mpsc::channel(SOCKET_CHANNEL_SIZE); @@ -311,13 +317,46 @@ async fn handle_websocket_connection( } }); - let mut cli = mutexes.clients.lock().await; - let mut client_id = mutexes.next_free_client_id.lock().await; - let client = Client{tx: msg_tx.clone(), id: *client_id}; - cli.push(client.clone()); - *client_id += 1; - drop(cli); - drop(client_id); + //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; + let mut client_id = mutexes.next_free_client_id.lock().await; + let mut module = Module{ + tx: msg_tx.clone(), + client_id: *client_id, + name: String::from("tmp"), + kind: ModuleKind::Extern(), + registered_types: Vec::new(), + }; + + *client_id += 1; + + if let Ok(name) = mod_header_val.to_str() { + module.name = name.to_string(); + } else { + let err = MizeError{ + code: 113, + kind: "don't know yet".to_string(), + message: "The mize-module Header could not be decoded into a String.".to_string(), + }; + msg_tx.send(err.to_message(proto::Origin::Module(module.clone()))).await; + return; + }; + println!("module {} connected", module.name); + mods.push(module.clone()); + drop(mods); + proto::Origin::Module(module) + + } else { + let mut cli = mutexes.clients.lock().await; + let mut client_id = mutexes.next_free_client_id.lock().await; + let client = Client{tx: msg_tx.clone(), id: *client_id}; + cli.push(client.clone()); + *client_id += 1; + drop(cli); + proto::Origin::Client(client) + }; // Reading and broadcasting messages while let Some(result) = socket_rx.next().await { @@ -326,21 +365,44 @@ async fn handle_websocket_connection( let bytes = match msg { Message::Binary(b) => b, + Message::Close(_) => { + match origin { + proto::Origin::Client(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......"); + clients.remove(index); + println!("Close from Client"); + return; + }, + proto::Origin::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....."); + modules.remove(index); + println!("Close from Module"); + return; + } + proto::Origin::Upstream(_) => { + return; + } + } + } _ => { let err = MizeError{ code: 11, kind: "don't know yet".to_string(), message: "the message type was not Binary".to_string(), }; - msg_tx.send(err.to_message(proto::Origin::Client(client.clone()))).await; + msg_tx.send(err.to_message(origin.clone())).await; vec![0] }, }; if let Err(err) = handle_mize_message( - proto::Message::from_bytes(bytes, proto::Origin::Client(client.clone())), mutexes.clone(), + proto::Message::from_bytes(bytes, origin.clone()), mutexes.clone(), ).await { - msg_tx.send(err.to_message(proto::Origin::Client(client.clone()))).await; + msg_tx.send(err.to_message(origin.clone())).await; }; }; } diff --git a/src/server/itemstore.rs b/src/server/itemstore.rs index 04d7405..c3e4122 100644 --- a/src/server/itemstore.rs +++ b/src/server/itemstore.rs @@ -120,14 +120,14 @@ impl Itemstore { } //remove field if new_val is empty - //if (new_val.len() == 0) { - //if let Some(key_index) = keys.iter().position(|x| *x == key.clone()){ - //keys.remove(key_index); - //} - //tr.del(key).await?; - //} else { - //tr.set(store_key.clone(), new_val).await?; - //} + if (new_val.len() == 0) { + if let Some(key_index) = old_keys.iter().position(|x| *x == key.clone()){ + old_keys.remove(key_index); + } + tr.del(key).await?; + } else { + tr.set(store_key.clone(), new_val).await?; + } } //write keys again diff --git a/src/server/proto.rs b/src/server/proto.rs index 35fe6ff..41bcbb8 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -62,7 +62,7 @@ use tokio::sync::{Mutex, mpsc}; use crate::server; use crate::error::MizeError; -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct Message { pub raw: Vec, version: u8, @@ -70,6 +70,7 @@ pub struct Message { origin: Origin, index: usize, meta_gotten: bool, + has_meta: bool, id: Option, } @@ -80,7 +81,7 @@ pub struct Message { // None, //} -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum Origin { Client(server::Client), //the client id Module(server::Module), //Module_name @@ -278,8 +279,11 @@ impl Message { 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} + 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; @@ -289,6 +293,11 @@ impl Message { 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(); @@ -297,7 +306,7 @@ impl Message { let modules = mutexes.modules.lock().await; for module in &*modules { if (module.name == mod_name) { - self.send(&Origin::Module(module.clone())); + self.send(&Origin::Module(module.clone())).await; return Ok(()); } } @@ -371,11 +380,11 @@ impl Message { for (mut key, mut val) in metadata.iter(){ meta_vec.append(&mut key.clone().into_bytes()); - meta_vec.push(';' as u8); + meta_vec.push('=' as u8); meta_vec.append(&mut val.clone().into_bytes()); - meta_vec.push('/' as u8); + meta_vec.push(';' as u8); } - meta_vec.remove(meta_vec.len()); + //meta_vec.remove(meta_vec.len()); meta_vec.push('}' as u8); self.raw.splice(2..2, meta_vec); @@ -383,7 +392,7 @@ impl Message { } pub fn get_metadata(&mut self) -> Result, MizeError> { - if !self.has_metadata() { + if !self.has_meta { return Err(MizeError { code: 108, kind: "faulty_message".to_string(), @@ -391,33 +400,37 @@ impl Message { }) }; + let meta_err = MizeError { + code: 107, + kind: "faulty_message".to_string(), + message: "Error while parsing the metadata of a message.".to_string(), + }; + 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(MizeError { - code: 107, - kind: "faulty_message".to_string(), - message: "Error while parsing the metadata of a message.".to_string(), - }) + return Err(meta_err); }; for ch in msg_iter { - meta_vec.push(*ch); if (ch == &('}' as u8)){ let mut map: HashMap = HashMap::new(); let string = String::from_utf8(meta_vec)?; - let sp: Vec<&str> = string.split("/").collect(); + let sp: Vec<&str> = string.split(";").collect(); + for st in sp.clone() { - let mut split = sp.split(|c| *c == ";"); - let key = split.nth(0).unwrap()[0]; - let val = split.nth(1).unwrap()[0]; + 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 { @@ -428,16 +441,12 @@ impl Message { } - pub fn has_metadata(&mut self) -> bool { - self.cmd >= 128 - } - 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_metadata() && !self.meta_gotten){ + if (self.has_meta && !self.meta_gotten){ self.get_metadata(); } let mut ch: u8 = 0; @@ -513,7 +522,7 @@ pub async fn handle_mize_message( ) -> Result<(), MizeError> { - //////////////////////////////////////////////////////////// + //###########################################################// //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(_)))) @@ -547,6 +556,7 @@ pub async fn handle_mize_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 @@ -604,14 +614,81 @@ pub async fn handle_mize_message( }; - //////////////////////////////////////////////////////////// + //###########################################################// //give from module if (message.cmd == MSG_GIVE && matches!(&message.origin, Origin::Module(_))){ - //TODO + let meta = message.get_metadata()?; + let id: String = message.get_id()?; + let client_id_str = meta.get("c").ok_or(MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "there is no \"c\" key in the metadata of a message gotten from a module".to_string(), + })?; + + let client_id: u64 = client_id_str.parse().map_err(|_| MizeError{ + code: 11, + kind: "don't know yet".to_string(), + message: "Error (std::num::ParseIntError) while parsing the \"c\" key in the Message metadata from a give Message from a Module".to_string() + })?; + + + 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{ + code: 11, + kind: "don't know yet".to_string(), + message: "".to_string() + }) + }; + + 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; + } + }; + + + //###########################################################// //update_request from module and client if (message.cmd == MSG_UPDATE_REQUEST && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ @@ -627,13 +704,12 @@ pub async fn handle_mize_message( } let update = Update::from_message(message.clone())?; - println!("UPDATE: {:?}", update.raw); 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()?; @@ -648,7 +724,7 @@ pub async fn handle_mize_message( } - //////////////////////////////////////////////////////////// + //###########################################################// //create from module and Client if (message.cmd == MSG_CREATE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){ @@ -712,8 +788,9 @@ pub async fn handle_mize_message( message.origin.send(Message::from_bytes(created_id_message, message.origin.clone())).await; } - //////////////////////////////////////////////////////////// + //###########################################################// //for every other message + println!("Got a Message That did not Trigger any if Statement...."); return Ok(()); } @@ -722,7 +799,6 @@ pub async fn handle_update(mut update: Update, mutexes: Mutexes, origin: Origin) //and either call itemstore.update(), send the update to the module or upstream, or spawn another //update by calling handle_update() - println!("handle_update"); ///////////////////////////////// TYPE CODE //////////////////////////////////////// //can change the update and spawn new updates (to a different item) (which would call the handle_update func again)