last commit of old idea

src/server.rs

@@ -23,6 +23,7 @@ use serde_json::Value as JsonValue;
 use serde::{Serialize, Deserialize};
 use uuid::Uuid;
 use std::str::FromStr;
+use std::borrow::Cow;
 
 use tokio::sync::mpsc::{Sender, channel, Receiver};
 use tokio_stream::wrappers::ReceiverStream;
@@ -39,7 +40,7 @@ use axum::Router;
 use axum::routing::{get, get_service};
 use std::net::SocketAddr;
 use axum::response::{IntoResponse, Html};
-use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
+use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade, CloseFrame};
 use axum::extract::{State, self};
 use axum::http::{StatusCode, Uri, Response, self};
 use axum::http::header::{HeaderName, HeaderValue, HeaderMap};
@@ -353,34 +354,6 @@ async fn handle_websocket_connection(
 
     let mut msg_rx = ReceiverStream::new(msg_rx);
 
-    // my own forward
-    tokio::spawn(async move {
-        while let Some(msg) = msg_rx.next().await {
-            match msg {
-                proto::MizeMessage::Json(json_msg) => {
-                    if let Ok(msg) = serde_json::to_string(&json_msg) {
-                        println!("SEINDING: {}", msg);
-                        socket_tx.send(Message::Text(msg)).await;
-                    } else {
-                        let err: MizeError = MizeError::new(11).extra_msg("error while serializing a json message").handle();
-                        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;
-                }
-                proto::MizeMessage::Pong() => {
-                    socket_tx.send(Message::Pong(Vec::new())).await;
-                }
-            }
-        }
-    });
-
-
     // check if mize-module header exists
     let origin = if let Some(mod_header_val) = headers.get("mize-module"){
 
@@ -420,12 +393,50 @@ async fn handle_websocket_connection(
         Peer::Client(client)
     };
 
+    // Forwarding messages
+    let myorigin = origin.clone();
+    let mymutexes = mutexes.clone();
+    tokio::spawn(async move {
+        while let Some(msg) = msg_rx.next().await {
+            match msg {
+                proto::MizeMessage::Json(json_msg) => {
+                    if let Ok(msg) = serde_json::to_string(&json_msg) {
+                        let result = socket_tx.send(Message::Text(msg.clone())).await;
+                        if let Err(e) = result {
+                            println!("Client no longer Connected: {:?}", e);
+                            remove_peer(myorigin.clone(), mymutexes.clone()).await;
+                        } else {
+                            println!("SENDING: {}", msg);
+                        }
+                    } 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;
+                }
+                proto::MizeMessage::Pong() => {
+                    socket_tx.send(Message::Pong(Vec::new())).await;
+                }
+                proto::MizeMessage::Close((num, reason)) => {
+                    println!("SENDING Close message");
+                    socket_tx.send(Message::Close(Some(CloseFrame{code: num, reason: Cow::Owned(reason)}))).await;
+                }
+            }
+        }
+    });
+
     // Reading messages
     while let Some(result) = socket_rx.next().await {
         let msg = if let Ok(msg) = result {msg} else {
-            let err = MizeError::new(115).handle();
+            let err = MizeError::new(115).extra_msg("TCP Connection Closed");
-            msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
+            remove_peer(origin.clone(), mutexes.clone());
-            continue;
+            return;
         };
         match msg {
             Message::Binary(b) => {
@@ -442,39 +453,18 @@ async fn handle_websocket_connection(
                 };
             },
 
-            Message::Close(_) => {
+            Message::Close(inner) => {
-                match origin {
+                match inner {
-                    Peer::Client(ref client) => {
+                    Some(CloseFrame{code, reason}) => {
-                        //remove client from client list
+                        origin.send(MizeMessage::Close((code, reason.into_owned()))).await;
-                        let mut clients = mutexes.clients.lock().await;
-                        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;
                     },
-                    Peer::Module(module) => {
+                    None => {
-                        //remove module from module list
+                        origin.send(MizeMessage::Close((1000, "You didn't send a Close Reason".to_owned()))).await;
-                        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;
                     }
-                    Peer::Upstream(_) => {
+                };
-                        return;
+
-                    }
+                println!("Close from Client");
-                }
+                remove_peer(origin.clone(), mutexes.clone()).await;
             }
 
 
@@ -487,7 +477,6 @@ async fn handle_websocket_connection(
                 let err = MizeError::new(11)
                     .extra_msg("unhandeld WebSocket-Message type: Pong")
                     .handle();
-
                 msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
             },
 
@@ -631,5 +620,37 @@ async fn init_server(mize_folder: String) -> Result<(Uuid, Itemstore, Vec<Render
     return Ok((server_uuid, itemstore, renders));
 }
 
-
+pub async fn remove_peer(origin: Peer, mutexes: Mutexes) {
+    match origin {
+        Peer::Client(ref client) => {
+            //remove client from client list
+            let mut clients = mutexes.clients.lock().await;
+            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);
+            return;
+        },
+        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.....");
+            modules.remove(index);
+            println!("Close from Module");
+            return;
+        }
+        Peer::Upstream(_) => {
+            return;
+        }
+    }
+}
 

src/server/itemstore.rs

@@ -12,7 +12,7 @@ use crate::server::Peer;
 use serde_json::Value as JsonValue;
 use serde::{Serialize, Deserialize};
 
-use super::proto::Delta;
+use super::proto::{Delta, MizeId};
 
 // The struct to do the storage and updates to it
 // is responsible, that no illegal states can occur in the storage, by using transactions
@@ -27,6 +27,16 @@ pub struct Item {
     pub json: JsonValue,
 }
 
+pub struct Array {
+}
+
+//later
+//pub struct Map {
+//}
+
+//pub struct Table {
+//}
+
 impl Item {
     pub fn get_id(&self) -> Result<String, MizeError> {
         if let JsonValue::Object(ob) = &self.json{
@@ -97,7 +107,7 @@ impl Item {
             *to_replace = value.into();
         }
 
-        println!("After applying a Delta, {}", self.json);
+        //println!("After applying a Delta, {}", self.json);
         return Ok(())
     }
 
@@ -116,8 +126,8 @@ impl Itemstore {
 
         let mut tr = ds.transaction(true, false).await?;
 
-        match tr.get(0u64.to_le_bytes()).await {
+        match tr.get(MizeId::zero().into_bytes()).await {
-            Ok(Some(val)) => println!("item 0 already created"),
+            Ok(Some(val)) => println!("Item 0 already created"),
             Ok(None) => {
                 let data = serde_json::json!({
                         "num_of_items": 1,
@@ -127,7 +137,7 @@ impl Itemstore {
                         "__item__": "0"
                 });
 
-                tr.set(0u64.to_be_bytes(), data.to_string()).await?;
+                tr.set(MizeId::zero().into_bytes(), data.to_string()).await?;
                 tr.commit().await?;
             },
 
@@ -136,28 +146,29 @@ impl Itemstore {
         return Ok(Itemstore {db:ds});
     }
 
-    pub async fn create(&self, mut item: Item, mutexes: Mutexes, origin: Peer) -> Result<u64, MizeError> {
+    pub async fn create(&self, mut item: Item, mutexes: Mutexes, origin: Peer) -> Result<MizeId, MizeError> {
         let mut tr = self.db.transaction(true, false).await?;
 
         //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))?)
+        let json: JsonValue = serde_json::from_str(&String::from_utf8(tr.get(MizeId::zero().into_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))?;
+        let new_id_u64: u64 = json.get("next_free_id").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?;
+        let new_id = MizeId::new(format!("{}", new_id_u64));
 
         let num_of_items: u64 = json.get("num_of_items").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?;
 
-        let new_next_id = new_id +1;
+        let new_next_id = new_id_u64 +1;
         let new_num = num_of_items + 1;
 
         //add mize-id field to item
         //let delta: Delta = serde_json::from_str(&format!("[[[\"__item__\"], \"{}\"]]", new_id))?;
         let mut delta = Delta::new();
-        delta.append(vec!["__item__"], serde_json::json!(new_id));
+        delta.append(vec!["__item__"], serde_json::json!(new_id_u64));
         item.apply_delta(delta)?;
 
-        tr.set(new_id.to_be_bytes(), item.json.to_string()).await?;
+        tr.set(new_id.clone().into_bytes(), item.json.to_string()).await?;
 
         tr.commit().await?;
 
@@ -172,12 +183,10 @@ impl Itemstore {
     }
 
 
-    pub async fn update(&self, id: u64, delta: Delta) -> Result<(), MizeError>{
+    pub async fn update(&self, id: MizeId, delta: Delta) -> Result<(), MizeError>{
-        println!("in itemstore.update");
-
         let mut tr = self.db.transaction(true, true).await?;
 
-        let item_bytes: Vec<u8> = tr.get(id.to_le_bytes()).await?
+        let item_bytes: Vec<u8> = tr.get(id.clone().into_bytes()).await?
             .ok_or(MizeError::new(11)
                 .extra_msg("The Item an update should be applied to does not exist."))?;
 
@@ -186,7 +195,7 @@ impl Itemstore {
         item.apply_delta(delta)?;
         //println!("after applying delta: {:?}", item);
 
-        tr.set(id.to_be_bytes(), serde_json::to_string(&item)?).await?;
+        tr.set(id.into_bytes(), serde_json::to_string(&item)?).await?;
 
         tr.commit().await?;
         return Ok(());
@@ -194,10 +203,10 @@ impl Itemstore {
 
 
 
-    pub async fn delete(&self, id: u64) -> Result<(), error::MizeError>{
+    pub async fn delete(&self, id: MizeId) -> Result<(), error::MizeError>{
         let mut tr = self.db.transaction(true, true).await?;
 
-        tr.del(id.to_be_bytes()).await?;
+        tr.del(id.into_bytes()).await?;
 
         tr.commit().await?;
 
@@ -205,16 +214,16 @@ impl Itemstore {
     }
 
 
-    pub async fn get(&self, id: u64) -> Result<Item, MizeError> {
+    pub async fn get(&self, id: MizeId) -> Result<Item, MizeError> {
         let mut tr = self.db.transaction(false, false).await?;
 
-        let item_res = tr.get(id.to_le_bytes()).await?;
+        let item_res = tr.get(id.clone().into_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(32).format(vec![&format!("{}", id)]));
+            return Err(MizeError::new(32).format(vec![&format!("{:?}", id)]));
         }
         return Err(MizeError::new(11));
     }

src/server/proto.rs

@@ -22,6 +22,7 @@ use crate::server;
 use crate::error::MizeError;
 use crate::server::itemstore::Item;
 use crate::server::proto;
+use axum::extract::ws::CloseFrame;
 
 use serde_json::Value as JsonValue;
 use serde::{Serialize, Deserialize};
@@ -33,6 +34,7 @@ pub enum MizeMessage {
     Json(JsonMessage),
     Bin(BinMessage),
     Pong(),
+    Close((u16, String)),
 }
 
 
@@ -221,7 +223,19 @@ impl MizeId {
         MizeId { main }
     }
 
-    pub fn as_string(self) -> String {
+    pub fn from_local(local: u64) -> MizeId {
+        MizeId::new(format!("{}", local))
+    }
+
+    pub fn into_bytes(self) -> Vec<u8> {
+        self.main.into_bytes()
+    }
+
+    pub fn zero() -> MizeId {
+        MizeId::new("0".to_owned())
+    }
+
+    pub fn to_string(self) -> String {
         self.main
     }
 
@@ -323,7 +337,7 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -
                     let itemstore = mutexes.itemstore.lock().await;
                     let response = GiveItemMessage {
                         id: msg.id,
-                        item: itemstore.get(id).await?,
+                        item: itemstore.get(MizeId::from_local(id)).await?,
                     };
                     origin.send(response).await;
                 },
@@ -354,7 +368,7 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -
                     let itemstore = mutexes.itemstore.lock().await;
                     let response = GiveItemMessage {
                         id: msg.id,
-                        item: itemstore.get(id).await?,
+                        item: itemstore.get(MizeId::from_local(id)).await?,
                     };
                     origin.send(response).await;
                 },
@@ -391,7 +405,7 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -
         JsonMessage::Create(msg) => {
             let itemstore = mutexes.itemstore.lock().await;
             let id = itemstore.create(msg.item, mutexes.clone(), origin.clone()).await?;
-            let response = CreatedIdMessage {id: MizeId::new(format!("{}", id))};
+            let response = CreatedIdMessage {id};
 
             origin.send(response).await;
             return Ok(());
@@ -402,7 +416,7 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -
             match msg.id.kind()? {
                 MizeIdKind::Local(id) => {
                     let itemstore = mutexes.itemstore.lock().await;
-                    itemstore.delete(id).await?;
+                    itemstore.delete(MizeId::from_local(id)).await?;
                     return Ok(());
                 },
                 _ => {return Err(MizeError::new(11).extra_msg("deleteng non Local items not implemented yet"));}
@@ -424,7 +438,7 @@ pub async fn handle_update(update: Update, origin: Peer, mutexes: Mutexes) -> Re
 
 pub async fn update_thread(mut update_rx: mpsc::Receiver<(Update, Peer)>, mutexes: Mutexes) -> Result<(), MizeError> {
     while let Some((update, origin)) = update_rx.recv().await {
-        println!("UPDATE: {:?}", update.iter().next().unwrap().0);
+        println!("UPDATE on id: {:?}", update.iter().next().unwrap().0.main);
         //TODO: run update code from modules and types
 
         let itemstore = mutexes.itemstore.lock().await;
@@ -434,7 +448,7 @@ pub async fn update_thread(mut update_rx: mpsc::Receiver<(Update, Peer)>, mutexe
 
             //update the items in the itemstore
             if let MizeIdKind::Local(id) = id.kind()? {
-                itemstore.update(id.clone(), delta.clone()).await?;
+                itemstore.update(MizeId::from_local(id), delta.clone()).await?;
             } else {
                 return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet"));
             }