fields are now deleted when the val is empty .... among other changes

2023-01-14 23:14:37 +01:00
parent 5d4ea7256e
commit 188a7b2a7e
8 changed files with 237 additions and 59 deletions
--- 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",
 ]

--- 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"
--- 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"
--- a/2
+++ b/2
--- 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,
--- 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<Result<ws::Message, warp::Error>>,
    tx: Sender<proto::Message>,
    id: u64,
 }

-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct Module {
    tx: Sender<proto::Message>,
    name: String,
@@ -84,11 +85,14 @@ pub struct Module {
    registered_types: Vec<String>,
 }

-#[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<String>, State(mutexes

 async fn websocket_handler(
        ws: WebSocketUpgrade,
-            State(mutexes): State<Mutexes>,
+        State(mutexes): State<Mutexes>,
+        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<proto::Message>, Receiver<proto::Message>) = 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;
        };
    };
 }
--- 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
--- 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<u8>,
    version: u8,
@@ -70,6 +70,7 @@ pub struct Message {
    origin: Origin,
    index: usize,
    meta_gotten: bool,
+    has_meta: bool,
    id: Option<String>,
 }

@@ -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
@@ -279,7 +280,10 @@ impl Message {
            if count == 2 {break};
        };
        
-        Message { raw: vec, version, cmd, index: 2 , origin, meta_gotten: false, id: None}
+        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;
@@ -289,6 +293,11 @@ impl Message {
        let mut meta: HashMap<String, String> = 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<HashMap<String, String>, 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<u8> = 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<String, String> = 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::<Vec<&str>>();
+                    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<String, MizeError> {
        //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)