From 4c566c6622e84e046af7026acdee1a199c8643b9 Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Fri, 14 Oct 2022 12:47:37 +0200 Subject: [PATCH] handling of update_request message implemented, but not tested --- src/server.rs | 12 ++--- src/server/proto.rs | 123 +++++++++++++++++++++++++++++++++++++------- 2 files changed, 111 insertions(+), 24 deletions(-) diff --git a/src/server.rs b/src/server.rs index e7be307..37cf357 100644 --- a/src/server.rs +++ b/src/server.rs @@ -6,7 +6,7 @@ pub mod proto; use futures_util::{FutureExt, StreamExt}; use warp::Filter; -use warp::ws::{WebSocket, Message}; +use warp::ws::{WebSocket, self}; use tokio::sync::mpsc::{UnboundedSender, unbounded_channel}; use tokio_stream::wrappers::UnboundedReceiverStream; @@ -15,7 +15,7 @@ use std::fs; use std::sync::{Arc}; use tokio::sync::Mutex; use tokio::sync::mpsc; -use crate::server::proto::Response; +use crate::server::proto::{Response}; use crate::server::proto::handle_mize_message; use self::itemstore::Itemstore; @@ -54,7 +54,7 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000; // struct Client { - tx: UnboundedSender> + tx: UnboundedSender> } pub fn run_server(args: Vec) { @@ -177,14 +177,14 @@ async fn handle_socket_connection( let msg = result.expect("Error when gettin message from WebSocket"); let itemstore = &*itemstore_clone.lock().await; - match handle_mize_message(msg.clone().into_bytes(), itemstore).await { + match handle_mize_message(proto::Message::new(msg.clone().into_bytes()), itemstore).await { Response::One(response) => { - client_tx.send(Ok(Message::binary(response.clone()))).unwrap() + client_tx.send(Ok(ws::Message::binary(response.clone()))).unwrap() }, Response::All(response) => { let clients = clients_clone.lock().await; for client in &clients[..] { - client.tx.send(Ok(Message::binary(response.clone()))); + client.tx.send(Ok(ws::Message::binary(response.clone()))); } drop(clients); }, diff --git a/src/server/proto.rs b/src/server/proto.rs index 07cf6ea..66400d2 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -12,13 +12,74 @@ pub enum Response { None, } +pub struct Message { + msg: Vec, + version: u8, + cmd: u8, + index: usize, +} + +impl Message { + pub fn new(vec: Vec) -> 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; + }; + + Message { msg: vec, version: version, cmd: cmd, index: 2 } + } + + pub fn get_id(&mut self) -> u64 { + self.get_u64() + } + + pub fn get_u32(&mut self) -> u32 { + let tmp = &self.msg[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.msg[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.msg[self.index..self.index + n]; + self.index += n; + + return tmp.to_vec(); + } + +} + const VERSION: u8 = 1; pub async fn handle_mize_message( - message: Vec, + mut message: Message, itemstore: &itemstore::Itemstore, ) -> Response { + let old_message = message.msg.clone(); + // 1: get // 2: give @@ -47,7 +108,7 @@ pub async fn handle_mize_message( let mut cmd: u8 = 0; let mut version: u8 = 0; - for i in message.clone() { + for i in old_message.clone() { if count == 0 { version = i; } @@ -64,7 +125,7 @@ pub async fn handle_mize_message( match cmd { 1 => { //let id_bytes = *&message[2..9].to_owned().clone(); - let tmp = &message[2..10]; + let tmp = &old_message[2..10]; let id: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]); // answer: @@ -102,11 +163,40 @@ pub async fn handle_mize_message( 5 => {return Response::None;}, 6 => {return Response::None;}, 7 => {return Response::None;}, - 8 => {return Response::None;}, + 8 => { + let mut answer: Vec = message.msg.clone(); + answer[1] = 10; + + let id = message.get_id(); + + let mut item = itemstore.get(id).await; + + let num_of_updates = message.get_u32(); + + for i in 0..num_of_updates { + let key_len = message.get_u32(); + let key = message.get_bytes(key_len as usize); + + let update_len = message.get_u32(); + let update = message.get_bytes(update_len as usize); + + //change the item + for mut field in item.clone() { + if field[0] == key { + field[1] = apply_update(field[1].clone(), update.clone()); + } + } + } + itemstore.update(id, item).await; + + return Response::All(answer); + }, 9 => {return Response::None;}, 10 => {return Response::None;}, + + //delete 11 => { - let tmp = &message[2..10]; + let tmp = &old_message[2..10]; let id: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]); if id == 0 { return Response::None; @@ -115,22 +205,24 @@ pub async fn handle_mize_message( itemstore.delete(id); return Response::None; }, + + //create 12 => { - let tmp = &message[2..6]; + let tmp = &old_message[2..6]; let num_of_fields: u32 = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]); let mut item: Vec<[Vec; 2]> = Vec::new(); let mut index = 6; for i in 0..num_of_fields { - let mut tmp = &message[index..index + 4]; + let mut tmp = &old_message[index..index + 4]; let key_len = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]); - let key = message[index + 4..(index + 4 + key_len as usize)].to_vec(); + let key = old_message[index + 4..(index + 4 + key_len as usize)].to_vec(); index += 4 + key_len as usize; - tmp = &message[index..index + 4]; + tmp = &old_message[index..index + 4]; let val_len = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]); - let val = message[index + 4..(index + 4 + val_len as usize)].to_vec(); + let val = old_message[index + 4..(index + 4 + val_len as usize)].to_vec(); index += 4 + val_len as usize; @@ -150,12 +242,7 @@ pub async fn handle_mize_message( } - - - - - - - - +pub fn apply_update(val: Vec, update: Vec) -> Vec{ + Vec::new() +}