handling of update_request message implemented, but not tested
This commit is contained in:
@@ -6,7 +6,7 @@ pub mod proto;
|
|||||||
|
|
||||||
use futures_util::{FutureExt, StreamExt};
|
use futures_util::{FutureExt, StreamExt};
|
||||||
use warp::Filter;
|
use warp::Filter;
|
||||||
use warp::ws::{WebSocket, Message};
|
use warp::ws::{WebSocket, self};
|
||||||
|
|
||||||
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
|
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
|
||||||
use tokio_stream::wrappers::UnboundedReceiverStream;
|
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||||
@@ -15,7 +15,7 @@ use std::fs;
|
|||||||
use std::sync::{Arc};
|
use std::sync::{Arc};
|
||||||
use tokio::sync::Mutex;
|
use tokio::sync::Mutex;
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
use crate::server::proto::Response;
|
use crate::server::proto::{Response};
|
||||||
use crate::server::proto::handle_mize_message;
|
use crate::server::proto::handle_mize_message;
|
||||||
|
|
||||||
use self::itemstore::Itemstore;
|
use self::itemstore::Itemstore;
|
||||||
@@ -54,7 +54,7 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000;
|
|||||||
//
|
//
|
||||||
|
|
||||||
struct Client {
|
struct Client {
|
||||||
tx: UnboundedSender<Result<Message, warp::Error>>
|
tx: UnboundedSender<Result<ws::Message, warp::Error>>
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn run_server(args: Vec<String>) {
|
pub fn run_server(args: Vec<String>) {
|
||||||
@@ -177,14 +177,14 @@ async fn handle_socket_connection(
|
|||||||
let msg = result.expect("Error when gettin message from WebSocket");
|
let msg = result.expect("Error when gettin message from WebSocket");
|
||||||
|
|
||||||
let itemstore = &*itemstore_clone.lock().await;
|
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) => {
|
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) => {
|
Response::All(response) => {
|
||||||
let clients = clients_clone.lock().await;
|
let clients = clients_clone.lock().await;
|
||||||
for client in &clients[..] {
|
for client in &clients[..] {
|
||||||
client.tx.send(Ok(Message::binary(response.clone())));
|
client.tx.send(Ok(ws::Message::binary(response.clone())));
|
||||||
}
|
}
|
||||||
drop(clients);
|
drop(clients);
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -12,13 +12,74 @@ pub enum Response {
|
|||||||
None,
|
None,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub struct Message {
|
||||||
|
msg: Vec<u8>,
|
||||||
|
version: u8,
|
||||||
|
cmd: u8,
|
||||||
|
index: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Message {
|
||||||
|
pub fn new(vec: Vec<u8>) -> 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<u8>{
|
||||||
|
let tmp = &self.msg[self.index..self.index + n];
|
||||||
|
self.index += n;
|
||||||
|
|
||||||
|
return tmp.to_vec();
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
const VERSION: u8 = 1;
|
const VERSION: u8 = 1;
|
||||||
|
|
||||||
pub async fn handle_mize_message(
|
pub async fn handle_mize_message(
|
||||||
message: Vec<u8>,
|
mut message: Message,
|
||||||
itemstore: &itemstore::Itemstore,
|
itemstore: &itemstore::Itemstore,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
|
|
||||||
|
let old_message = message.msg.clone();
|
||||||
|
|
||||||
|
|
||||||
// 1: get
|
// 1: get
|
||||||
// 2: give
|
// 2: give
|
||||||
@@ -47,7 +108,7 @@ pub async fn handle_mize_message(
|
|||||||
let mut cmd: u8 = 0;
|
let mut cmd: u8 = 0;
|
||||||
let mut version: u8 = 0;
|
let mut version: u8 = 0;
|
||||||
|
|
||||||
for i in message.clone() {
|
for i in old_message.clone() {
|
||||||
if count == 0 {
|
if count == 0 {
|
||||||
version = i;
|
version = i;
|
||||||
}
|
}
|
||||||
@@ -64,7 +125,7 @@ pub async fn handle_mize_message(
|
|||||||
match cmd {
|
match cmd {
|
||||||
1 => {
|
1 => {
|
||||||
//let id_bytes = *&message[2..9].to_owned().clone();
|
//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]]);
|
let id: u64 = u64::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3], tmp[4], tmp[5], tmp[6], tmp[7]]);
|
||||||
|
|
||||||
// answer:
|
// answer:
|
||||||
@@ -102,11 +163,40 @@ pub async fn handle_mize_message(
|
|||||||
5 => {return Response::None;},
|
5 => {return Response::None;},
|
||||||
6 => {return Response::None;},
|
6 => {return Response::None;},
|
||||||
7 => {return Response::None;},
|
7 => {return Response::None;},
|
||||||
8 => {return Response::None;},
|
8 => {
|
||||||
|
let mut answer: Vec<u8> = 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;},
|
9 => {return Response::None;},
|
||||||
10 => {return Response::None;},
|
10 => {return Response::None;},
|
||||||
|
|
||||||
|
//delete
|
||||||
11 => {
|
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]]);
|
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 {
|
if id == 0 {
|
||||||
return Response::None;
|
return Response::None;
|
||||||
@@ -115,22 +205,24 @@ pub async fn handle_mize_message(
|
|||||||
itemstore.delete(id);
|
itemstore.delete(id);
|
||||||
return Response::None;
|
return Response::None;
|
||||||
},
|
},
|
||||||
|
|
||||||
|
//create
|
||||||
12 => {
|
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 num_of_fields: u32 = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]);
|
||||||
let mut item: Vec<[Vec<u8>; 2]> = Vec::new();
|
let mut item: Vec<[Vec<u8>; 2]> = Vec::new();
|
||||||
|
|
||||||
let mut index = 6;
|
let mut index = 6;
|
||||||
for i in 0..num_of_fields {
|
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_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;
|
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_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;
|
index += 4 + val_len as usize;
|
||||||
|
|
||||||
@@ -150,12 +242,7 @@ pub async fn handle_mize_message(
|
|||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn apply_update(val: Vec<u8>, update: Vec<u8>) -> Vec<u8>{
|
||||||
|
Vec::new()
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user