ppc/mize
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

some error handling

Browse Source
This commit is contained in:
Sebastian Moser
2022-11-04 11:16:03 +01:00
parent 626f9eda58
commit 134318c13b
9 changed files with 325 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
Cargo.lock generated
View File

@@ -1550,6 +1550,12 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "json"
version = "0.12.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "078e285eafdfb6c4b434e0d31e8cfcb5115b651496faca5749b88fafd4f23bfd"
[[package]]
name = "kv-log-macro"
version = "1.0.7"
@@ -1723,6 +1729,7 @@ version = "0.1.0"
dependencies = [
"futures-util",
"itertools",
"json",
"surrealdb",
"tokio",
"tokio-stream",

1
Cargo.toml
View File

@@ -8,6 +8,7 @@ edition = "2021"
[dependencies]
futures-util = "0.3.24"
itertools = "0.10.5"
json = "0.12.4"
surrealdb = "1.0.0-beta.7"
tokio = {version = "1.21.0", features = ["full"]}
tokio-stream = "0.1.9"

33
errorcodes.yml Normal file
View File

@@ -0,0 +1,33 @@
Can't delete item 0:
code: 100
category: "don't know yet"
message: "You can't delete item number 0. This item contains mandatory Config for the Server."
Error with Surrealdb:
code: 128
category: "data_storage"
num_of_items missing in item0:
code: 101
category: "key_missing::item0::num_of_items"
next_free_id missing in item0:
code: 102
category: "key_missing::item0::next_free_id"
_commit missing in item0:
code: 103
category: "key_missing::item0::_commit"
_commit missing:
code: 103
category: "key_missing::_commit"
could not read index of item:
code: 104
category: "data_storage::index_not_found"
_commit not 8 bytes long:
code: 105
category: "don't know yet"

2
js-client

Submodule js-client updated: c12d6a2a4e...a3e42a9e23

39
src/error.rs
View File

@@ -1,8 +1,41 @@
use json;
use crate::server::proto;
#[derive(Debug)]
pub struct MizeError {
pub message: String
pub kind: String,
pub code: u32,
pub message: String,
}
pub fn display_error_to_user(error: MizeError){
println!("{}", error.message);
impl MizeError {
pub fn to_json(self: MizeError) -> json::JsonValue{
let data = json::object!{
kind: self.kind,
code: self.code,
message: self.message,
};
return data;
}
pub fn to_message(self: MizeError) -> proto::Response {
let data = self.to_json();
let mut msg = vec![crate::PROTO_VERSION, proto::MSG_ERROR];
msg.extend(json::stringify(data).into_bytes());
return proto::Response::One(msg);
}
}
impl From<surrealdb::Error> for MizeError {
fn from(sur_err: surrealdb::Error) -> MizeError{
let cat = format!("data-storage:surrealdb:{:?}", sur_err);
let msg = format!("Something went wrong internally with the data-storage:surrealdb: {}", sur_err);
let err = MizeError {
code: 128,
kind: cat,
message: msg,
};
return err;
}
}

3
src/lib.rs
View File

@@ -1,5 +1,8 @@
#![ allow( warnings ) ]
static PROTO_VERSION: u8 = 1;
pub mod server;
pub mod error;

52
src/server.rs
View File

@@ -53,7 +53,7 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000;
//";
//
struct Client {
pub struct Client {
tx: UnboundedSender<Result<ws::Message, warp::Error>>
}
@@ -74,29 +74,6 @@ pub fn run_server(args: Vec<String>) {
}
}
//### prep the mize_folder
//if let Err(err) = crate::server::server_utils::init_mize_folder(mize_folder) {
//println!("{}", err.message);
//}
//println!("TEST");
//testing
//let val = "mize.works".to_string().into_bytes();
//let mut update: Vec<u8> = Vec::new();
//update.push(2);
//update.extend(u32::to_be_bytes(3));
//update.extend(u32::to_be_bytes(4));
//update.extend("$$".to_string().into_bytes());
//update.push(0);
//update.extend(u32::to_be_bytes(5));
//update.extend(u32::to_be_bytes(7));
//update.extend("$$".to_string().into_bytes());
//let new_val = proto::apply_update(&val, &update);
//println!("TEST");
//println!("TEST: {}", String::from_utf8(new_val).unwrap());
//run the webserver
warp_server(mize_folder);
}
@@ -115,7 +92,7 @@ async fn warp_server(mize_folder: String) {
//## $api/rest/
// get itemstore
let itemstore = crate::server::itemstore::Itemstore::new(mize_folder + "/db").await;
let itemstore = crate::server::itemstore::Itemstore::new(mize_folder + "/db").await.expect("error creating itemstore");
let mut clients: Arc<Mutex<Vec<Client>>> = Arc::new(Mutex::new(Vec::new()));
let mut itemstore_mutex: Arc<Mutex<Itemstore>> = Arc::new(Mutex::new(itemstore));
@@ -188,21 +165,32 @@ 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(proto::Message::new(msg.clone().into_bytes()), itemstore).await {
let responses = handle_mize_message(proto::Message::new(msg.clone().into_bytes()), itemstore).await;
for response in responses {
match response {
Response::One(response) => {
client_tx.send(Ok(ws::Message::binary(response.clone()))).unwrap()
},
Response::All(response) => {
let clients = clients_clone.lock().await;
println!("Clients: {}", clients.len());
for client in &clients[..] {
client.tx.send(Ok(ws::Message::binary(response.clone())));
}
drop(clients);
send_to_all_clients(ws::Message::binary(response), clients_clone.clone()).await;
},
Response::AllSubbed(id, response) => {
send_to_all_subed_clients(id, ws::Message::binary(response), clients_clone.clone()).await;
},
Response::None => {let t = 0;},
};
};
};
}
pub async fn send_to_all_clients(message: ws::Message, clients_clone: Arc<Mutex<Vec<Client>>>){
let clients = clients_clone.lock().await;
for client in &clients[..] {
client.tx.send(Ok(ws::Message::binary(message.clone())));
}
}
pub async fn send_to_all_subed_clients(id: u64, message: ws::Message, clients: Arc<Mutex<Vec<Client>>>){
}

204
src/server/itemstore.rs
View File

@@ -1,7 +1,10 @@
use std::net::ToSocketAddrs;
use std::{io, fmt::format, vec};
use surrealdb::{Datastore, Key, Val};
use itertools::Itertools;
use crate::error;
use crate::error::MizeError;
pub struct Itemstore {
db: surrealdb::Datastore,
@@ -10,33 +13,34 @@ pub struct Itemstore {
pub type Item = Vec<[Vec<u8>; 2]>;
impl Itemstore {
pub async fn new(path: String) -> Itemstore {
pub async fn new(path: String) -> Result<Itemstore, MizeError> {
let ds = Datastore::new(&(String::from("file://") + &path[..] + "/db")[..]).await.expect("Could not open database");
let mut tr = ds.transaction(true, false).await.expect("creation of transaction failed");
let mut tr = ds.transaction(true, false).await?;
match tr.get(vec!('0' as u8)).await {
Ok(Some(val)) => println!("item 0 already created"),
Ok(None) => {
let init = vec!["num_of_items".as_bytes().to_vec(), "next_free_id".as_bytes().to_vec()];
tr.set("0".to_string().into_bytes(), encode(init)).await;
tr.set("0".to_string().into_bytes(), encode(init)).await?;
let one: u64 = 1;
let zero: u64 = 0;
tr.set("0:num_of_items".to_string().into_bytes(), one.to_be_bytes()).await;
tr.set("0:next_free_id".to_string().into_bytes(), one.to_be_bytes()).await;
tr.set("0:_commit".to_string().into_bytes(), zero.to_be_bytes()).await;
tr.set("0:_type".to_string().into_bytes(), "mize-main".as_bytes().to_vec()).await;
tr.commit().await.expect("error commiting transaction in new function");
tr.set("0:num_of_items".to_string().into_bytes(), one.to_be_bytes()).await?;
tr.set("0:next_free_id".to_string().into_bytes(), one.to_be_bytes()).await?;
tr.set("0:_commit".to_string().into_bytes(), zero.to_be_bytes()).await?;
tr.set("0:_type".to_string().into_bytes(), "mize-main".as_bytes().to_vec()).await?;
tr.commit().await?;
},
Err(e) => println!("Error while getting item index with id 0: {}", e)
Err(e) => {return Err(MizeError::from(e))},
};
return Itemstore {db:ds};
return Ok(Itemstore {db:ds});
}
pub async fn create(&self, item: Item) -> u64 {
pub async fn create(&self, item: Item) -> Result<u64, MizeError> {
let mut new_id:u64 = 0;
let mut tr = self.db.transaction(true, false).await.expect("creation of transaction failed");
let mut tr = self.db.transaction(true, false).await?;
let id = tr.get("0:next_free_id").await
.expect("no item with id 0 when there definetly should be one")
.expect("error reading item 0 from db");
@@ -45,7 +49,7 @@ impl Itemstore {
//set id:_commit
let key = format!("{}:_commit", id_u64).into_bytes();
tr.set(key.clone(), vec![0,0,0,0,0,0,0,0]).await;
tr.set(key.clone(), vec![0,0,0,0,0,0,0,0]).await?;
keys.push("_commit".to_owned().into_bytes());
@@ -55,100 +59,130 @@ impl Itemstore {
keys.push(field[0].clone());
let val = field[1].clone();
tr.set(key, val).await;
tr.set(key, val).await?;
}
tr.set(format!("{}", id_u64), encode(keys)).await;
tr.set(format!("{}", id_u64), encode(keys)).await?;
//increment 0:number_of_items by 1
let num = tr.get("0:num_of_items".to_string().into_bytes()).await
.expect("error getting \"0:num_of_items\"")
.expect("error reading item 0 from db");
let num_res = tr.get("0:num_of_items".to_string().into_bytes()).await?;
if let Some(num) = num_res {
let mut num_u64 = u64::from_be_bytes([num[0], num[1], num[2], num[3], num[4], num[5], num[6], num[7]]) +1;
tr.set("0:num_of_items".to_string().into_bytes(), num_u64.to_be_bytes()).await;
//imcrement next_free_id by 1
let num = tr.get("0:next_free_id".to_string().into_bytes()).await
.expect("error getting \"0:next_free_id\"")
.expect("error reading item 0 from db");
let mut num_u64 = u64::from_be_bytes([num[0], num[1], num[2], num[3], num[4], num[5], num[6], num[7]]) +1;
tr.set("0:next_free_id".to_string().into_bytes(), num_u64.to_be_bytes()).await;
let res = tr.commit().await.expect("error commiting transaction in create function");
return u64::from_be_bytes([id[0], id[1], id[2], id[3], id[4], id[5], id[6], id[7], ])
new_id = num_u64;
tr.set("0:num_of_items".to_string().into_bytes(), num_u64.to_be_bytes()).await?;
} else {
return Err(MizeError{
code: 101,
kind: "key-missing:item0".to_string(),
message: "The Item0 should have a key num_of_items, but that is not there.".to_string(),
});
}
pub async fn update(&self, id: u64, new_item: Item){
let mut tr = self.db.transaction(true, true).await.expect("creation of transaction failed");
//imcrement next_free_id by 1
let num_res = tr.get("0:next_free_id".to_string().into_bytes()).await?;
if let Some(num) = num_res {
let mut num_u64 = u64::from_be_bytes([num[0], num[1], num[2], num[3], num[4], num[5], num[6], num[7]]) +1;
tr.set("0:next_free_id".to_string().into_bytes(), num_u64.to_be_bytes()).await?;
} else {
return Err(MizeError{
code: 101,
kind: "key-missing:item0".to_string(),
message: "The Item0 should have a key next_free_id, but that is not there.".to_string(),
});
}
let mut keys: Vec<Vec<u8>> = match tr.get(format!("{}", id).into_bytes()).await {
Ok(Some(val)) => decode(val),
Ok(None) => {
tr.cancel();
self.create(new_item);
return;
},
Err(e) => panic!("Error while querying the db: {}", e),
};
tr.commit().await?;
return Ok(new_id)
}
pub async fn update(&self, id: u64, new_item: Item) -> Result<(), MizeError>{
let mut tr = self.db.transaction(true, true).await?;
let mut keys: Vec<Vec<u8>> = Vec::new();
for field in new_item {
let mut key = format!("{}:", id).into_bytes();
key.extend(field[0].clone());
let val = field[1].clone();
keys.push(field[0].clone());
tr.set(key, val).await;
tr.set(key, val).await?;
}
let keys_res = tr.get(format!("{}", id).into_bytes()).await?;
if let Some(old_keys) = keys_res {
keys.extend(decode(old_keys));
keys = keys.into_iter().unique().collect::<Vec<Vec<u8>>>();
tr.set(format!("{}", id).into_bytes(), encode(keys)).await?;
} else {
return Err(MizeError{
code: 104,
kind: "data_storage::index_not_found".to_string(),
message: "Internal Datastorage Error: the Index of the Item was not found".to_string(),
});
}
//increment commit number
let mut commit_key = format!("{}:_commit", id).into_bytes();
let mut commit_num: u64 = match tr.get(&*commit_key).await {
Ok(Some(val)) => u64::from_be_bytes(val.try_into().expect("error converting _commit num to u64")),
Ok(None) => {
panic!("item {} has no _commit", id);
let commit_res = tr.get(&*commit_key).await?;
if let Some(commit_vec) = commit_res {
let val: [u8;8] = match commit_vec.try_into() {
Ok(val) => val,
Err(_) => {
return Err(MizeError{
code: 105,
kind: "don't know yet".to_string(),
message: "_commit of this item is no valid u64 (not 8 bytes long)".to_string(),
});
},
Err(e) => panic!("Error while querying the db: {}", e),
};
let mut commit_num = u64::from_be_bytes(val);
commit_num += 1;
tr.set(commit_key, commit_num.to_be_bytes()).await;
keys = keys.into_iter().unique().collect::<Vec<Vec<u8>>>();
tr.set(format!("{}", id).into_bytes(), encode(keys)).await;
let res = tr.commit().await.expect("error commiting transaction in update function");
tr.set(commit_key, commit_num.to_be_bytes()).await?;
} else {
return Err(MizeError{
code: 103,
kind: "key_missing::_commit".to_string(),
message: "This item has no _commit key, which every item must have.".to_string(),
});
}
pub async fn delete(&self, id: u64){
let mut tr = self.db.transaction(true, true).await.expect("creation of transaction failed");
let mut keys: Vec<Vec<u8>> = match tr.get(format!("{}", id).into_bytes()).await {
Ok(Some(val)) => decode(val),
Ok(None) => {
panic!("index of item {} not found", id);
},
Err(e) => panic!("Error while querying the db: {}", e),
};
for key in keys {
tr.del(key).await;
}
tr.del(format!("{}", id)).await;
let res = tr.commit().await.expect("error commiting transaction in delete function");
tr.commit().await?;
return Ok(());
}
pub async fn get(&self, id: u64) -> Item {
let mut tr = self.db.transaction(false, false).await.expect("creation of transaction failed");
pub async fn delete(&self, id: u64) -> Result<(), error::MizeError>{
let mut tr = self.db.transaction(true, true).await?;
let keys_res = tr.get(format!("{}", id).into_bytes()).await?;
if let Some(keys) = keys_res {
for key in decode(keys) {
tr.del(key).await?;
}
} else {
return Err(MizeError{
code: 104,
kind: "data_storage::index_not_found".to_string(),
message: "Internal Datastorage Error: the Index of the Item was not found".to_string(),
});
}
tr.del(format!("{}", id)).await?;
tr.commit().await?;
return Ok(());
}
pub async fn get(&self, id: u64) -> Result<Item, MizeError> {
let mut tr = self.db.transaction(false, false).await?;
let mut item: Item = Vec::new();
let mut keys: Vec<Vec<u8>> = match tr.get(format!("{}", id).into_bytes()).await {
Ok(Some(val)) => decode(val),
Ok(None) => {
panic!("index of item {} not found", id);
},
Err(e) => panic!("Error while querying the db: {}", e),
};
for key in keys {
let keys_res = tr.get(format!("{}", id).into_bytes()).await?;
if let Some(keys) = keys_res {
for key in decode(keys) {
let mut field: [Vec<u8>; 2] = [Vec::new(), Vec::new()];
field[0] = key.clone();
let mut ke = format!("{}:", id).into_bytes();
@@ -159,8 +193,16 @@ impl Itemstore {
Err(e) => panic!("Error while querying the db: {}", e),
};
item.push(field);
};
return Ok(item);
} else {
return Err(MizeError{
code: 104,
kind: "data_storage::index_not_found".to_string(),
message: "Internal Datastorage Error: the Index of the Item was not found".to_string(),
});
}
return item;
return Ok(item);
}
}

126
src/server/proto.rs
View File

@@ -1,13 +1,54 @@
// 1: get
// 2: give (by default only give the first 1k bytes, and the first 100 fields)
// 3: get_val (to get more than 1k bytes)
// 4: give_val
// 5: unauthorized
// 6: sub
// 7: unsub
// 8: update_request
// 9: update_deny
// 10: update
// 11: delete
// 12: create
// 13: created_id
// 14: unsupported_version
// 15: get_and_sub
// 16: get_fields
// 17: error
pub static MSG_GET: u8 = 1;
pub static MSG_GIVE: u8 = 2;
pub static MSG_GET_VAL: u8 = 3;
pub static MSG_GIVE_VAL: u8 = 4;
pub static MSG_UNAUTHORIZED: u8 = 5;
pub static MSG_SUB: u8 = 6;
pub static MSG_UNSUB: u8 = 7;
pub static MSG_UPDATE_REQUEST: u8 = 8;
pub static MSG_UPDATE_DENY: u8 = 9;
pub static MSG_UPDATE: u8 = 10;
pub static MSG_DELETE: u8 = 11;
pub static MSG_CREATE: u8 = 12;
pub static MSG_CREATED_ID: u8 = 13;
pub static MSG_UNSUPPORTED_VERSION: u8 = 14;
pub static MSG_GET_AND_SUB: u8 = 15;
pub static MSG_GET_FIELDS: u8 = 16;
pub static MSG_ERROR: u8 = 17;
//use std::panic::update_hook;
use std::vec;
use crate::server::itemstore;
use crate::server::itemstore::encode;
use crate::error;
pub enum Response {
One(Vec<u8>),
All(Vec<u8>),
AllSubbed(u64, Vec<u8>),
None,
}
@@ -75,26 +116,11 @@ const VERSION: u8 = 1;
pub async fn handle_mize_message(
mut message: Message,
itemstore: &itemstore::Itemstore,
) -> Response {
) -> Vec<Response> {
let old_message = message.msg.clone();
// 1: get
// 2: give
// 3: get_val
// 4: give_val
// 5: unauthorized
// 6: sub
// 7: unsub
// 8: update_request
// 9: update_deny
// 10: update
// 11: delete
// 12: create
// 13: created_id
// 14: unsupported_version
//println!("message: {:?}", message);
//let version = message.clone().into_iter().nth(0).expect("message has no 0th byte");
//let cmd = message.clone().into_iter().nth(1).expect("message has no 1th byte");
@@ -122,6 +148,7 @@ pub async fn handle_mize_message(
//println!("CMD: {}", cmd);
match cmd {
//get
1 => {
//let id_bytes = *&message[2..9].to_owned().clone();
let tmp = &old_message[2..10];
@@ -140,7 +167,10 @@ pub async fn handle_mize_message(
let mut answer: Vec<u8> = vec![VERSION, 2];
answer.extend(id.to_be_bytes());
let item = itemstore.get(id).await;
let mut item = match itemstore.get(id).await {
Ok(item) => item,
Err(err) => {return vec![err.to_message()]},
};
let num_of_fields = item.len() as u32;
answer.extend(num_of_fields.to_be_bytes());
@@ -149,19 +179,19 @@ pub async fn handle_mize_message(
let key_len = field[0].len() as u32;
answer.extend(key_len.to_be_bytes());
answer.extend(field[0].clone());
let val_len = field[1].len() as u32;
let val_len = field[1].len() as u32;
answer.extend(val_len.to_be_bytes());
answer.extend(field[1].clone());
}
return Response::One(answer);
return vec![Response::One(answer)];
},
2 => {return Response::None;},
3 => {return Response::None;},
4 => {return Response::None;},
5 => {return Response::None;},
6 => {return Response::None;},
7 => {return Response::None;},
2 => {return Vec::new()},
3 => {return Vec::new()},
4 => {return Vec::new()},
5 => {return Vec::new()},
6 => {return Vec::new()},
7 => {return Vec::new()},
//update_request
8 => {
@@ -170,7 +200,10 @@ pub async fn handle_mize_message(
let id = message.get_id();
let mut item = itemstore.get(id).await;
let mut item = match itemstore.get(id).await {
Ok(item) => item,
Err(err) => {return vec![err.to_message()]},
};
let num_of_updates = message.get_u32();
@@ -184,7 +217,6 @@ pub async fn handle_mize_message(
let mut found = false;
for a in 0..item.len() as usize {
println!("OTHER KEY: {}", String::from_utf8(item[a][0].clone()).expect("here utf-8"));
if item[a][0] == key {
item[a][1] = apply_update(&item[a][1], &update);
found = true;
@@ -192,31 +224,39 @@ pub async fn handle_mize_message(
}
}
if found == false {
println!("should not be here");
let index = item.len() +1;
//item[index][1] = apply_update(&item[index][1], &update)
item.push([key.clone(), apply_update(&item[index][1], &update)])
}
}
//println!("ITEM KEY: {}", String::from_utf8(item[1][0].clone()).unwrap());
//println!("ITEM VAL: {}", String::from_utf8(item[1][1].clone()).unwrap());
itemstore.update(id, item).await;
return Response::All(answer);
if let Err(err) = itemstore.update(id, item).await {
return vec![err.to_message()];
};
return vec![Response::All(answer)];
},
9 => {return Response::None;},
10 => {return Response::None;},
9 => {return Vec::new()},
10 => {return Vec::new()},
//delete
11 => {
let response: Vec<Response> = Vec::new();
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;
let err = error::MizeError{
kind: "don't know yet".to_string(),
code: 100,
message: "You can't delete item number 0. This item contains mandatory Config for the Server.".to_string(),
};
return vec![err.to_message()];
}
itemstore.delete(id);
return Response::None;
if let Err(err) = itemstore.delete(id).await {
return vec![err.to_message()];
}
return response;
},
//create
@@ -236,14 +276,16 @@ pub async fn handle_mize_message(
item.push([key, val]);
}
itemstore.create(item).await;
if let Err(err) = itemstore.create(item).await {
return vec![err.to_message()];
};
let anser: Vec<u8> = vec![1,1];
return Response::None;
return vec![Response::None];
},
13 => {return Response::None;},
14 => {return Response::None;},
_ => {return Response::None;},
13 => {return Vec::new()},
14 => {return Vec::new()},
_ => {return Vec::new()},
}