diff --git a/src/server.rs b/src/server.rs index 0554a91..8dce319 100644 --- a/src/server.rs +++ b/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(); - msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await; - continue; + let err = MizeError::new(115).extra_msg("TCP Connection Closed"); + remove_peer(origin.clone(), mutexes.clone()); + return; }; match msg { Message::Binary(b) => { @@ -442,39 +453,18 @@ async fn handle_websocket_connection( }; }, - Message::Close(_) => { - 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); - println!("Close from Client"); - return; + Message::Close(inner) => { + match inner { + Some(CloseFrame{code, reason}) => { + origin.send(MizeMessage::Close((code, reason.into_owned()))).await; }, - 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; + None => { + origin.send(MizeMessage::Close((1000, "You didn't send a Close Reason".to_owned()))).await; } - 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 { + //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; + } + } +} diff --git a/src/server/itemstore.rs b/src/server/itemstore.rs index 8e362ca..13b8ec2 100644 --- a/src/server/itemstore.rs +++ b/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 { 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 { - Ok(Some(val)) => println!("item 0 already created"), + match tr.get(MizeId::zero().into_bytes()).await { + 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 { + pub async fn create(&self, mut item: Item, mutexes: Mutexes, origin: Peer) -> Result { let mut tr = self.db.transaction(true, false).await?; //get next_free_id - 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>{ - println!("in itemstore.update"); - + pub async fn update(&self, id: MizeId, delta: Delta) -> Result<(), MizeError>{ let mut tr = self.db.transaction(true, true).await?; - let item_bytes: Vec = tr.get(id.to_le_bytes()).await? + let item_bytes: Vec = 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 { + pub async fn get(&self, id: MizeId) -> Result { 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)); } diff --git a/src/server/proto.rs b/src/server/proto.rs index 21c53e7..67aade7 100644 --- a/src/server/proto.rs +++ b/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 { + 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")); }