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

major server rewrite now all working

Browse Source
This commit is contained in:
Sebastian Moser
2023-01-05 11:19:43 +01:00
parent 134318c13b
commit 3cae3664c3
12 changed files with 972 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
errorcodes.yml
View File

@@ -1,3 +1,6 @@
Not sure yet:
code: 11
category: "don't know yet"
Can't delete item 0:
code: 100
@@ -31,3 +34,32 @@ could not read index of item:
_commit not 8 bytes long:
code: 105
category: "don't know yet"
error while parsing int:
code: 106
category: "parse_error"
error with message sent to the server:
code: 107
category: "faulty_message"
missing metadata:
code: 108
category: "faulty_message"
something that should be utf8 valid was not:
code: 109
category: "faulty_message"
ParseIntError:
code: 110
category: "faulty_message"
_commit can't be updated:
code: 111
category: "faulty_update"
no / on the end of id:
code: 112
category: "faulty_message"

2
first-render

Submodule first-render updated: 51815cb289...ace514689e

2
js-client

Submodule js-client updated: a3e42a9e23...1db0ffbc31

17
src/error.rs
View File

@@ -1,4 +1,6 @@
use std::string::FromUtf8Error;
use json;
use crate::server::proto;
@@ -19,11 +21,22 @@ impl MizeError {
return data;
}
pub fn to_message(self: MizeError) -> proto::Response {
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::Response::One(msg);
return proto::Message::from_bytes(msg, origin);
}
}
impl From<FromUtf8Error> for MizeError {
fn from(utf_err: FromUtf8Error) -> MizeError{
let err = MizeError {
code: 109,
kind: "faulty_message".to_string(),
message: "std::string::FromUtf8Error while parsing something from the message".to_string(),
};
return err;
}
}

0
src/integrated_types/async.rs Normal file
View File

0
src/integrated_types/keep_same.rs Normal file
View File

0
src/integrated_types/sync.rs Normal file
View File

2
src/integrated_types/test_same.rs Normal file
View File

@@ -0,0 +1,2 @@
pub fn main(){}

3
src/lib.rs
View File

@@ -5,6 +5,9 @@ static PROTO_VERSION: u8 = 1;
pub mod server;
pub mod error;
pub mod integrated_types {
pub mod test_same;
}

203
src/server.rs
View File

@@ -4,23 +4,25 @@ pub mod server_utils;
pub mod itemstore;
pub mod proto;
use futures_util::{FutureExt, StreamExt};
//use futures_util::lock::Mutex;
use futures_util::{FutureExt, StreamExt, SinkExt};
use warp::Filter;
use warp::ws::{WebSocket, self};
use std::collections::HashMap;
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tokio::sync::mpsc::{Sender, channel, Receiver};
use tokio_stream::wrappers::ReceiverStream;
use std::alloc::handle_alloc_error;
use std::fs;
use std::sync::{Arc};
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use crate::server::proto::{Response};
use crate::server::proto::handle_mize_message;
use self::itemstore::Itemstore;
//static API_PATH_STRING: &str = "$api";
static SOCKET_CHANNEL_SIZE: usize = 200;
// max file sizez in the mize/data folder (in bytes)
static MAX_INDEX_FILE_SIZE: usize = 1_800_000;
@@ -51,18 +53,58 @@ static MAX_DATA_FILE_SIZE: usize = 3_000_000_000;
//static VERSION_MESSAGE: &str = "\
//Version: 0
//";
//
#[derive(Clone, Debug)]
pub struct Client {
tx: UnboundedSender<Result<ws::Message, warp::Error>>
// rx: UnboundedReceiverStream<Result<ws::Message, warp::Error>>,
tx: Sender<proto::Message>,
id: u64,
}
pub fn run_server(args: Vec<String>) {
#[derive(Clone, Debug)]
pub struct Module {
tx: Sender<proto::Message>,
name: String,
client_id: u64,
//later
//registered_types: Vec<String>,
}
//pub struct Upstream maybe???? ... I'd say later....
// A collection of all the things, that most functions need
#[derive(Clone)]
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>>>>,
//maybe a list of upstream servers??
itemstore: Arc<Mutex<Itemstore>>,
}
impl Mutexes {
pub fn clone(mutexes: &Mutexes) -> Mutexes {
Mutexes {
next_free_client_id: Arc::clone(&mutexes.next_free_client_id),
clients: Arc::clone(&mutexes.clients),
modules: Arc::clone(&mutexes.modules),
subs: Arc::clone(&mutexes.subs),
itemstore: Arc::clone(&mutexes.itemstore),
}
}
}
#[tokio::main]
pub async fn run_server(args: Vec<String>) {
/*
* WHAT IT DOES
*
*/
//
//### get the mize_folder
let mut mize_folder = String::new();
for i in 1..args.len() {
@@ -74,13 +116,27 @@ pub fn run_server(args: Vec<String>) {
}
}
//create the itemstore
let itemstore = crate::server::itemstore::Itemstore::new(mize_folder.clone() + "/db").await.expect("error creating itemstore");
// A collection of all the things, that most functions need
let mutexes: Mutexes = Mutexes {
next_free_client_id: Arc::new(Mutex::new(0)),
clients: Arc::new(Mutex::new(Vec::new())),
subs: Arc::new(Mutex::new(HashMap::new())),
modules: Arc::new(Mutex::new(Vec::new())),
itemstore: Arc::new(Mutex::new(itemstore)),
};
//listen on the local unix socket
//local_socket_server(mize_folder);
//run the webserver
warp_server(mize_folder);
warp_server(mize_folder, mutexes).await;
}
#[tokio::main]
async fn warp_server(mize_folder: String) {
async fn warp_server(mize_folder: String, mutexes: Mutexes) {
/*
*
*
@@ -92,35 +148,30 @@ async fn warp_server(mize_folder: String) {
//## $api/rest/
// get itemstore
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));
let mutexes_clone = Mutexes::clone(&mutexes);
let socket_route = warp::path!("$api" / "socket")
.and(warp::ws())
.map(move |ws: warp::ws::Ws| {
let clients_clone = Arc::clone(&clients);
let itemstore_clone = Arc::clone(&itemstore_mutex);
ws.on_upgrade(move |socket| handle_socket_connection(
socket,
clients_clone,
itemstore_clone
)
)
let mutexes_clone = Mutexes::clone(&mutexes_clone);
ws.on_upgrade(move |socket| handle_websocket_connection(socket, mutexes_clone))
});
let render_route = warp::path!("$api" / "render").map(move || "render route");
let file_route = warp::path!("$api" / "file").map(move || "file route");
//temporary
let render_route = warp::path!("$api" / "render" / "first").map(move || {
let answer = fs::read_to_string("/home/sebastian/work/mize/first-render/src/main.js").unwrap();
warp::http::Response::builder().header("content-type", "text/javascript").body(answer)
});
let main_route = warp::path!("$api" / "client" / "main.js").map(move || {
let answer = fs::read_to_string("/home/sebastian/work/mize/js-client/src/main.js").unwrap();
warp::http::Response::builder().header("content-type", "application/javascript").body(answer)
});
let defuatl_route = warp::path::full().map(|path|{
//normally should be included in the binary, but for developing just gonna read the file
//so I don't have to recompile all the time
@@ -144,54 +195,96 @@ async fn warp_server(mize_folder: String) {
warp::serve(routes).run(([127, 0, 0, 1], 3000)).await;
}
async fn handle_socket_connection(
async fn handle_websocket_connection(
socket: WebSocket,
clients_clone: Arc<Mutex<Vec<Client>>>,
itemstore_clone: Arc<Mutex<Itemstore>>,
mutexes: Mutexes,
){
let (socket_tx, mut socket_rx) = socket.split();
let (client_tx, client_rx) = mpsc::unbounded_channel();
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 client_rx = UnboundedReceiverStream::new(client_rx);
tokio::spawn(client_rx.forward(socket_tx));
let mut msg_rx = ReceiverStream::new(msg_rx);
let mut cli = clients_clone.lock().await;
cli.push(Client{tx: client_tx.clone()});
// tokio::spawn(msg_rx.forward(socket_tx));
//my own forward
tokio::spawn(async move {
while let Some(msg) = msg_rx.next().await {
socket_tx.send(warp::ws::Message::binary(msg.raw)).await;
}
});
let mut cli = mutexes.clients.lock().await;
let mut client_id = mutexes.next_free_client_id.lock().await;
let client = Client{tx: msg_tx.clone(), id: *client_id};
cli.push(client.clone());
*client_id += 1;
drop(cli);
drop(client_id);
// Reading and broadcasting messages
while let Some(result) = socket_rx.next().await {
println!("got message ==================================");
let msg = result.expect("Error when gettin message from WebSocket");
let msg = result.expect("Error when getting message from WebSocket");
println!("got message: {:?}", msg);
let itemstore = &*itemstore_clone.lock().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) => {
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;},
};
if let Err(err) = handle_mize_message(
proto::Message::from_bytes(msg.clone().into_bytes(),proto::Origin::Client(client.clone())), mutexes.clone(),
).await {
msg_tx.send(err.to_message(proto::Origin::Client(client.clone()))).await;
};
};
}
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 handle_unix_socket_connection(){
//}
pub async fn send_to_all_subed_clients(id: u64, message: ws::Message, clients: Arc<Mutex<Vec<Client>>>){
}
//async fn handle_connection(con: Connection){
//}
//pub async fn send_to_all_clients(message: ws::Message, mutexes_clone: Mutexes){
// let clients = mutexes_clone.clients.lock().await;
// for client in &clients.clients[..] {
// client.tx.send(Ok(ws::Message::binary(message.clone())));
// }
// drop(clients);
// let modules = mutexes_clone.modules.lock().await;
// for module in &modules.modules[..] {
// module.tx.send(Ok(ws::Message::binary(message.clone())));
// }
//}
//pub async fn send_to_all_subbed_clients(id: String, message: ws::Message, mutexes_clone: Mutexes){
// let clients = mutexes_clone.clients.lock().await;
// for client in &clients.clients[..] {
// if client.sub.contains(&id) {
// client.tx.send(Ok(ws::Message::binary(message.clone())));
// }
// }
// let modules = mutexes_clone.modules.lock().await;
// for module in &modules.modules[..] {
// if module.sub.contains(&id) {
// module.tx.send(Ok(ws::Message::binary(message.clone())));
// }
// }
//}
//that does not work like this
//impl From<proto::Message> for Result<warp::ws::Message, warp::Error> {
// fn from(msg: proto::Message) -> Result<warp::ws::Message, warp::Error> {
// let msg = warp::ws::Message::binary(msg.raw);
// }
//}
//impl From<Result<warp::ws::Message, warp::Error>> for proto::Message {
// fn from(ws_msg: Result<warp::ws::Message, warp::Error>) -> proto::Message {
// let msg = ws_msg.expect("Error in From trait implementation: warp::ws::Messate to proto::Message");
// }
//}
//191 | cli.push(Client{tx: client_tx, id: client_id});
// = note: expected struct `tokio::sync::mpsc::Sender<proto::Message>`
// found struct `tokio::sync::mpsc::Sender<`

101
src/server/itemstore.rs
View File

@@ -5,7 +5,13 @@ use surrealdb::{Datastore, Key, Val};
use itertools::Itertools;
use crate::error;
use crate::error::MizeError;
use crate::server::proto;
use super::proto::Delta;
// The struct to do the storage and updates to it
// is responsible, that no illegal states can occur in the storage, by using transactions
// currently it's just a surrealdb Datastore. should be replaced by a completely custom File-System in the future.
pub struct Itemstore {
db: surrealdb::Datastore,
}
@@ -94,68 +100,83 @@ impl Itemstore {
return Ok(new_id)
}
pub async fn update(&self, id: u64, new_item: Item) -> Result<(), MizeError>{
pub async fn update(&self, update: proto::Update) -> Result<(), MizeError>{
let mut tr = self.db.transaction(true, true).await?;
let old_item_index = tr.get(update.id.clone().into_bytes()).await?;
let old_item = for key in old_item_index {
let mut key_get = format!("{}:", update.id).into_bytes();
key_get.extend(key.clone());
tr.get(key_get);
};
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?;
}
//check the _type of the item and maybe run type-code
//TODO: typecode
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 {
//read keys
let mut keys = match tr.get(format!("{}", update.id).into_bytes()).await? {
Some(keys) => decode(keys),
None => {
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 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(),
});
},
};
let mut commit_num = u64::from_be_bytes(val);
commit_num += 1;
tr.set(commit_key, commit_num.to_be_bytes()).await?;
//write new values
for (mut key, delta) in update.raw {
let mut store_key = (update.id.clone() + ":").into_bytes();
store_key.append(&mut key);
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 {
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(),
});
keys.push(key.clone());
new_val = proto::apply_delta(Vec::new(), delta)?;
}
//remove field if new_val is empty
if (new_val.len() == 0) {
if let Some(key_index) = keys.iter().position(|x| *x == key.clone()){
keys.remove(key_index);
}
tr.del(key).await?;
} else {
tr.set(store_key.clone(), new_val).await?;
}
}
//write keys again
tr.set(update.id.into_bytes(), encode(keys)).await?;
tr.commit().await?;
return Ok(());
}
pub async fn delete(&self, id: u64) -> Result<(), error::MizeError>{
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 {
code: 110,
kind: "faulty_message".to_string(),
message: format!("std::num::ParseIntError while parsing id \"{}\" into an Integer", id_str),
});
}
};
let mut tr = self.db.transaction(true, true).await?;
let keys_res = tr.get(format!("{}", id).into_bytes()).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?;
@@ -168,7 +189,7 @@ impl Itemstore {
});
}
tr.del(format!("{}", id)).await?;
tr.del(id_str.into_bytes()).await?;
tr.commit().await?;
return Ok(());

860
src/server/proto.rs
View File

@@ -1,5 +1,4 @@
// 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)
@@ -17,6 +16,16 @@
// 15: get_and_sub
// 16: get_fields
// 17: error
// 18: give_fields
// ID
// just a number: localy storeWEK_K2JNQqkd
// #hello.mize.works#034.312 upstream in the internet ... domain specified
// !youtube!WEK_K2JNQqk upstream in a module ... if module is not on local server
// forwarded to configured upstream server (would happen, if we are running in daemon mode)
pub static VERSION: u8 = 1;
pub static MSG_GET: u8 = 1;
pub static MSG_GIVE: u8 = 2;
@@ -35,32 +44,221 @@ 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;
pub static MSG_GIVE_FIELDS: u8 = 18;
use std::time::Duration;
//use std::panic::update_hook;
use std::vec;
use std::collections::HashMap;
use crate::server::itemstore;
use crate::server::Mutexes;
use crate::server::itemstore::encode;
use crate::error;
use crate::server::Client;
use std::sync::Arc;
use futures_util::stream::Collect;
use tokio::sync::{Mutex, mpsc};
use crate::server;
use crate::error::MizeError;
pub enum Response {
One(Vec<u8>),
All(Vec<u8>),
AllSubbed(u64, Vec<u8>),
None,
}
#[derive(Clone)]
pub struct Message {
msg: Vec<u8>,
pub raw: Vec<u8>,
version: u8,
cmd: u8,
origin: Origin,
index: usize,
meta_gotten: bool,
id: Option<String>,
}
//pub enum Response {
// One(Vec<u8>),
// All(Vec<u8>),
// AllSubbed(String, Vec<u8>),
// None,
//}
#[derive(Clone, Debug)]
pub enum Origin {
Client(server::Client), //the client id
Module(server::Module), //Module_name
Upstream(String), //a hostname Type, but for now just a string
}
impl Origin {
pub async fn send(&self, message: Message){
match self {
Origin::Client(client) => {client.tx.send(message).await;},
Origin::Module(module) => {module.tx.send(message).await;},
Origin::Upstream(_) => {},
}
}
pub fn get_id(&self) -> u64{
match self {
Origin::Client(client) => client.id,
Origin::Module(module) => module.client_id,
Origin::Upstream(_) => 0,
}
}
}
#[derive(Clone)]
pub struct Update {
pub id: String,
origin: Origin,
pub raw: Vec<(Vec<u8>, Delta)>,
}
#[derive(Clone, Debug)]
pub struct Delta {
raw: Vec<DeltaCmd>,
}
#[derive(Clone, Debug)]
pub enum DeltaCmd {
Replace(u32, u32, Vec<u8>),
Insert(u32, u32, Vec<u8>),
Delete(u32, u32),
}
impl Delta {
pub fn from_bytes(bytes: Vec<u8>) -> Result<Delta, MizeError> {
let out_of_bytes = Err(MizeError{
code: 107,
kind: "faulty_message".to_string(),
message: "There were not eneough Bytes in the message to complete a Delta.".to_string()
});
let mut bytes = bytes.iter();
let mut delta = Delta{raw: Vec::new()};
while true {
let cmd = if let Some(cmd) = bytes.next() {
cmd
} else {
//end of bytes
return Ok(delta);
};
match cmd {
0 => {
let mut start_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}};
let start = u32::from_be_bytes(start_bytes);
let mut stop_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}};
let stop = u32::from_be_bytes(stop_bytes);
let mut val: Vec<u8> = Vec::new();
for i in 0..stop-start {val.push(if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;})};
delta.raw.push(DeltaCmd::Replace(start, stop, val));
},
1 => {
let mut start_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}};
let start = u32::from_be_bytes(start_bytes);
let mut stop_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}};
let stop = u32::from_be_bytes(stop_bytes);
let mut val: Vec<u8> = Vec::new();
for i in 0..stop-start {val.push(if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;})};
delta.raw.push(DeltaCmd::Insert(start, stop, val));
},
2 => {
let mut start_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {start_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}};
let start = u32::from_be_bytes(start_bytes);
let mut stop_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {stop_bytes[i] = if let Some(byte) = bytes.next(){*byte} else {return out_of_bytes;}};
let stop = u32::from_be_bytes(stop_bytes);
delta.raw.push(DeltaCmd::Delete(start, stop));
},
_ => {
return Err(MizeError{
code: 107,
kind: "faulty_message".to_string(),
message: format!("The update message included a DeltaCmd of {}, only 0,1,2 (for: replace, insert, delete) are allowed.",cmd)
});
},
}
};
return Ok(delta);
}
}
impl Update {
pub async fn change_whole_val(id: String, key: Vec<u8>, new_val: Vec<u8>, origin: Origin, mutexes: Mutexes) -> Result<Update, MizeError> {
let mut old_val_len: u32 = 0;
let itemstore = mutexes.itemstore.lock().await;
let id_u64 = match id.parse::<u64>(){
Ok(id) => id,
Err(err) => {
return Err(MizeError{
code: 11,
kind: "don't know yet".to_string(),
message: "Error while parsing id to u64 (std::num::ParseIntError) in order to get it from the itemstore. (get or get_sub message)".to_string(),
});
}
};
let item = itemstore.get(id_u64).await?;
for field in item {
if (field[0] == key){
old_val_len = field[1].len() as u32;
}
}
let delete_all = DeltaCmd::Delete(0, old_val_len);
let replace_all = DeltaCmd::Replace(0, new_val.len() as u32, new_val);
let deltas = Delta{raw: vec![delete_all, replace_all]};
let update: Update = Update { id, raw: vec![(key, deltas)], origin};
return Ok(update);
}
pub fn to_message(self, origin: Origin) -> Message{
return Message::from_update(self, origin);
}
pub fn from_message(mut message: Message) -> Result<Update, MizeError> {
message.index = 2;
message.id = None;
let id = message.get_id()?;
let mut update = Update{id, raw: Vec::new(), origin: message.clone().origin};
let num_of_updates = message.get_u32();
for i in 0..num_of_updates as usize {
let key_len = message.get_u32();
let key = message.get_bytes(key_len as usize);
let update_len = message.get_u32();
let update_bytes = message.get_bytes(update_len as usize);
//get DeltaCmds
let delta = Delta::from_bytes(update_bytes)?;
update.raw.push((key, delta));
}
return Ok(update);
}
}
impl Message {
pub fn new(vec: Vec<u8>) -> Message {
pub fn from_bytes(vec: Vec<u8>, origin: Origin) -> Message {
// this weired version does not panic, while others do for some unknown reason
let mut count = 0;
@@ -77,17 +275,214 @@ impl Message {
}
count += 1;
if count == 2 {break};
};
Message { msg: vec, version: version, cmd: cmd, index: 2 }
Message { raw: vec, version, cmd, index: 2 , origin, meta_gotten: false, id: None}
}
pub async fn send(self, origin: &Origin) {
origin.send(self).await;
}
pub fn get_id(&mut self) -> u64 {
self.get_u64()
pub async fn forward_module(mut self, id: String, mutexes: Mutexes) -> Result<(), MizeError>{
let mut meta: HashMap<String, String> = HashMap::new();
meta.insert("c".to_string(), format!("{}", self.origin.get_id()));
self.set_metadata(meta)?;
let mut id_iter = id.chars();
//scip the first "!"
id_iter.next();
let mod_name = id_iter.take_while(|&ch| ch != '!').collect::<String>();
let modules = mutexes.modules.lock().await;
for module in &*modules {
if (module.name == mod_name) {
self.send(&Origin::Module(module.clone()));
return Ok(());
}
}
return Ok(());
}
pub fn from_update(update: Update, origin: Origin) -> Message{
let mut message: Vec<u8> = vec![VERSION, MSG_UPDATE];
// id
message.extend(update.id.into_bytes());
message.push("/".to_string().into_bytes()[0]);
// update_len
let update_len = update.raw.len() as u32;
message.extend(update_len.to_be_bytes());
for update in update.raw {
//key_len
let key_len = update.0.len() as u32;
message.extend(key_len.to_be_bytes());
// key
message.extend(update.0);
//delta_len
let mut delta_len: u32 = 0;
// deltas
let mut delta_part: Vec<u8> = Vec::new();
// loop
for delta in update.1.raw {
match delta {
DeltaCmd::Replace(start, stop, data) => {
delta_len += 9; //for len of start, stop and cmd
delta_len += data.len() as u32;
delta_part.push(0);
delta_part.extend(start.to_be_bytes());
delta_part.extend(stop.to_be_bytes());
delta_part.extend(data)
},
DeltaCmd::Insert(start, stop, data) => {
delta_len += 9; //for len of start, stop and cmd
delta_len += data.len() as u32;
delta_part.push(1);
delta_part.extend(start.to_be_bytes());
delta_part.extend(stop.to_be_bytes());
delta_part.extend(data)
},
DeltaCmd::Delete(start, stop) => {
delta_len += 9; //for len of start, stop and cmd
//
delta_part.push(2);
delta_part.extend(start.to_be_bytes());
delta_part.extend(stop.to_be_bytes());
},
}
};
message.extend(delta_len.to_be_bytes());
message.extend(delta_part);
}
return Message::from_bytes(message, origin);
}
pub fn set_metadata(&mut self, metadata: HashMap<String, String>) -> Result<(), MizeError> {
let mut meta_vec: Vec<u8> = Vec::new();
meta_vec.push('{' as u8);
for (mut key, mut val) in metadata.iter(){
meta_vec.append(&mut key.clone().into_bytes());
meta_vec.push(';' as u8);
meta_vec.append(&mut val.clone().into_bytes());
meta_vec.push('/' as u8);
}
meta_vec.remove(meta_vec.len());
meta_vec.push('}' as u8);
self.raw.splice(2..2, meta_vec);
return Ok(());
}
pub fn get_metadata(&mut self) -> Result<HashMap<String, String>, MizeError> {
if !self.has_metadata() {
return Err(MizeError {
code: 108,
kind: "faulty_message".to_string(),
message: "The message should have medatata, but it hasn't. (cmd <= 128)".to_string(),
})
};
let mut meta_vec:Vec<u8> = Vec::new();
let mut msg_iter = self.raw[self.index..].iter();
if (msg_iter.nth(0).unwrap() != &('{' as u8)){
return Err(MizeError {
code: 107,
kind: "faulty_message".to_string(),
message: "Error while parsing the metadata of a message.".to_string(),
})
};
for ch in msg_iter {
meta_vec.push(*ch);
if (ch == &('}' as u8)){
let mut map: HashMap<String, String> = HashMap::new();
let string = String::from_utf8(meta_vec)?;
let sp: Vec<&str> = string.split("/").collect();
for st in sp.clone() {
let mut split = sp.split(|c| *c == ";");
let key = split.nth(0).unwrap()[0];
let val = split.nth(1).unwrap()[0];
map.insert(key.to_string(), val.to_string());
}
self.meta_gotten = true;
return Ok(map);
}
};
return Err(MizeError {
code: 107,
kind: "faulty_message".to_string(),
message: "Error while parsing the metadata of a message.".to_string(),
})
}
pub fn has_metadata(&mut self) -> bool {
self.cmd >= 128
}
pub fn get_id(&mut self) -> Result<String, MizeError> {
//get the metadata just to increment the index
if let Some(id) = &self.id {
return Ok(id.clone());
}
if (self.has_metadata() && !self.meta_gotten){
self.get_metadata();
}
let mut ch: u8 = 0;
let mut id = String::new();
for ch in &self.raw[self.index..] {
if (*ch == 47 as u8){
self.index += 1;
self.id = Some(id.clone());
return Ok(id);
} else {
id.push(*ch as char);
self.index += 1;
}
}
return Err(MizeError {
code: 112,
kind: "faulty_message".to_string(),
message: "There is no / to indicate where the id ends".to_string(),
});
// let len = self.raw.len();
// while self.index < 10000 {
// if (self.index == len){
// return Err(MizeError {
// code: 112,
// kind: "faulty_message".to_string(),
// message: "There is no / to indicate where the id ends".to_string(),
// })
// }
// let ch: u8 = self.raw[self.index];
// if (ch == 47) {break};
// self.index += 1;
// }
//skip "/"
// self.index += 1;
// self.id_gotten = true;
// return Ok(id);
}
pub fn get_u32(&mut self) -> u32 {
let tmp = &self.msg[self.index..self.index + 4];
let tmp = &self.raw[self.index..self.index + 4];
let num: u32 = u32::from_be_bytes([tmp[0], tmp[1], tmp[2], tmp[3]]);
self.index += 4;
@@ -95,7 +490,7 @@ impl Message {
}
pub fn get_u64(&mut self) -> u64 {
let tmp = &self.msg[self.index..self.index + 8];
let tmp = &self.raw[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;
@@ -103,7 +498,7 @@ impl Message {
}
pub fn get_bytes(&mut self, n: usize) -> Vec<u8>{
let tmp = &self.msg[self.index..self.index + n];
let tmp = &self.raw[self.index..self.index + n];
self.index += n;
return tmp.to_vec();
@@ -111,65 +506,84 @@ impl Message {
}
const VERSION: u8 = 1;
pub async fn handle_mize_message(
mut message: Message,
itemstore: &itemstore::Itemstore,
) -> Vec<Response> {
let old_message = message.msg.clone();
mutexes: Mutexes,
) -> Result<(), MizeError> {
//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");
//println!("after chaos");
////////////////////////////////////////////////////////////
//get and get_and_sub cmd from client or module
if ((message.cmd == MSG_GET || message.cmd == MSG_GET_AND_SUB) &&
(matches!(&message.origin, Origin::Client(_)) || matches!(&message.origin, Origin::Module(_))))
{
// for some weired reason this stupid solution does not panic, while the obove one and vec[0]
// do
let mut count = 0;
let mut cmd: u8 = 0;
let mut version: u8 = 0;
for i in old_message.clone() {
if count == 0 {
version = i;
}
if count == 1 {
cmd = i
}
count += 1;
let id: String = message.get_id()?;
let client_id: u64 = match message.origin {
Origin::Client(ref Client) => Client.id,
Origin::Module(ref Module) => Module.client_id,
Origin::Upstream(_) => 0,
};
//println!("VERSION: {}", version);
//println!("CMD: {}", cmd);
//sub to item in case of MSG_GET_AND_SUB
if (message.cmd == MSG_GET_AND_SUB) {
let mut subs = mutexes.subs.lock().await;
if let Some(mut vec_ids) = subs.get_mut(&id) {
vec_ids.push(message.origin.clone());
} else {
subs.insert(id.clone(), vec![message.origin.clone()]);
};
drop(subs);
};
//forward to module in case the item is from a Module
//append medatata
let first_char = id.chars().nth(0).ok_or(MizeError{
code: 11,
kind: "don't know yet".to_string(),
message: "Somehow the id is empty. while handling a get or get_sub message.".to_string(),
})?;
if (first_char == '!'){
message.clone().forward_module(id.clone(), mutexes.clone()).await;
}
//forward to upstream server in case the item is from another Server
if (first_char == '#'){
return Ok(());
}
match cmd {
//get
1 => {
//let id_bytes = *&message[2..9].to_owned().clone();
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:
// u8: version
// u8: cmd (2 for give)
// u64: id
// id: terminated by "/"
// u32: num_of_fields
// as often as num_of_fields:
// u64: key_len
// key_len: key
// u64: val_len
// val_len: val
let mut answer: Vec<u8> = vec![VERSION, 2];
answer.extend(id.to_be_bytes());
let mut answer: Vec<u8> = vec![VERSION, MSG_GIVE];
answer.extend(format!("{}/", id).into_bytes());
let mut item = match itemstore.get(id).await {
let itemstore = mutexes.itemstore.lock().await;
let id_u64 = match id.parse::<u64>(){
Ok(id) => id,
Err(err) => {
return Err(MizeError{
code: 11,
kind: "don't know yet".to_string(),
message: "Error while parsing id to u64 in order to get it from the itemstore. (get or get_sub message)".to_string(),
});
}
};
let mut item = match itemstore.get(id_u64).await {
Ok(item) => item,
Err(err) => {return vec![err.to_message()]},
Err(err) => {return Err(err)},
};
let num_of_fields = item.len() as u32;
@@ -184,83 +598,59 @@ pub async fn handle_mize_message(
answer.extend(val_len.to_be_bytes());
answer.extend(field[1].clone());
}
return vec![Response::One(answer)];
},
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 => {
let mut answer: Vec<u8> = message.msg.clone();
answer[1] = 10;
let id = message.get_id();
let mut item = match itemstore.get(id).await {
Ok(item) => item,
Err(err) => {return vec![err.to_message()]},
let msg = Message::from_bytes(answer, message.origin.clone());
message.origin.send(msg).await;
};
let num_of_updates = message.get_u32();
for i in 0..num_of_updates as usize {
let key_len = message.get_u32();
let key = message.get_bytes(key_len as usize);
//println!("KEY: {}", String::from_utf8(key.clone()).expect("here utf-8"));
let update_len = message.get_u32();
let update = message.get_bytes(update_len as usize);
let mut found = false;
for a in 0..item.len() as usize {
if item[a][0] == key {
item[a][1] = apply_update(&item[a][1], &update);
found = true;
break;
}
}
if found == false {
let index = item.len() +1;
//item[index][1] = apply_update(&item[index][1], &update)
item.push([key.clone(), apply_update(&item[index][1], &update)])
}
}
if let Err(err) = itemstore.update(id, item).await {
return vec![err.to_message()];
////////////////////////////////////////////////////////////
//give from module
if (message.cmd == MSG_GIVE && matches!(&message.origin, Origin::Module(_))){
//TODO
};
return vec![Response::All(answer)];
},
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 {
let err = error::MizeError{
////////////////////////////////////////////////////////////
//update_request from module and client
if (message.cmd == MSG_UPDATE_REQUEST && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){
let id: String = message.get_id()?;
let first_char = id.chars().nth(0).ok_or(MizeError{
code: 11,
kind: "don't know yet".to_string(),
message: "Somehow the id is empty. while handling a get or get_sub message.".to_string(),
})?;
if (first_char == '!'){
message.clone().forward_module(id, mutexes.clone()).await;
}
let update = Update::from_message(message.clone())?;
handle_update(update, mutexes, message.origin.clone()).await?;
return Ok(());
};
////////////////////////////////////////////////////////////
//delete from Client or Module
if (message.cmd == MSG_DELETE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){
let id: String = message.get_id()?;
if id == "0".to_string() {
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()];
return Err(err);
}
}
if let Err(err) = itemstore.delete(id).await {
return vec![err.to_message()];
}
return response;
},
//create
12 => {
////////////////////////////////////////////////////////////
//create from module and Client
if (message.cmd == MSG_CREATE && (matches!(&message.origin, Origin::Module(_)) || matches!(&message.origin, Origin::Client(_)))){
let num_of_fields = message.get_u32();
let mut item: Vec<[Vec<u8>; 2]> = Vec::new();
@@ -276,118 +666,158 @@ pub async fn handle_mize_message(
item.push([key, val]);
}
if let Err(err) = itemstore.create(item).await {
return vec![err.to_message()];
};
let itemstore = mutexes.itemstore.lock().await;
let new_id = itemstore.create(item).await?;
let anser: Vec<u8> = vec![1,1];
return vec![Response::None];
},
13 => {return Vec::new()},
14 => {return Vec::new()},
_ => {return Vec::new()},
let mut item0: Vec<[Vec<u8>; 2]> = Vec::new();
match itemstore.get(0).await {
Err(err) => {return Err(err);},
Ok(item) => {item0 = item;},
}
//update item0
let mut next_free_id: Vec<u8> = Vec::new();
let mut num_of_items: Vec<u8> = Vec::new();
for field in item0 {
if field[0] == "next_free_id".to_string().into_bytes() {
// soooooo ugly
let mut tmp = u64::from_be_bytes(field[1].clone().try_into().unwrap());
tmp += 1;
next_free_id = tmp.to_be_bytes().to_vec();
}
if field[0] == "num_of_items".to_string().into_bytes() {
// soooooo ugly
let mut tmp = u64::from_be_bytes(field[1].clone().try_into().unwrap());
tmp += 1;
num_of_items = tmp.to_be_bytes().to_vec();
}
}
//because Update::change_whole_val trys to lock the itemstore as well
drop(itemstore);
let update_next_id: Update = Update::change_whole_val("0".to_string(), "next_free_id".to_string().into_bytes(), next_free_id.clone(), message.origin.clone(), mutexes.clone()).await?;
let mut update: Update = Update::change_whole_val("0".to_string(), "num_of_items".to_string().into_bytes(), num_of_items.clone(), message.origin.clone(), mutexes.clone()).await?;
update.raw.extend(update_next_id.raw);
handle_update(update, mutexes.clone(), message.origin.clone()).await?;
//send back MSG_CREATED_ID
let mut created_id_message = vec![VERSION, MSG_CREATED_ID];
created_id_message.extend(new_id.to_be_bytes());
message.origin.send(Message::from_bytes(created_id_message, message.origin.clone())).await;
}
////////////////////////////////////////////////////////////
//for every other message
return Ok(());
}
pub async fn handle_update(mut update: Update, mutexes: Mutexes, origin: Origin) -> Result<(), MizeError>{
//handle type-code
//and either call itemstore.update(), send the update to the module or upstream, or spawn another
//update by calling handle_update()
///////////////////////////////// TYPE CODE ////////////////////////////////////////
//can change the update and spawn new updates (to a different item) (which would call the handle_update func again)
//the "type code" for the Null type (code that effects every item)
//eg. increment _commit
let itemstore = mutexes.itemstore.lock().await;
let id_u64 = match update.id.clone().parse::<u64>(){
Ok(id) => id,
Err(err) => {
return Err(MizeError{
code: 11,
kind: "don't know yet".to_string(),
message: "Error while parsing id to u64 in order to get it from the itemstore. (get or get_sub message)".to_string(),
});
}
};
let item = itemstore.get(id_u64).await?;
let commit_pos = item.iter().position(|field| field[0] == "_commit".to_string().into_bytes()).ok_or(MizeError{
code: 103,
kind: "key_missing::_commit".to_string(),
message: format!("Item {} has not _commit", id_u64),
})?;
let cur_commit = u64::from_be_bytes(item[commit_pos][1].clone().try_into().map_err(|_|MizeError{
code: 105,
kind: "don't know yet".to_string(),
message: "_commit is not 8 bytes long".to_string(),
})?);
//because Update::change_whole_val trys to lock the itemstore as well
drop(itemstore);
update.raw.extend(Update::change_whole_val(update.id.clone(), "_commit".to_string().into_bytes(), (cur_commit +1).to_be_bytes().to_vec(), update.origin.clone(), mutexes.clone()).await?.raw);
//run some type-code here
// if let Some(index) = item.iter().position(|&field| field[0] == "_type".to_string().into_bytes()) {
// let type_string = String::from_utf8(item[index][0])?;
// let types = type_string.split(" ").collect::Vec<String>();
// for &typ[..] in types {
// match typ {
// "test-same" => crate::integrated_types::test_same::main(update, item);
// }
// };
// };
//if update is changing _type, then run type::create(), or type::destroy(), for types that got added or deleted
//apply item in itemstore
let itemstore = mutexes.itemstore.lock().await;
itemstore.update(update.clone()).await?;
//send update to clients/modules
let message = update.clone().to_message(origin.clone());
let subs = mutexes.subs.lock().await;
let empty_vec = &Vec::new();
let subbed_origins = subs.get(&update.id).unwrap_or(empty_vec);
for origin in subbed_origins {
message.clone().send(origin).await;
}
Ok(())
}
// just like all of this crate, definetly could be done with less clones().... and better error
// handling
pub fn apply_update(val: &Vec<u8>, updates: &Vec<u8>) -> Vec<u8>{
println!("VAL: {:?}", val.clone());
println!("Update: {:?}", updates.clone());
let not_enough_bytes = "not enough bytes in update";
let mut val = val.clone();
let mut update_iter = updates.clone().into_iter();
while true {
if let Some(operation) = update_iter.next(){
let mut new_val: Vec<u8> = Vec::new();
let mut val_iter = val.clone().into_iter();
match operation {
pub fn apply_delta(mut val: Vec<u8>, delta: Delta) -> Result<Vec<u8>, MizeError>{
let mut update_iter = delta.raw.iter();
for delta_cmd in delta.raw.iter() {
match delta_cmd {
//r,start:u32,stop:u32,bytes start..stop
0 => {
//get start and stop
let mut start_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {start_bytes[i] = update_iter.next().expect(not_enough_bytes);};
let start = u32::from_be_bytes(start_bytes);
let mut stop_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {stop_bytes[i] = update_iter.next().expect(not_enough_bytes);};
let stop = u32::from_be_bytes(stop_bytes);
//add the stuff before
for i in 0..start {new_val.push(val_iter.next().expect(not_enough_bytes));};
//add the new stuff
for i in start..stop {new_val.push(update_iter.next().expect(not_enough_bytes));};
//skip all the bytes that should be replaced
for i in 0..stop-start {val_iter.next().expect(not_enough_bytes);}
//add stuff after
for byte in val_iter {
new_val.push(byte);
DeltaCmd::Replace(start, stop, bytes) => {
let bytes = (*bytes).clone();
let start = *start as usize;
let stop = *stop as usize;
let val_len = val.len();
if (stop > val_len) {
val.extend(vec![0; stop-val_len])
}
val.splice(start..stop, bytes);
},
//i,start:u32, stop:u32, bytes stop-start
1 => {
//get start and stop
let mut start_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {start_bytes[i] = update_iter.next().expect(not_enough_bytes);};
let start = u32::from_be_bytes(start_bytes);
let mut stop_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {stop_bytes[i] = update_iter.next().expect(not_enough_bytes);};
let stop = u32::from_be_bytes(stop_bytes);
//add the stuff before
for i in 0..start {new_val.push(val_iter.next().expect(not_enough_bytes));};
//add the new stuff
for i in start..stop {new_val.push(update_iter.next().expect(not_enough_bytes));};
//add stuff after
for byte in val_iter {
new_val.push(byte);
}
//while true {
//if let Some(byte) = update_iter.next() {
//new_val.push(byte);
//} else {break;}
//}
DeltaCmd::Insert(start, stop, bytes) => {
val.splice((*start as usize)..(*start as usize), (*bytes).clone());
},
//d,start:u32,stop:u32
2 => {
//get start and stop
let mut start_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {start_bytes[i] = update_iter.next().expect(not_enough_bytes);};
let start = u32::from_be_bytes(start_bytes);
let mut stop_bytes: [u8; 4] = [0,0,0,0];
for i in 0..4 {stop_bytes[i] = update_iter.next().expect(not_enough_bytes);};
let stop = u32::from_be_bytes(stop_bytes);
//add the stuff before
for i in 0..start {new_val.push(val_iter.next().expect("not_enough_bytes in val_iter"));};
//skip all the bytes that should be replaced
for i in 0..stop-start {let h = val_iter.next().expect("not_enough_bytes in val_iter"); println!("VAL ITER: {}", h);}
//add stuff after
for byte in val_iter {
new_val.push(byte);
}
DeltaCmd::Delete(start, stop) => {
val.splice((*start as usize)..(*stop as usize), Vec::new());
},
_ => {panic!("unknown update command")}
}
val = new_val;
} else {break;}
};
return val;
}
return Ok(val);
}