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

basic JSON message handling. not tested yet at all

Browse Source
This commit is contained in:
Sebastian Moser
2023-02-12 23:25:50 +01:00
parent d3ecfcabf4
commit 9825c5242b
7 changed files with 790 additions and 1081 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
Cargo.lock generated
View File

@@ -664,6 +664,17 @@ dependencies = [
"unicode-width",
]
[[package]]
name = "colored"
version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b3616f750b84d8f0de8a58bda93e08e2a81ad3f523089b05f1dffecab48c6cbd"
dependencies = [
"atty",
"lazy_static",
"winapi",
]
[[package]]
name = "concurrent-queue"
version = "1.2.4"
@@ -679,6 +690,12 @@ version = "0.4.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fbdcdcb6d86f71c5e97409ad45898af11cbc995b4ee8112d59095a28d376c935"
[[package]]
name = "convert_case"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
[[package]]
name = "cookie"
version = "0.14.4"
@@ -899,6 +916,19 @@ dependencies = [
"syn",
]
[[package]]
name = "derive_more"
version = "0.99.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4fb810d30a7c1953f91334de7244731fc3f3c10d7fe163338a35b9f640960321"
dependencies = [
"convert_case",
"proc-macro2",
"quote",
"rustc_version 0.4.0",
"syn",
]
[[package]]
name = "deunicode"
version = "1.3.2"
@@ -1617,12 +1647,6 @@ 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"
@@ -1803,10 +1827,13 @@ dependencies = [
"axum",
"axum-extra",
"axum-macros",
"colored",
"derive_more",
"futures-util",
"itertools",
"json",
"lazy_static",
"serde",
"serde_json",
"surrealdb",
"tokio",
"tokio-stream",
@@ -2476,18 +2503,18 @@ checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
[[package]]
name = "serde"
version = "1.0.145"
version = "1.0.152"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b"
checksum = "bb7d1f0d3021d347a83e556fc4683dea2ea09d87bccdf88ff5c12545d89d5efb"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.145"
version = "1.0.152"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "81fa1584d3d1bcacd84c277a0dfe21f5b0f6accf4a23d04d4c6d61f1af522b4c"
checksum = "af487d118eecd09402d70a5d72551860e788df87b464af30e5ea6a38c75c541e"
dependencies = [
"proc-macro2",
"quote",
@@ -2496,9 +2523,9 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.85"
version = "1.0.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44"
checksum = "877c235533714907a8c2464236f5c4b2a17262ef1bd71f38f35ea592c8da6883"
dependencies = [
"itoa",
"ryu",
@@ -2845,9 +2872,9 @@ dependencies = [
[[package]]
name = "syn"
version = "1.0.102"
version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3fcd952facd492f9be3ef0d0b7032a6e442ee9b361d4acc2b1d0c4aaa5f613a1"
checksum = "1f4064b5b16e03ae50984a5a8ed5d4f8803e6bc1fd170a3cda91a1be4b18e3f5"
dependencies = [
"proc-macro2",
"quote",

5
Cargo.toml
View File

@@ -9,10 +9,13 @@ edition = "2021"
axum = { version = "0.6.1", features = ["ws", "headers"] }
axum-extra = { version = "0.4.2", features = ["spa"] }
axum-macros = "0.3.0"
colored = "2.0.0"
derive_more = "0.99.17"
futures-util = "0.3.24"
itertools = "0.10.5"
json = "0.12.4"
lazy_static = "1.4.0"
serde = "1.0.152"
serde_json = "1.0.91"
surrealdb = "1.0.0-beta.7"
tokio = {version = "1.21.0", features = ["full"]}
tokio-stream = "0.1.9"

51
errors.toml
View File

@@ -2,27 +2,42 @@
[[error]]
code = 11
category = "don't know yet"
message = "Something went wrong. Someone was too lazy to define an exact error for that."
message = "Something went wrong. //TODO: give this error it's own category and code."
[[error]]
category = "data_storage"
code = 28
message = "Something went wrong in the data storage."
[[error]]
code = 29
category = "data_storage:surrealdb_error"
message = "Something went wrong internally with the data-storage (surrealdb/rocksdb): {}"
[[error]]
code = 30
category = "data_storage:surrealdb_error"
message = "Could not create data_storeage at {}"
[[error]]
code = 31
category = "data_storage:surrealdb_error"
message = "No item with id 0. There definetly should be one"
[[error]]
code = 32
category = "data_storage"
message = "item {} was not found in storage"
[[error]]
code = 100
category = "don't know yet"
message = "You can't delete item number 0. This item contains mandatory Config for the Server."
[[error]]
category = "data_storage"
code = 128
message = "Something went wrong in the data storage."
[[error]]
code = 129
category = "data_storage:surrealdb_error"
message = "Something went wrong internally with the data-storage (surrealdb): {}"
[[error]]
code = 101
category = "key_missing::item0::num_of_items"
messate = "num_of_items missing in item0"
message = "num_of_items missing in item0"
[[error]]
code = 102
@@ -30,12 +45,12 @@ category = "key_missing::item0::next_free_id"
message = "next_free_id missing in item0"
[[error]]
_commit missing in item0:
message = "_commit missing in item0"
code = 103
category = "key_missing::item0::_commit"
[[error]]
_commit missing:
message = "_commit missing"
code = 103
category = "key_missing::_commit"
@@ -80,7 +95,13 @@ category = "don't know yet"
message = "module header not a String"
[[error]]
code 114
code = 114
category = "error with error system"
message = "the error with code {} was not found in the errors.toml file"
[[error]]
code = 115
category = "WebSocket Error"
message = "Error reviecing a message from the WebSocket"

173
src/error.rs
View File

@@ -1,72 +1,171 @@
use std::collections::HashMap;
use std::string::FromUtf8Error;
use toml::Value;
use colored::Colorize;
use serde::{Serialize, Deserialize};
use serde_json::Value as JsonValue;
use toml;
use crate::server::proto::{self, JsonMessage};
use crate::error::proto::ErrMessage;
use crate::server::proto::MizeMessage;
use crate::server::Peer;
lazy_static::lazy_static! {
static ref ERRORS: HashMap<u32, MizeError> = parse_errors();
pub static ref ERRORS: HashMap<u32, MizeError> = parse_errors();
}
fn parse_errors() -> HashMap<u32, MizeError>{
let toml_string = include_str!("../errors.toml");
let mut map = HashMap::new();
let binding = include_str!("../errors.toml")
.parse::<Value>()
.expect("Error pasting errors.toml file to Toml.");
let toml_error = binding.get("error")
.expect("Error getting error table from errors.toml file for Error handling");
if let Value::Array(error_arr) = toml_error {
for err in error_arr {
let code: u32 = if let Value::Integer(code) = err.get("code")
.expect("some error object in errors.toml has no code field.") {
u32::try_from(*code).expect(&format!("Error converting the error code {} &i64 to u32", code))
} else {
panic!("code value exists in errors.toml, but is not an Integer")
};
let message = if let Value::String(message) = err.get("message")
.unwrap_or(&Value::String("".to_string())) {message.to_string()} else {
panic!("message value exists in errors.toml on error obect with code {}, but is not a String", code)
};
let category = if let Value::String(category) = err.get("category")
.expect(&format!("the error object in errors.toml with code {} has no category field", code)) {category.to_string()} else {
panic!("category value exists in errors.toml on error obect with code {}, but is not a String", code)
};
let error = MizeError{code, message, category, extra_msg: None, caused_by_msg: None, code_location: None};
map.insert(code, error);
}
} else {
panic!("errors.toml File wrong. error path is not an array")
}
return map;
}
use json;
use toml;
use crate::server::proto;
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MizeError {
pub category: String,
pub code: u32,
pub message: String,
pub extra_msg: Option<String>,
pub caused_by_msg: Option<CausedByMessage>,
pub code_location: Option<MizeCodeLocation>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CausedByMessage {
client_id: u64,
msg: Box<JsonMessage>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MizeCodeLocation {
file: String,
line: u32,
column: u32,
}
impl From<&std::panic::Location <'_>> for MizeCodeLocation {
fn from(panic_location: &std::panic::Location) -> MizeCodeLocation {
let file = panic_location.file().to_string();
let line = panic_location.line();
let column = panic_location.column();
MizeCodeLocation {file, line, column}
}
}
impl MizeError {
#[track_caller]
pub fn new(code: u32) -> MizeError {
//let error = ERRORS.
let caller_location = std::panic::Location::caller();
MizeError {
category: "error with error system".to_string(),
code: 114,
message: "The error with code {} was not found in the errors.toml file".to_string(),
extra_msg: None
message: "The error with code {} was not found in the Errors that where imported from the errors.toml file at build time.".to_string(),
extra_msg: None,
caused_by_msg: None,
code_location: Some(caller_location.into()),
}
}
// pub fn format(&self, values: [str]){
// }
pub fn extra_msg(mut self, msg: &str) -> MizeError{
pub fn extra_msg(mut self, msg: &str) -> MizeError {
self.extra_msg = Some(msg.to_string());
return self;
}
pub fn to_json(self: MizeError) -> json::JsonValue{
if let Some(extra_msg) = self.extra_msg {
json::object!{
category: self.category,
code: self.code,
message: self.message,
extra_msg: extra_msg,
pub fn format(mut self, values: Vec<&str>) -> MizeError {
let count = 0;
for val in values {
self.message += &(" FORMAT_".to_string() + &format!("{}", count) + ": " + val)
}
return self;
}
fn set_msg(self, msg: MizeMessage, origin: Peer) -> MizeError {
MizeError::new(11).extra_msg("in set_msg TODO!!")
}
fn send(self) -> MizeError{
self
}
fn add_to_err_item(self) -> MizeError {
self
}
fn write_to_stderr(self) -> MizeError {
eprintln!("[{}] {}", "ERROR".red(), self.to_json());
self
}
pub fn handle(self) -> MizeError {
return self.send().add_to_err_item().write_to_stderr();
}
pub fn to_json(&self) -> JsonValue{
if let Ok(string) = serde_json::to_string(self) {
if let Ok(json) = serde_json::from_str(&string) {
return json;
} else {
return serde_json::json!({
"category": "error with error system",
"code": 114,
"message": "error converting a String to a JsonValue for the ErrMsg",
})
}
} else {
json::object!{
category: self.category,
code: self.code,
message: self.message,
}
return serde_json::json!({
"category": "error with error system",
"code": 114,
"message": "error converting an MizeError to String",
})
}
}
pub fn to_message(self: MizeError, origin: proto::Origin) -> proto::Message {
let data = self.to_json();
let mut msg = vec![crate::PROTO_VERSION, proto::MSG_ERROR];
msg.extend(json::stringify(data).into_bytes());
return proto::Message::from_bytes(msg, origin);
pub fn to_json_message(self) -> JsonMessage {
proto::JsonMessage::ErrMsg(
ErrMessage{
err: self,
}
)
}
}
impl From<FromUtf8Error> for MizeError {
@@ -76,7 +175,15 @@ impl From<FromUtf8Error> for MizeError {
}
impl From<surrealdb::Error> for MizeError {
fn from(sur_err: surrealdb::Error) -> MizeError{
fn from(sur_err: surrealdb::Error) -> MizeError {
MizeError::new(129)
}
}
impl From<serde_json::Error> for MizeError {
fn from(serde_err: serde_json::Error) -> MizeError {
MizeError::new(11).extra_msg("From an serde_json::Error. TODO: include actual error msg.")
}
}

159
src/server.rs
View File

@@ -2,6 +2,7 @@
pub mod server_utils;
pub mod itemstore;
pub mod proto;
//use futures_util::lock::Mutex;
@@ -11,6 +12,11 @@ use std::boxed::Box;
use std::path::Path;
use crate::error;
use crate::error::MizeError;
use crate::error::ERRORS;
use crate::server::proto::MizeMessage;
use serde_json::Value as JsonValue;
use serde::{Serialize, Deserialize};
use tokio::sync::mpsc::{Sender, channel, Receiver};
use tokio_stream::wrappers::ReceiverStream;
@@ -19,7 +25,6 @@ use std::fs;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use crate::server::proto::handle_mize_message;
use self::itemstore::Itemstore;
@@ -69,17 +74,42 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000;
//Version: 0
//";
#[derive(Clone, Debug)]
pub enum Peer {
Client(Client), //the client id
Module(Module), //Module_name
Upstream(String), //a hostname Type, but for now just a string
}
impl Peer {
pub async fn send<T> (&self, message: T) where T: Into<proto::MizeMessage>{
let message: proto::MizeMessage = message.into();
match self {
Peer::Client(client) => {client.tx.send(message).await;},
Peer::Module(module) => {module.tx.send(message).await;},
Peer::Upstream(_) => {},
}
}
pub fn get_id(&self) -> u64{
match self {
Peer::Client(client) => client.id,
Peer::Module(module) => module.client_id,
Peer::Upstream(_) => 0,
}
}
}
#[derive(Clone, Debug)]
pub struct Client {
// rx: UnboundedReceiverStream<Result<ws::Message, warp::Error>>,
tx: Sender<proto::Message>,
tx: Sender<proto::MizeMessage>,
id: u64,
}
#[derive(Clone, Debug)]
pub struct Module {
tx: Sender<proto::Message>,
tx: Sender<proto::MizeMessage>,
name: String,
client_id: u64,
kind: ModuleKind,
@@ -88,9 +118,10 @@ pub struct Module {
#[derive(Clone, Debug)]
pub enum ModuleKind {
Binary(),
Rust(),
Python(),
Node(),
JS(),
Lua(),
//connects to ws server
Extern(),
@@ -104,6 +135,16 @@ pub struct Render {
folder: String,
}
trait SendMize<T> where T: Into<MizeMessage> {
fn send(&self, msg: T);
}
impl<T> SendMize<T> for Sender<MizeMessage> where MizeMessage: From<T> {
fn send(&self, msg: T){
self.send(msg.into());
}
}
//pub struct Upstream maybe???? ... I'd say later....
@@ -113,7 +154,7 @@ pub struct Mutexes {
next_free_client_id: Arc<Mutex<u64>>,
clients: Arc<Mutex<Vec<Client>>>,
modules: Arc<Mutex<Vec<Module>>>,
subs: Arc<Mutex<HashMap<String, Vec<proto::Origin>>>>,
subs: Arc<Mutex<HashMap<String, Vec<Peer>>>>,
renders: Arc<Mutex<Vec<Render>>>,
//maybe a list of upstream servers??
itemstore: Arc<Mutex<Itemstore>>,
@@ -134,10 +175,6 @@ impl Mutexes {
}
}
fn test(){
println!("test func");
}
#[tokio::main]
pub async fn run_server(args: Vec<String>) {
/*
@@ -157,18 +194,11 @@ pub async fn run_server(args: Vec<String>) {
}
}
//load the errors.toml file and init the error system
//let error_toml_file = include_str!("../errors.toml");
//println!("ERRORS: {:?}", crate::error::ERRORS);
error::init();
println!("hello in run_server");
test();
//create the itemstore
let itemstore = crate::server::itemstore::Itemstore::new(mize_folder.clone() + "/db").await.expect("error creating itemstore");
//load modules and renders
let ren_mods = std::fs::read_dir(mize_folder.clone() + "/modules-renders")
let ren_mods = std::fs::read_dir(mize_folder.clone() + "/mr")
.expect("error reading the modules-renders dir in the mize-folder")
.filter(|entry| entry.as_ref().unwrap().file_type().unwrap().is_dir());
@@ -318,14 +348,30 @@ async fn handle_websocket_connection(
headers: HeaderMap,
){
let (mut socket_tx, mut socket_rx) = socket.split();
let (msg_tx, msg_rx): (Sender<proto::Message>, Receiver<proto::Message>) = mpsc::channel(SOCKET_CHANNEL_SIZE);
let (msg_tx, msg_rx): (Sender<proto::MizeMessage>, Receiver<proto::MizeMessage>) = mpsc::channel(SOCKET_CHANNEL_SIZE);
let mut msg_rx = ReceiverStream::new(msg_rx);
//my own forward
tokio::spawn(async move {
while let Some(msg) = msg_rx.next().await {
socket_tx.send(Message::Binary(msg.raw)).await;
match msg {
proto::MizeMessage::Json(json_msg) => {
if let Ok(msg) = serde_json::to_string(&json_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;
}
}
}
});
@@ -350,13 +396,13 @@ async fn handle_websocket_connection(
let err = MizeError::new(113)
.extra_msg("The mize-module Header could not be decoded into a String.");
msg_tx.send(err.to_message(proto::Origin::Module(module.clone()))).await;
msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
return;
};
println!("module {} connected", module.name);
mods.push(module.clone());
drop(mods);
proto::Origin::Module(module)
Peer::Module(module)
} else {
let mut cli = mutexes.clients.lock().await;
@@ -365,53 +411,82 @@ async fn handle_websocket_connection(
cli.push(client.clone());
*client_id += 1;
drop(cli);
proto::Origin::Client(client)
Peer::Client(client)
};
// Reading and broadcasting messages
// Reading messages
while let Some(result) = socket_rx.next().await {
let msg = result.expect("Error when getting message from WebSocket");
println!("got message: {:?}", msg);
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;
};
match msg {
Message::Binary(b) => {
println!("Recieved a Binary Message. Those are not implemented yet.")
},
Message::Text(text) => {
let json_msg: proto::JsonMessage = serde_json::from_str(&text).expect("error parsing json");
match json_msg {
proto::JsonMessage::ForItem(item_msg) => {
if let Err(err) = proto::handle_item_msg(item_msg, origin.clone(), mutexes.clone()).await {
let err_msg: MizeMessage = err.handle().into();
origin.send(err_msg).await;
};
},
_ => {println!("Got a non ItemMessage")},
}
}
let bytes = match msg {
Message::Binary(b) => b,
Message::Close(_) => {
match origin {
proto::Origin::Client(client) => {
Peer::Client(ref client) => {
//remove client from client list
let mut clients = mutexes.clients.lock().await;
let index = clients.iter().position(|client_iter| client_iter.id == client.id).expect("A Client disconected, that had never connected......");
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;
},
proto::Origin::Module(module) => {
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.....");
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;
}
proto::Origin::Upstream(_) => {
Peer::Upstream(_) => {
return;
}
}
}
_ => {
let err = MizeError::new(11)
.extra_msg("the message type was not Binary");
.extra_msg("unhandeld WebSocket-Message type")
.handle();
msg_tx.send(err.to_message(origin.clone())).await;
vec![0]
msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
},
};
if let Err(err) = handle_mize_message(
proto::Message::from_bytes(bytes, origin.clone()), mutexes.clone(),
).await {
msg_tx.send(err.to_message(origin.clone())).await;
};
};
}

272
src/server/itemstore.rs
View File

@@ -6,6 +6,10 @@ use itertools::Itertools;
use crate::error;
use crate::error::MizeError;
use crate::server::proto;
use crate::server::Mutexes;
use serde_json::Value as JsonValue;
use serde::{Serialize, Deserialize};
use super::proto::Delta;
@@ -16,37 +20,72 @@ pub struct Itemstore {
db: surrealdb::Datastore,
}
pub type Item = Vec<[Vec<u8>; 2]>;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Item {
#[serde(flatten)]
pub json: JsonValue,
}
impl Item {
pub fn get_id(&self) -> Result<String, MizeError> {
if let JsonValue::Object(ob) = &self.json{
if let Some(id_val) = ob.get("mize-id") {
if let JsonValue::String(id) = id_val {
return Ok(id.to_string())
} else {
return Err(MizeError::new(11).extra_msg("mize-id field in an item was not a String"));
}
} else {
return Err(MizeError::new(11).extra_msg("There was no mize-id field in an item"));
}
} else {
return Err(MizeError::new(11).extra_msg("An Item was not a JsonValue::Object"));
}
}
pub fn apply_delta(&mut self, delta: Delta) -> Result<(), MizeError> {
for (path, value) in delta.raw {
let mut to_replace = &mut self.json;
for name in path {
to_replace = if to_replace.get(&name).is_none() {
return Err(MizeError::new(11)
.extra_msg("Error applying a Delta to an Item. The path could not be followed."));
} else {
to_replace.get_mut(&name).ok_or(MizeError::new(11))?
};
}
*to_replace = value;
}
return Ok(())
}
pub fn from_bytes(vec: Vec<u8>) -> Result<Item, MizeError> {
let string = String::from_utf8(vec)?;
let item = serde_json::from_str(&string)?;
return Ok(item);
}
}
impl 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 ds = Datastore::new(&(String::from("file://") + &path[..] + "/db")[..]).await
.map_err(|_| MizeError::new(30).format(vec![&path]))?;
let mut tr = ds.transaction(true, false).await?;
match tr.get(vec!('0' as u8)).await {
match tr.get(0u64.to_le_bytes()).await {
Ok(Some(val)) => println!("item 0 already created"),
Ok(None) => {
let keys = vec![
"num_of_items".as_bytes().to_vec(),
"next_free_id".as_bytes().to_vec(),
"_commit".as_bytes().to_vec(),
"_type".as_bytes().to_vec(),
//"renders".as_bytes().to_vec(),
//"modules".as_bytes().to_vec(),
];
tr.set("0".to_string().into_bytes(), encode(keys)).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.set("0:renders".to_string().into_bytes(), Vec::new());
//tr.set("0:modules".to_string().into_bytes(), Vec::new());
let data = serde_json::json!({
"num_of_items": 0,
"next_free_id": 0,
"_commit": 0,
"mize-type": "mize-main",
"mize-id": "0"
});
tr.set(0u64.to_be_bytes(), data.to_string()).await?;
tr.commit().await?;
},
@@ -55,83 +94,50 @@ impl Itemstore {
return Ok(Itemstore {db:ds});
}
pub async fn create(&self, item: Item) -> Result<u64, MizeError> {
let mut new_id:u64 = 0;
pub async fn create(&self, mut item: Item, mutexes: Mutexes) -> Result<u64, MizeError> {
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");
let id_u64 = u64::from_be_bytes(id.clone().try_into().unwrap());
let mut keys: Vec<Vec<u8>> = Vec::new();
//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?;
keys.push("_commit".to_owned().into_bytes());
//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))?)
.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))?;
for field in item {
let mut key = format!("{}:", id_u64).into_bytes();
key.extend(field[0].clone());
keys.push(field[0].clone());
let num_of_items: u64 = json.get("num_of_items").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?;
let val = field[1].clone();
tr.set(key, val).await?;
}
tr.set(format!("{}", id_u64), encode(keys)).await?;
let new_next_id = new_id +1;
let new_num = num_of_items + 1;
//add mize-id field to item
let delta: Delta = serde_json::from_str(&format!("[[[\"mize-id\"], \"{}\"]]", new_id))?;
item.apply_delta(delta)?;
tr.set(new_id.to_be_bytes(), item.json.to_string()).await?;
tr.commit().await?;
return Ok(new_id)
//handle update to item0
let delta_item0: Delta = serde_json::from_str(&format!("[[[\"num_of_items\"], {}], [[\"next_free_id\"], {}]]", new_num, new_next_id))?;
proto::handle_update(proto::MizeId::Local(0), delta_item0, mutexes.clone(), None).await?;
return Ok(new_id);
}
pub async fn update(&self, update: proto::Update) -> Result<(), MizeError>{
println!("itemstore.update func");
pub async fn update(&self, id: u64, delta: Delta) -> Result<(), MizeError>{
let mut tr = self.db.transaction(true, true).await?;
//read keys
let mut old_keys = match tr.get(format!("{}", update.id).into_bytes()).await? {
Some(keys) => decode(keys),
None => {
return Err(MizeError::new(104)
.extra_msg("Internal Datastorage Error: the Index of the Item was not found"));
},
};
let item_bytes: Vec<u8> = tr.get(id.to_le_bytes()).await?
.ok_or(MizeError::new(11)
.extra_msg("The Item an update should be applied to does not exist."))?;
//write new values
for (mut key, delta) in update.raw {
let mut item = Item::from_bytes(item_bytes)?;
let mut store_key = (update.id.clone() + ":").into_bytes();
store_key.extend(key.clone());
item.apply_delta(delta)?;
let mut new_val = Vec::new();
if let Some(old_val) = tr.get(store_key.clone()).await? {
new_val = proto::apply_delta(old_val, delta)?;
tr.set(store_key.clone(), new_val.clone()).await?;
} else {
old_keys.push(key.clone());
new_val = proto::apply_delta(Vec::new(), delta)?;
tr.set(store_key.clone(), new_val.clone()).await?;
}
//remove field if new_val is empty
if (new_val.len() == 0) {
if let Some(key_index) = old_keys.iter().position(|x| *x == key.clone()){
old_keys.remove(key_index);
}
let mut store_key = update.id.clone().into_bytes();
store_key.push(':' as u8);
store_key.extend(key);
tr.del(store_key).await?;
} else {
tr.set(store_key.clone(), new_val).await?;
}
}
//write keys again
tr.set(update.id.into_bytes(), encode(old_keys)).await?;
tr.set(id.to_be_bytes(), serde_json::to_string(&item)?).await?;
tr.commit().await?;
return Ok(());
@@ -139,107 +145,29 @@ impl Itemstore {
pub async fn delete(&self, id_str: String) -> Result<(), error::MizeError>{
let id_u64: u64 = match id_str.parse() {
Ok(id) => id,
Err(err) => {
return Err(MizeError::new(110)
.extra_msg("std::num::ParseIntError while parsing id \"{}\" into an Integer"));
}
};
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_u64).into_bytes()).await?;
if let Some(keys) = keys_res {
for key in decode(keys) {
tr.del(key).await?;
}
} else {
return Err(MizeError::new(104)
.extra_msg("Internal Datastorage Error: the Index of the Item was not found, \
This means most likely that this item does not exist."));
}
tr.del(id_str.into_bytes()).await?;
tr.del(id.to_be_bytes()).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 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();
ke.extend(key.clone());
field[1] = match tr.get(ke).await {
Ok(Some(val)) => val,
Ok(None) => panic!("a field that should be there wasn't"),
Err(e) => panic!("Error while querying the db: {}", e),
};
item.push(field);
};
return Ok(item);
let item_res = tr.get(id.to_le_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(104)
.extra_msg("Internal Datastorage Error: the Index of the Item was not found"));
return Err(MizeError::new(32).format(vec![&format!("{}", id)]));
}
return Ok(item);
return Err(MizeError::new(11));
}
}
pub fn decode(bytes: Vec<u8>) -> Vec<Vec<u8>> {
let mut count = 0;
let mut index: Vec<Vec<u8>> = Vec::new();
let mut key: Vec<u8> = Vec::new();
loop {
if bytes[count] == '$' as u8 {
if key.len() != 0 {
index.push(key.clone());
key.clear();
}
} else if bytes[count] == '!' as u8 {
count += 1;
key.push(bytes[count]);
} else {
key.push(bytes[count]);
}
if count == bytes.len() -1 {
index.push(key.clone());
}
count += 1;
if count >= bytes.len() {break};
}
return index;
}
pub fn encode(index: Vec<Vec<u8>>) -> Vec<u8> {
let mut bytes: Vec<u8> = Vec::new();
for key in index {
let mut count = 0;
loop {
if key[count] == '!' as u8 || key[count] == '$' as u8 {
bytes.insert(count, '!' as u8);
count += 1;
}
count += 1;
if count >= key.len() {break};
}
bytes.push('$' as u8);
bytes.extend(key);
}
return bytes;
}

1154
src/server/proto.rs
View File

File diff suppressed because it is too large Load Diff