fresh start

This commit is contained in:
Sebastian Moser
2023-12-22 22:43:54 +01:00
parent 1e464fd46a
commit 512ca994e2
17 changed files with 19 additions and 4747 deletions

2826
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,24 +1,23 @@
[package] [package]
name = "mize" name = "mize"
version = "0.1.0" version = "0.1.1"
edition = "2021" edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies] [dependencies]
axum = { version = "0.6.1", features = ["ws", "headers"] } #axum = { version = "0.6.1", features = ["ws", "headers"] }
axum-extra = { version = "0.4.2", features = ["spa"] } #axum-extra = { version = "0.4.2", features = ["spa"] }
axum-macros = "0.3.0" #axum-macros = "0.3.0"
colored = "2.0.0" #colored = "2.0.0"
derive_more = "0.99.17" #derive_more = "0.99.17"
futures-util = "0.3.24" #futures-util = "0.3.24"
itertools = "0.10.5" #itertools = "0.10.5"
lazy_static = "1.4.0" #lazy_static = "1.4.0"
serde = "1.0.152" #rmp-serde = "1.1.1"
serde_json = "1.0.91" #rmpv = { version = "1.0.0", features = ["serde", "with-serde"] }
surrealdb = { version = "1.0.0-beta.7", features = ["kv-rocksdb"] } #serde = "1.0.152"
tokio = {version = "1.21.0", features = ["full"]} #serde_json = "1.0.91"
tokio-stream = "0.1.9" #tokio = {version = "1.21.0", features = ["full"]}
toml = "0.5.10" #tokio-stream = "0.1.9"
tower-http = "0.3.5" #toml = "0.5.10"
uuid = "1.3.0" #tower-http = "0.3.5"
#uuid = "1.3.0"

View File

@@ -1,107 +0,0 @@
[[error]]
code = 11
category = "don't know yet"
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]]
code = 101
category = "key_missing::item0::num_of_items"
message = "num_of_items missing in item0"
[[error]]
code = 102
category = "key_missing::item0::next_free_id"
message = "next_free_id missing in item0"
[[error]]
message = "_commit missing in item0"
code = 103
category = "key_missing::item0::_commit"
[[error]]
message = "_commit missing"
code = 103
category = "key_missing::_commit"
[[error]]
code = 104
category = "data_storage::index_not_found"
message = "could not read index of item"
[[error]]
code = 105
category = "don't know yet"
message = "_commit not 8 bytes long"
[[error]]
code = 106
category = "parse_error"
message = "error while parsing to an int"
[[error]]
code = 107
category = "faulty_message"
message = "some error with message sent to the server"
[[error]]
code = 108
category = "faulty_message"
message = "There was not metadata in a message, that should have some."
[[error]]
code = 109
category = "faulty_message"
message = "something that should be utf8 valid was not"
[[error]]
code = 111
category = "faulty_update"
message = "mize-commit can't be updated manually"
[[error]]
code = 113
category = "don't know yet"
message = "module header not a String"
[[error]]
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"

View File

@@ -1,15 +0,0 @@
# The miZe Protocol
Simple Messages that are sent between client and server via WebSockets or unix sockets.
## A Message
## Commands
### From client to server
-
### From server to client

View File

@@ -1,254 +0,0 @@
use std::collections::HashMap;
use std::fmt::format;
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! {
pub static ref ERRORS: HashMap<u32, MizeError> = parse_errors();
}
fn parse_errors() -> HashMap<u32, MizeError>{
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: Vec::new(), 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;
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MizeError {
pub category: String,
pub code: u32,
pub message: String,
pub extra_msg: Vec<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 caller_location = std::panic::Location::caller();
if let Some(err_ref) = ERRORS.get(&code) {
let mut err = err_ref.clone();
err.code_location = Some(caller_location.into());
return err;
} else {
return MizeError {
category: "error with error system".to_string(),
code: 114,
message: format!("The error with code {} was not found in the Errors that where imported from the errors.toml file at build time.", code),
extra_msg: Vec::new(),
caused_by_msg: None,
code_location: Some(caller_location.into()),
};
}
}
pub fn extra_msg<E>(mut self, msg: E) -> MizeError
where E: std::fmt::Display
{
self.extra_msg.push(format!("{}", msg));
return self;
}
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 is_critical(self){
panic!("Aborting because of a critical Error")
}
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 {
return serde_json::json!({
"category": "error with error system",
"code": 114,
"message": "error converting an MizeError to String",
})
}
}
pub fn to_json_message(self) -> JsonMessage {
proto::JsonMessage::ErrMsg(
ErrMessage{
err: self,
}
)
}
}
pub trait MizeResultExtension<T> {
fn extra_msg<E>(self, msg: E) -> Result<T, MizeError>
where E: std::fmt::Display
;
fn handle(self) -> Result<T, MizeError>;
fn is_critical(self) -> T;
}
impl<T> MizeResultExtension<T> for Result<T, MizeError> {
fn extra_msg<E>(mut self, msg: E) -> Result<T, MizeError>
where E: std::fmt::Display
{
match self {
Err(ref mut e) => {
e.extra_msg.push(format!("{}", msg));
return self;
},
Ok(something) => Ok(something),
}
}
fn handle(self) -> Result<T, MizeError> {
match self {
Err(e) => {
Err(e.send().add_to_err_item().write_to_stderr())
},
Ok(something) => Ok(something)
}
}
fn is_critical(self) -> T {
self.expect("Aborting because of a critical Error")
}
}
impl From<FromUtf8Error> for MizeError {
#[track_caller]
fn from(utf_err: FromUtf8Error) -> MizeError{
let caller_location = std::panic::Location::caller();
let mut err = MizeError::new(109).extra_msg(format!("From std::string::FromUtf8Error: {}", utf_err));
err.code_location = Some(caller_location.into());
return err;
}
}
impl From<surrealdb::error::Db> for MizeError {
#[track_caller]
fn from(sur_err: surrealdb::error::Db) -> MizeError {
let caller_location = std::panic::Location::caller();
let mut err = MizeError::new(129).extra_msg(format!("From surrealdb Error: {}", sur_err));
err.code_location = Some(caller_location.into());
return err;
}
}
impl From<serde_json::Error> for MizeError {
#[track_caller]
fn from(serde_err: serde_json::Error) -> MizeError {
let caller_location = std::panic::Location::caller();
let mut err = MizeError::new(11).extra_msg(format!("From an serde_json::Error: {}", serde_err));
err.code_location = Some(caller_location.into());
return err;
}
}

View File

@@ -1,34 +0,0 @@
use std::convert::Infallible;
use std::net::SocketAddr;
use hyper::{Body, Request, Response, Server};
use hyper::service::{make_service_fn, service_fn};
use hyper::body::Bytes;
#[tokio::main]
async fn hyper_server() {
// Construct our SocketAddr to listen on...
let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
// And a MakeService to handle each connection...
let make_service = make_service_fn(|_conn| async {
Ok::<_, Infallible>(service_fn(handle))
});
// Then bind and serve...
let server = Server::bind(&addr).serve(make_service);
// And run forever...
if let Err(e) = server.await {
eprintln!("server error: {}", e);
}
}
async fn handle(req: Request<Body>) -> Result<Response<Body>, Infallible> {
let (mut sender, _body) = Body::channel();
sender.send_data(Bytes::from("sending from the sender\n"));
sender.abort();
println!("{:?}", req.body());
//println!("{:?}", Body::to_bytes());
Ok(Response::new(Body::from("hi")))
}

View File

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

View File

@@ -1,13 +0,0 @@
#![ allow( warnings ) ]
static PROTO_VERSION: u8 = 1;
pub mod server;
pub mod error;
pub mod integrated_types {
pub mod test_same;
}

View File

@@ -1,91 +0,0 @@
#![ allow( warnings ) ]
extern crate mize;
static HELP_MESSAGE: &str = "\
Usage:
mize-server [options] [command] [options for command]
Available Commands:
run starts the server
help prints this help message
version prints the version
import imports a file as a item of type file into the server
Available options
-h --help prints this help message
-v --version prints the version
-f --folder the mize-server-folder
";
//some flags e.g: "--file /tmp" can require that the next argument belongs to them instead of being
//the command
static FLAGS_WITH_ARGUMENTS: [&str; 2] = ["--folder", "-f"];
static AVAILABLE_COMMANDS: [&str; 6] = ["run", "help", "version", "import-js", "import-render", "import"];
static AVAILABLE_FLAGS: [&str; 4] = ["--version", "--help", "--folder", "-f"];
//static AVAILABLE_ONE_LETTER_FLAGS: [&str; 2] = ["v", "h"];
static VERSION_MESSAGE: &str = "\
Version: 0
";
fn main() {
/*
* WHAT IT DOES
* - get the arguments passed to the programm
*
* - sort out the flags for the main programm and the args for the comand
* - in "mize -h -u --lol server --file=/tmp --folder /home" the flags for the main programm would be ["-h","-u","--lol"]
* - and the args for the command would be ["server", "--file=/tmp --folder /home"]
*
* - get the command to "run"
* - in "mize -h -u --lol server --file=/tmp" it would be "server"
*
* - "run" the command and pass the options and subcommands to it
*
*/
//### get the arguments passed to the programm
let args: Vec<String> = std::env::args().collect();
//### sort out the flags for the main program and the args for the command
let mut main_flags:Vec<String> = Vec::new();
let mut command= String::from("");
let mut command_args = Vec::new();
for i in 1..args.len() {
if args[i].chars().next().unwrap() == '-' {
main_flags.push(args[i].clone());
// i != 0 bcs what if the programm name is in FLAGS_WITH_ARGUMENTS for some reason?
} else if i != 0 && FLAGS_WITH_ARGUMENTS.contains(&&args[i -1][..]) {
main_flags.push(args[i].clone());
//#### get the command to "run"
} else if AVAILABLE_COMMANDS.contains(&&args[i][..]){
command = args[i].clone();
for a in i..args.len(){
command_args.push(args[a].clone());
}
break;
} else {
println!("Command \"{}\" not found!!\n", args[i]);
println!("{}", HELP_MESSAGE);
std::process::exit(1);
}
}
//#### "run" the command and pass the options and subcommands to it
match &command[..] {
"run" => mize::server::run_server(args),
"import" => mize::server::server_utils::import(args),
"help" => println!("{}", HELP_MESSAGE),
"version" => println!("{}", VERSION_MESSAGE),
_ => {println!("Command not found!!\n"); println!("{}", HELP_MESSAGE);},
}
}

View File

@@ -1,656 +0,0 @@
pub mod server_utils;
pub mod itemstore;
pub mod proto;
//use futures_util::lock::Mutex;
use futures_util::{FutureExt, StreamExt, SinkExt};
use std::collections::HashMap;
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 crate::server::proto::MizeId;
use crate::server::proto::Update;
use crate::server::proto::update_thread;
use crate::error::MizeResultExtension;
use serde_json::Value as JsonValue;
use serde::{Serialize, Deserialize};
use uuid::Uuid;
use std::str::FromStr;
use std::borrow::Cow;
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 self::itemstore::Itemstore;
use axum_extra::routing::SpaRouter;
use axum::Router;
use axum::routing::{get, get_service};
use std::net::SocketAddr;
use axum::response::{IntoResponse, Html};
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade, CloseFrame};
use axum::extract::{State, self};
use axum::http::{StatusCode, Uri, Response, self};
use axum::http::header::{HeaderName, HeaderValue, HeaderMap};
use tower_http::services::{ServeDir, ServeFile};
static SERVER_PORT: u16 = 9432;
//static API_PATH_STRING: &str = "$api";
static SOCKET_CHANNEL_SIZE: usize = 200;
static UPDATE_CHANNEL_SIZE: usize = 200;
// max file sizez in the mize/data folder (in bytes)
static MAX_INDEX_FILE_SIZE: usize = 1_800_000;
static MAX_FIELDS_FILE_SIZE: usize = 2_600_000;
static MAX_KEYS_FILE_SIZE: usize = 30_000_000;
static MAX_DATA_FILE_SIZE: usize = 3_000_000_000;
//static HELP_MESSAGE: &str = "\
//Usage:
//mize-server server [options]
//
//Available options
//-h --help prints this help message
//-v --version prints the version
//--folder=<folder> mize-folder: where mize stores all it's stuff.
//";
//some flags e.g: "--file /tmp" can require that the next argument belongs to them instead of being
//the command
//static FLAGS_WITH_ARGUMENTS: [&str; 0] = [];
//static AVAILABLE_COMMANDS: [&str; 3] = ["run", "help", "version"];
//static AVAILABLE_FLAGS: [&str; 2] = ["--version", "--help"];
//static AVAILABLE_ONE_LETTER_FLAGS: [&str; 2] = ["v", "h"];
//static VERSION_MESSAGE: &str = "\
//Version: 0
//";
#[derive(Clone, Debug)]
pub enum Peer {
Client(Client), //the client id
Module(Module), //Module_name
Upstream(Upstream), //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 Upstream {
id: Uuid,
}
#[derive(Clone, Debug)]
pub struct Client {
// rx: UnboundedReceiverStream<Result<ws::Message, warp::Error>>,
tx: Sender<proto::MizeMessage>,
id: u64,
}
#[derive(Clone, Debug)]
pub struct Module {
tx: Sender<proto::MizeMessage>,
name: String,
client_id: u64,
kind: ModuleKind,
registered_types: Vec<String>,
}
#[derive(Clone, Debug)]
pub enum ModuleKind {
Rust(),
Python(),
JS(),
Lua(),
//connects to ws server
Extern(),
}
#[derive(Deserialize, Serialize, Debug)]
pub struct Render {
#[serde(rename="type")]
render_type: RenderType,
webroot: String,
name: String,
main: String,
folder: String,
}
#[derive(Deserialize, Serialize, Debug)]
pub enum RenderType {
#[serde(rename="json-ui")]
JsonUi,
#[serde(rename="webcomponent")]
WebComponent,
}
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....
// 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<MizeId, Vec<Peer>>>>,
renders: Arc<Mutex<Vec<Render>>>,
//maybe a list of upstream servers??
itemstore: Arc<Mutex<Itemstore>>,
mize_folder: String,
server_uuid: Uuid,
update_tx: mpsc::Sender<(Update, Peer)>,
}
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),
renders: Arc::clone(&mutexes.renders),
subs: Arc::clone(&mutexes.subs),
mize_folder: mutexes.mize_folder.clone(),
itemstore: Arc::clone(&mutexes.itemstore),
server_uuid: mutexes.server_uuid,
update_tx: mutexes.update_tx.clone(),
}
}
}
#[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() {
if args[i].contains('=') {
let split: Vec<_> = args[i].split('=').collect();
if split[0] == "--folder" {
mize_folder += split[1];
}
}
}
let (server_uuid, itemstore, renders) = init_server(mize_folder.clone()).await.extra_msg("Could not init server").handle().is_critical();
// the channel to send updates to, that are then handeled in another thread
let (update_tx, update_rx) = mpsc::channel::<(Update, Peer)>(UPDATE_CHANNEL_SIZE);
// 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())),
renders: Arc::new(Mutex::new(renders)),
itemstore: Arc::new(Mutex::new(itemstore)),
mize_folder: mize_folder.clone(),
server_uuid,
update_tx,
};
// the update handling task
let update_mutexes = mutexes.clone();
tokio::spawn(async move {
proto::update_thread(update_rx, update_mutexes).await.handle().extra_msg("update thread crashed").is_critical();
});
//listen on the local unix socket
//local_socket_server(mize_folder);
//run the webserver
axum_server(mize_folder, mutexes).await;
}
async fn axum_server(mize_folder: String, mutexes: Mutexes) {
/*
*
*
*/
//### Main Endpoints:
//## any number (with "-" in them) is interpreted as an item id
//## $api/socket/id
//## $api/rest/
let mutexes_clone = Mutexes::clone(&mutexes);
let serve_client = get_service(ServeDir::new("js-client/src")).handle_error(handle_error);
let mut app = Router::new()
.route("/api/socket", get(websocket_handler))
.route("/api/render/:id", get(get_render_main))
// .route("/$api/file", get(get_file_handler))
// .merge(SpaRouter::new("/$api/client", "js-client/src").index_file("main.html"))
.nest_service("/api/client", serve_client)
.with_state(mutexes.clone())
.fallback(render_item);
let renders = mutexes.renders.lock().await;
for render in &*renders {
match render.render_type {
RenderType::JsonUi => {},
RenderType::WebComponent => {
let url_path = String::from("/api/render/") + &render.name + "/webroot";
let file_path = mutexes.mize_folder.clone() + "/mr/" + &render.folder[..] + "/" + &render.webroot.clone()[..];
let serve_dir = get_service(ServeDir::new(&file_path)).handle_error(handle_error);
app = app.nest_service(&url_path, serve_dir);
},
}
}
drop(renders);
let addr = SocketAddr::from(([0, 0, 0, 0], SERVER_PORT));
// tracing::debug!("listening on {}", addr);
axum::Server::bind(&addr)
.serve(app.into_make_service())
.await
.unwrap();
}
async fn handle_error(_err: std::io::Error) -> impl IntoResponse {
(StatusCode::INTERNAL_SERVER_ERROR, "Something went wrong...")
}
async fn render_item(uri: Uri) -> impl IntoResponse {
Html(fs::read_to_string("js-client/src/main.html").unwrap())
}
//#[axum_macros::debug_handler]
async fn get_render_main(extract::Path(id): extract::Path<String>, State(mutexes): State<Mutexes>) -> http::Response<String> {
let renders = &*mutexes.renders.lock().await;
let render = renders.iter().filter(|&render| render.name == id).nth(0)
.unwrap_or(renders.iter().filter(|&render| render.name == "mize-mmejs").nth(0).expect("mize-mmejs"));
let file_path = mutexes.mize_folder.clone() + "/mr/" + &render.folder + "/" + &render.main;
match render.render_type {
RenderType::JsonUi => {
return Response::builder()
.header("content-type", "application/json")
.status(StatusCode::OK)
.body(fs::read_to_string(file_path).unwrap()).unwrap();
},
RenderType::WebComponent => {
return Response::builder()
.header("content-type", "application/javascript")
.status(StatusCode::OK)
.body(fs::read_to_string(file_path).unwrap()).unwrap();
},
}
}
async fn websocket_handler(
ws: WebSocketUpgrade,
State(mutexes): State<Mutexes>,
headers: HeaderMap,
) -> impl IntoResponse {
ws.on_upgrade(|socket| handle_websocket_connection(socket, mutexes, headers))
}
async fn handle_websocket_connection(
socket: WebSocket,
mutexes: Mutexes,
headers: HeaderMap,
){
let (mut socket_tx, mut socket_rx) = socket.split();
let (msg_tx, msg_rx): (Sender<proto::MizeMessage>, Receiver<proto::MizeMessage>) = mpsc::channel(SOCKET_CHANNEL_SIZE);
let mut msg_rx = ReceiverStream::new(msg_rx);
// check if mize-module header exists
let origin = if let Some(mod_header_val) = headers.get("mize-module"){
let mut mods = mutexes.modules.lock().await;
let mut client_id = mutexes.next_free_client_id.lock().await;
let mut module = Module{
tx: msg_tx.clone(),
client_id: *client_id,
name: String::from("tmp"),
kind: ModuleKind::Extern(),
registered_types: Vec::new(),
};
*client_id += 1;
if let Ok(name) = mod_header_val.to_str() {
module.name = name.to_string();
} else {
let err = MizeError::new(113)
.extra_msg("The mize-module Header could not be decoded into a String.");
msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
return;
};
println!("module {} connected", module.name);
mods.push(module.clone());
drop(mods);
Peer::Module(module)
} else {
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);
Peer::Client(client)
};
// Forwarding messages
let myorigin = origin.clone();
let mymutexes = mutexes.clone();
tokio::spawn(async move {
while let Some(msg) = msg_rx.next().await {
match msg {
proto::MizeMessage::Json(json_msg) => {
if let Ok(msg) = serde_json::to_string(&json_msg) {
let result = socket_tx.send(Message::Text(msg.clone())).await;
if let Err(e) = result {
println!("Client no longer Connected: {:?}", e);
remove_peer(myorigin.clone(), mymutexes.clone()).await;
} else {
println!("SENDING: {}", msg);
}
} 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;
}
proto::MizeMessage::Pong() => {
socket_tx.send(Message::Pong(Vec::new())).await;
}
proto::MizeMessage::Close((num, reason)) => {
println!("SENDING Close message");
socket_tx.send(Message::Close(Some(CloseFrame{code: num, reason: Cow::Owned(reason)}))).await;
}
}
}
});
// Reading messages
while let Some(result) = socket_rx.next().await {
let msg = if let Ok(msg) = result {msg} else {
let err = MizeError::new(115).extra_msg("TCP Connection Closed");
remove_peer(origin.clone(), mutexes.clone());
return;
};
match msg {
Message::Binary(b) => {
println!("Recieved a Binary Message. Those are not implemented yet.")
},
Message::Text(text) => {
println!("GOT: {}", text);
let json_msg: proto::JsonMessage = serde_json::from_str(&text).expect("error parsing json");
if let Err(err) = proto::handle_json_msg(json_msg, origin.clone(), mutexes.clone()).await {
let err_msg: MizeMessage = err.handle().into();
origin.send(err_msg).await;
};
},
Message::Close(inner) => {
match inner {
Some(CloseFrame{code, reason}) => {
origin.send(MizeMessage::Close((code, reason.into_owned()))).await;
},
None => {
origin.send(MizeMessage::Close((1000, "You didn't send a Close Reason".to_owned()))).await;
}
};
println!("Close from Client");
remove_peer(origin.clone(), mutexes.clone()).await;
}
Message::Ping(_) => {
msg_tx.send(MizeMessage::Pong());
},
Message::Pong(_) => {
let err = MizeError::new(11)
.extra_msg("unhandeld WebSocket-Message type: Pong")
.handle();
msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await;
},
};
};
}
fn load_mr(mize_folder: String) -> Result<Vec<Render>, MizeError> {
//load modules and renders
let mr_folders = 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());
let mut renders: Vec<Render> = Vec::new();
for mr_folder in mr_folders {
let mr_folder = match mr_folder {
Err(_) => {MizeError::new(11).extra_msg("Error unwrapping DirEntry").handle(); continue;},
Ok(idk) => idk,
};
let mr_string = match fs::read_to_string(format!("{}/mize.toml", mr_folder.path().display())) {
Err(_) => {MizeError::new(11).extra_msg("Error parsing mize.toml file to a toml string").handle(); continue;},
Ok(idk) => idk,
};
let mut mr_data = match mr_string.parse::<toml::Value>() {
Err(_) => {MizeError::new(11).extra_msg("Error parsing mize.toml string to toml::Value").handle(); continue;},
Ok(idk) => idk,
};
if let Some(local_renders) = mr_data.get_mut("render") {
if let Some(arr) = local_renders.as_array_mut(){
//add folder to local_renders
for mut el in arr.iter_mut() {
if let Some(table) = el.as_table_mut(){
table.insert("folder".to_owned(), toml::Value::String(mr_folder.file_name().into_string().unwrap()));
table.insert("type".to_owned(), toml::Value::String("json-ui".to_owned()));
table.insert("webroot".to_owned(), toml::Value::String("".to_owned()));
}
let render = match Render::deserialize(el.clone()) {
Ok(render) => render,
Err(err) => {MizeError::new(11).extra_msg(&format!("{}", err)).handle(); continue;}
};
renders.push(render);
}
}
}
if let Some(local_webcomponents) = mr_data.get_mut("webcomponent") {
if let Some(arr) = local_webcomponents.as_array_mut(){
//add folder to local_renders
for mut el in arr.iter_mut() {
if let Some(table) = el.as_table_mut(){
table.insert("folder".to_owned(), toml::Value::String(mr_folder.file_name().into_string().unwrap()));
table.insert("type".to_owned(), toml::Value::String("webcomponent".to_owned()));
}
let render = match Render::deserialize(el.clone()) {
Ok(render) => render,
Err(err) => {MizeError::new(11).extra_msg(&format!("{}", err)).handle(); continue;}
};
renders.push(render);
}
}
}
}
return Ok((renders));
}
async fn init_server(mize_folder: String) -> Result<(Uuid, Itemstore, Vec<Render>), MizeError> {
/*
* This function initializes the server.
* - create a mize folder if not there yet
* - get or generate the uuid for the server
* - load the renders and modules from the <mize-folder>/mr folder
* - create a surrealdb::Datastore inside the <mize-folder>/db folder
*
*/
let server_uuid;
//create mize.toml with id inside in folder if net yet done
match fs::read_to_string(format!("{}/mize.toml", mize_folder)) {
Err(err) => {
match err.kind() {
std::io::ErrorKind::NotFound => {
//create the file with new uuid
server_uuid = uuid::Uuid::new_v4();
let mut toml_val = toml::Value::Table(toml::map::Map::new());
toml_val.as_table_mut()
.ok_or(MizeError::new(11))?
.insert("uuid".to_owned(), toml::Value::String(server_uuid.to_string()));
fs::write(format!("{}/mize.toml", mize_folder), toml::to_string(&toml_val)
.map_err(|e| MizeError::new(11).extra_msg(e))?)
.map_err(|e| MizeError::new(11).extra_msg(e))?;
}
_ => {
server_uuid = uuid::Uuid::new_v4();
Err(err)
.map_err(|e| MizeError::new(11).extra_msg(e))
.extra_msg("Could not read <mize-folder>/mize.toml file")?;
}
}
},
Ok(toml_string) => {
let mut toml_val = toml_string.parse::<toml::Value>().map_err(|e| MizeError::new(11).extra_msg(e))?;
if let Some(uuid) = toml_val.get("uuid") {
server_uuid = Uuid::from_str(uuid.as_str().ok_or(MizeError::new(11))?)
.map_err(|e| MizeError::new(11).extra_msg(e))?;
} else {
//generate uuid and write that to toml file
server_uuid = uuid::Uuid::new_v4();
toml_val.as_table_mut().ok_or(MizeError::new(11))?.insert("uuid".to_owned(), toml::Value::String(server_uuid.to_string()));
fs::write(format!("{}/mize.toml", mize_folder), toml::to_string(&toml_val)
.map_err(|e| MizeError::new(11).extra_msg(e))?)
.map_err(|e| MizeError::new(11).extra_msg(e))?;
}
},
};
// create the itemstore
let itemstore = crate::server::itemstore::Itemstore::new(mize_folder.clone() + "/db").await.expect("error creating itemstore");
// load the modules and renders from <mize-folder>/mr
let renders = load_mr(mize_folder.clone())
.map_err(|e| MizeError::new(11))
.extra_msg("Error loading Modules and Renders")?;
return Ok((server_uuid, itemstore, renders));
}
pub async fn remove_peer(origin: Peer, mutexes: Mutexes) {
match origin {
Peer::Client(ref client) => {
//remove client from client list
let mut clients = mutexes.clients.lock().await;
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);
return;
},
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.....");
modules.remove(index);
println!("Close from Module");
return;
}
Peer::Upstream(_) => {
return;
}
}
}

View File

@@ -1,231 +0,0 @@
use std::net::ToSocketAddrs;
use std::{io, fmt::format, vec};
use surrealdb::kvs::{Datastore, Key, Val};
use itertools::Itertools;
use crate::error;
use crate::error::MizeError;
use crate::server::proto::{self};
use crate::server::Mutexes;
use crate::server::Peer;
use serde_json::Value as JsonValue;
use serde::{Serialize, Deserialize};
use super::proto::{Delta, MizeId};
// 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: Datastore,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Item {
#[serde(flatten)]
pub json: JsonValue,
}
pub struct Array {
}
//later
//pub struct Map {
//}
//pub struct Table {
//}
impl Item {
pub fn get_id(&self) -> Result<String, MizeError> {
if let JsonValue::Object(ob) = &self.json{
if let Some(id_val) = ob.get("__item__") {
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> {
//println!("DELTA: {:?}", delta);
for (path, value) in delta.delta {
let mut to_replace = &mut self.json;
let mut counter = path.len() as i32;
for name in path {
counter -= 1;
// if let Some(inner) = to_replace.get(&name) {
// to_replace = inner
// } else if let JsonValue::Object(ob) = to_replace {
//let mut new_map = JsonValue::Object(serde_json::Map::new());
//ob.insert(name, new_map);
//to_replace = &mut new_map;
// } else {
// return Err(MizeError::new(11)
// .extra_msg("Error applying a Delta to an Item. The path could not be followed."));
// }
//on last iteration
if counter == 0 {
if value.is_none() {
if to_replace.is_object() {
to_replace.as_object_mut().ok_or(MizeError::new(11))?.remove(&name);
return Ok(())
}
}
}
to_replace = if to_replace.get_mut(&name).is_some() {
to_replace.get_mut(&name).ok_or(MizeError::new(11))?
} else if to_replace.is_object() {
// when path does not exist
let mut new_map = JsonValue::Object(serde_json::Map::new());
to_replace.as_object_mut().ok_or(MizeError::new(11))?.insert(name.clone(), new_map);
to_replace.get_mut(&name).ok_or(MizeError::new(11))?
} else {
return Err(MizeError::new(11)
.extra_msg("Error applying a Delta to an Item. The path could not be followed."));
};
}
// replace with value from the delta
*to_replace = value.into();
}
//println!("After applying a Delta, {}", self.json);
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> {
//println!("PATH of DB: {}", path);
let ds = Datastore::new(&(String::from("file://") + &path[..])[..]).await
.map_err(|e| MizeError::new(30).extra_msg(format!("surrealdb Error: {}", e)).extra_msg(format!("Trying to create at Location: {}", path)))?;
let mut tr = ds.transaction(true, false).await?;
match tr.get(MizeId::zero().into_bytes()).await {
Ok(Some(val)) => println!("Item 0 already created"),
Ok(None) => {
let data = serde_json::json!({
"num_of_items": 1,
"next_free_id": 1,
"__commit__": 0,
"__type__": "mize-main",
"__item__": "0"
});
tr.set(MizeId::zero().into_bytes(), data.to_string()).await?;
tr.commit().await?;
},
Err(e) => {return Err(MizeError::from(e))},
};
return Ok(Itemstore {db:ds});
}
pub async fn create(&self, mut item: Item, mutexes: Mutexes, origin: Peer) -> Result<MizeId, MizeError> {
let mut tr = self.db.transaction(true, false).await?;
//get next_free_id
let json: JsonValue = serde_json::from_str(&String::from_utf8(tr.get(MizeId::zero().into_bytes()).await?.ok_or(MizeError::new(31))?)
.map_err(|_| MizeError::new(11))?)
.map_err(|_| MizeError::new(11))?;
let new_id_u64: u64 = json.get("next_free_id").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?;
let new_id = MizeId::new(format!("{}", new_id_u64));
let num_of_items: u64 = json.get("num_of_items").ok_or(MizeError::new(11))?.as_u64().ok_or(MizeError::new(11))?;
let new_next_id = new_id_u64 +1;
let new_num = num_of_items + 1;
//add mize-id field to item
//let delta: Delta = serde_json::from_str(&format!("[[[\"__item__\"], \"{}\"]]", new_id))?;
let mut delta = Delta::new();
delta.append(vec!["__item__"], serde_json::json!(new_id_u64));
item.apply_delta(delta)?;
tr.set(new_id.clone().into_bytes(), item.json.to_string()).await?;
tr.commit().await?;
//handle update to item0
let mut delta_item0: Delta = Delta::new();
delta_item0.append(vec!["num_of_items"], serde_json::json!(new_num));
delta_item0.append(vec!["next_free_id"], serde_json::json!(new_next_id));
proto::handle_update(vec![(proto::MizeId::new(format!("0")), delta_item0)], origin, mutexes.clone()).await?;
return Ok(new_id);
}
pub async fn update(&self, id: MizeId, delta: Delta) -> Result<(), MizeError>{
let mut tr = self.db.transaction(true, true).await?;
let item_bytes: Vec<u8> = tr.get(id.clone().into_bytes()).await?
.ok_or(MizeError::new(11)
.extra_msg("The Item an update should be applied to does not exist."))?;
let mut item = Item::from_bytes(item_bytes)?;
item.apply_delta(delta)?;
//println!("after applying delta: {:?}", item);
tr.set(id.into_bytes(), serde_json::to_string(&item)?).await?;
tr.commit().await?;
return Ok(());
}
pub async fn delete(&self, id: MizeId) -> Result<(), error::MizeError>{
let mut tr = self.db.transaction(true, true).await?;
tr.del(id.into_bytes()).await?;
tr.commit().await?;
return Ok(());
}
pub async fn get(&self, id: MizeId) -> Result<Item, MizeError> {
let mut tr = self.db.transaction(false, false).await?;
let item_res = tr.get(id.clone().into_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(32).format(vec![&format!("{:?}", id)]));
}
return Err(MizeError::new(11));
}
}

View File

@@ -1,470 +0,0 @@
use derive_more::From;
use futures_util::stream::select_all::Iter;
use serde::de::Error;
use serde_json::json;
use serde::de::Visitor;
use serde::de;
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::Peer;
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;
use crate::server::itemstore::Item;
use crate::server::proto;
use axum::extract::ws::CloseFrame;
use serde_json::Value as JsonValue;
use serde::{Serialize, Deserialize};
//###//===================================================
//all the structs and enums
#[derive(From, Debug, Clone)]
pub enum MizeMessage {
Json(JsonMessage),
Bin(BinMessage),
Pong(),
Close((u16, String)),
}
#[derive(Debug, Clone)]
pub struct BinMessage {
pub raw: Vec<u8>,
}
#[derive(From, Debug, Clone, Serialize, Deserialize)]
pub struct ErrMessage {
//pub cat: String,
pub err: MizeError,
}
//enum variant and a struct for every cmd a good idea???
#[derive(From, Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "cmd")]
pub enum JsonMessage {
#[serde(rename="err")]
ErrMsg(ErrMessage),
#[serde(rename="item.get")]
Get(GetItemMessage),
#[serde(rename="item.get-sub")]
GetSub(GetSubMessage),
#[serde(rename="item.give")]
Give(GiveItemMessage),
#[serde(rename="item.create")]
Create(CreateItemMessage),
#[serde(rename="item.created-id")]
CreatedId(CreatedIdMessage),
#[serde(rename="item.delete")]
Delete(DeleteItemMessage),
#[serde(rename="item.sub")]
Sub(SubItemMessage),
#[serde(rename="item.unsub")]
Unsub(UnsubItemMessage),
#[serde(rename="item.update-req")]
UpdateRequest(UpdateRequestMessage),
#[serde(rename="item.update")]
Update(UpdateMessage),
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GetItemMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GetSubMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GiveItemMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
item: Item,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CreateItemMessage {
//pub cat: String,
//cmd: String,
item: Item,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CreatedIdMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeleteItemMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubItemMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UnsubItemMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UpdateRequestMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
#[serde(flatten)]
delta: Delta,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UpdateMessage {
//pub cat: String,
//cmd: String,
#[serde(flatten)]
id: MizeId,
#[serde(flatten)]
delta: Delta,
}
#[serde(from = "SerdeDelta")]
#[serde(into = "SerdeDelta")]
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
pub struct Delta {
pub delta: Vec<(Path, Option<JsonValue>)>,
}
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
pub struct SerdeDelta {
pub delta: Vec<SerdeDeltaEntry>,
}
impl From<SerdeDelta> for Delta {
fn from(serde: SerdeDelta) -> Delta {
let mut delta = Delta {delta: Vec::new()};
for SerdeDeltaEntry(path, json) in serde.delta {
delta.delta.push((path, json));
}
return delta;
}
}
impl From<Delta> for SerdeDelta {
fn from(delta: Delta) -> SerdeDelta {
let mut serde = SerdeDelta {delta: Vec::new()};
for (path, json) in delta.delta {
serde.delta.push(SerdeDeltaEntry(path, json));
}
return serde;
}
}
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
pub struct SerdeDeltaEntry (
pub Path,
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
pub Option<JsonValue>,
);
pub type Path = Vec<String>;
pub type Update = Vec<(MizeId, Delta)>;
#[derive(Debug, Default, Clone, Serialize, Deserialize, Eq, Hash, PartialEq)]
pub struct MizeId {
#[serde(rename="id")]
main: String,
}
pub enum MizeIdKind<'a>{
Module{mod_name: &'a str, id: &'a str},
Upstream(&'a server::Peer),
Local(u64),
}
impl MizeId {
pub fn new(main: String) -> MizeId {
MizeId { main }
}
pub fn from_local(local: u64) -> MizeId {
MizeId::new(format!("{}", local))
}
pub fn into_bytes(self) -> Vec<u8> {
self.main.into_bytes()
}
pub fn zero() -> MizeId {
MizeId::new("0".to_owned())
}
pub fn to_string(self) -> String {
self.main
}
pub fn new_local(local_part: &str) -> MizeId {
MizeId { main: local_part.to_owned() }
}
pub fn kind(&self) -> Result<MizeIdKind, MizeError> {
let first_char = match self.main.chars().nth(0) {
Some(ch) => ch,
None => {
return Err(MizeError::new(11).extra_msg("Couldn't get the first char from a MizeId"));
},
};
if (first_char == '#'){
let iter = self.main.chars();
//scip the #
let mod_name_end = iter.enumerate().find(|&r| r.1 == '#' || r.1 == '/').ok_or(MizeError::new(11))?;
let mod_name = &self.main[0..mod_name_end.0];
let id = &self.main[mod_name_end.0..];
return Ok(MizeIdKind::Module{ mod_name, id});
} else {
let id: u64 = match self.main.parse(){
Ok(num) => num,
Err(_) => {
return Err(MizeError::new(11).extra_msg("Couldn't parse a local id into an integer"))
},
};
return Ok(MizeIdKind::Local(id));
}
}
}
impl Delta {
pub fn new() -> Delta{
Delta { delta: Vec::new() }
}
pub fn append(&mut self, path: Vec<&str>, json: JsonValue){
let new_path: Vec<String> = path.into_iter().map(|e| e.to_owned()).collect();
self.delta.push((new_path, Some(json)));
}
}
//###//===================================================
//From trait implementations, to be able to treat everything as a MizeMessage that can be
impl From<GiveItemMessage> for MizeMessage {
fn from(give: GiveItemMessage) -> MizeMessage {
let two: JsonMessage = give.into();
return two.into();
}
}
impl From<MizeError> for MizeMessage {
fn from(err: MizeError) -> MizeMessage {
let hi: ErrMessage = err.into();
let two: JsonMessage = hi.into();
return two.into();
}
}
impl From<CreatedIdMessage> for MizeMessage {
fn from(give: CreatedIdMessage) -> MizeMessage {
let two: JsonMessage = give.into();
return two.into();
}
}
impl From<UpdateMessage> for MizeMessage {
fn from(give: UpdateMessage) -> MizeMessage {
let two: JsonMessage = give.into();
return two.into();
}
}
//###//===================================================
// functions
pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) -> Result<(), MizeError>{
match msg {
JsonMessage::Get(msg) => {
match msg.id.kind()? {
//MizeId::Module { mod_name, id } => {
//TODO: get type of module with mod_name .... and then either forward msg or call
//module_get_code
//},
MizeIdKind::Local(id) => {
let itemstore = mutexes.itemstore.lock().await;
let response = GiveItemMessage {
id: msg.id,
item: itemstore.get(MizeId::from_local(id)).await?,
};
origin.send(response).await;
},
_ => {return Err(MizeError::new(11).extra_msg("id type is not implemented"));},
};
return Ok(());
},
JsonMessage::GetSub(msg) => {
//sub to item
let mut subs = mutexes.subs.lock().await;
if let Some(mut vec_ids) = subs.get_mut(&msg.id) {
vec_ids.push(origin.clone());
} else {
subs.insert(msg.id.clone(), vec![origin.clone()]);
};
match msg.id.kind()? {
//MizeId::Module { mod_name, id } => {
//TODO: get type of module with mod_name .... and then either forward msg or call
//module_get_code
//},
MizeIdKind::Local(id) => {
let itemstore = mutexes.itemstore.lock().await;
let response = GiveItemMessage {
id: msg.id,
item: itemstore.get(MizeId::from_local(id)).await?,
};
origin.send(response).await;
},
_ => {return Err(MizeError::new(11).extra_msg("id types not implemented"));},
}
return Ok(());
}
JsonMessage::UpdateRequest(mut msg) => {
let update: Update = vec![(msg.id, msg.delta)];
handle_update(update, origin, mutexes.clone()).await?;
// let subs = mutexes.subs.lock().await;
// for (id, delta) in new_update {
//send that delta to every origin that is subbed the item
// let peers = match subs.get(&id) {
// Some(vec) => vec,
// None => continue,
// };
// let message = UpdateMessage{id, delta};
// for peer in peers {
//TODO: send should only take a reference
// peer.send(message.clone()).await
// }
// }
},
JsonMessage::Create(msg) => {
let itemstore = mutexes.itemstore.lock().await;
let id = itemstore.create(msg.item, mutexes.clone(), origin.clone()).await?;
let response = CreatedIdMessage {id};
origin.send(response).await;
return Ok(());
},
JsonMessage::Delete(msg) => {
match msg.id.kind()? {
MizeIdKind::Local(id) => {
let itemstore = mutexes.itemstore.lock().await;
itemstore.delete(MizeId::from_local(id)).await?;
return Ok(());
},
_ => {return Err(MizeError::new(11).extra_msg("deleteng non Local items not implemented yet"));}
}
},
_ => {
return Err(MizeError::new(11).extra_msg("ItemMessage cmd not handeld"));
},
};
return Ok(())
}
pub async fn handle_update(update: Update, origin: Peer, mutexes: Mutexes) -> Result<(), MizeError> {
mutexes.update_tx.send((update, origin)).await.map_err(|err| MizeError::new(11).extra_msg(format!("From SendError: {}", err)));
Ok(())
}
pub async fn update_thread(mut update_rx: mpsc::Receiver<(Update, Peer)>, mutexes: Mutexes) -> Result<(), MizeError> {
while let Some((update, origin)) = update_rx.recv().await {
println!("UPDATE on id: {:?}", update.iter().next().unwrap().0.main);
//TODO: run update code from modules and types
let itemstore = mutexes.itemstore.lock().await;
let subs = mutexes.subs.lock().await;
for (id, delta) in update {
//update the items in the itemstore
if let MizeIdKind::Local(id) = id.kind()? {
itemstore.update(MizeId::from_local(id), delta.clone()).await?;
} else {
return Err(MizeError::new(11).extra_msg("updates to non local items are not handeld yet"));
}
//send a update msg to subbed clients
let peers = match subs.get(&id) {
Some(hi) => hi,
None => break,
};
let msg = UpdateMessage{id, delta};
for peer in peers {
peer.send(msg.clone()).await;
}
}
}
Ok(())
}

View File

@@ -1,28 +0,0 @@
use std::io::{self, ErrorKind, SeekFrom};
use std::collections::HashMap;
use std::io::prelude::*;
use std::fs;
use std::fs::File;
use std::path::Path;
use crate::error::MizeError;
pub struct Fields{
key: u64,
data: u64
}
pub struct Index{
commit_number: u64,
//fields: Vec<Fields>
}
pub fn import(args: Vec<String>){
}
pub fn update_item(id: u64, key: String, update_string: String){
}