large rewrite to make mize run on every platform

src/platform/os/cli/call.rs

@@ -0,0 +1,4 @@
+use clap::ArgMatches;
+
+pub fn call(sub_matches: &ArgMatches) {
+}

src/platform/os/cli/daemon.rs

@@ -0,0 +1,66 @@
+use std::path::PathBuf;
+
+use clap::ArgMatches;
+use home::home_dir;
+use std::fs::File;
+use log::{trace, debug, info, warn, error};
+use nix::sys::signal::{self, Signal};
+
+use mize::error::{MizeResultTrait, IntoMizeResult, MizeError};
+use mize::instance::Instance;
+
+
+pub fn daemon(sub_matches: &ArgMatches) {
+    println!("not implemented");
+    /*
+    let instance_path = match sub_matches.get_one::<String>("store") {
+        Some(a) => PathBuf::from(a),
+        None => {
+            let mut home_dir = home_dir()
+            .ok_or(MizeError::new().category("io").category("env").msg("could not get the home directory, where the mize store is by default")).critical();
+            home_dir.push(".mize");
+            home_dir
+        },
+    };
+
+    let instance = Instance::new(instance_path.clone()).await.critical();
+
+    info!("Opened Instance at \"{}\"", instance_path.clone().display());
+
+    match sub_matches.subcommand() {
+        // stop the daemon
+        Some(("stop", _)) => {
+            if instance.daemon_is_running().await.critical() {
+                //instance.stop_daemon();
+                println!("Todo")
+            } else {
+                warn!("Daemon is already not running.");
+            };
+        },
+        Some(("status", _)) => {
+            if instance.daemon_is_running().await.critical() {
+                eprintln!("Daemon is running");
+            } else {
+                eprintln!("Daemon is not running");
+            };
+        },
+
+        Some(("kill", _)) => {
+            signal::kill(instance.get_daemon_pid().await.critical(), Signal::SIGTERM).mize_result_msg("Could not kill daemon").critical();
+        },
+
+        Some((cmd, _)) => {
+            error!("Unknown Command: {}", cmd);
+        }
+
+        // just start the daemon
+        None => {
+            if !instance.daemon_is_running().await.critical() {
+                debug!("Daemon is not running. starting...");
+                instance.start_deamon().await.critical();
+                info!("Daemon Started");
+            };
+        },
+    }
+    */
+}

src/platform/os/cli/get.rs

@@ -0,0 +1,24 @@
+use std::path::PathBuf;
+
+use clap::ArgMatches;
+use home::home_dir;
+
+use mize::error::{MizeResultTrait, IntoMizeResult, MizeError};
+use mize::instance::Instance;
+
+
+
+pub fn get(sub_matches: &ArgMatches) {
+    let instance_folder_path = match sub_matches.get_one::<String>("store") {
+        Some(a) => PathBuf::from(a),
+        None => {
+            let mut home_dir = home_dir()
+            .ok_or(MizeError::new().category("io").category("env").msg("could not get the home directory, where the mize store is by default")).critical();
+            home_dir.push(".mize");
+            home_dir
+        },
+    };
+    println!("unfinished");
+    //let instance = Instance::new(instance_folder_path);
+}
+

src/platform/os/cli/mount.rs

@@ -0,0 +1,4 @@
+use clap::ArgMatches;
+
+pub fn mount(sub_matches: &ArgMatches) {
+}

src/platform/os/cli/serve.rs

@@ -0,0 +1,4 @@
+use clap::ArgMatches;
+
+pub fn serve(sub_matches: &ArgMatches) {
+}

src/platform/os/cli/set.rs

@@ -0,0 +1,4 @@
+use clap::ArgMatches;
+
+pub fn set(sub_matches: &ArgMatches) {
+}

src/platform/os/cli/show.rs

@@ -0,0 +1,4 @@
+use clap::ArgMatches;
+
+pub fn show(sub_matches: &ArgMatches) {
+}

src/platform/os/main.rs

@@ -0,0 +1,337 @@
+
+#![ allow( warnings ) ]
+
+use std::path::PathBuf;
+use clap::{ArgAction, ArgMatches};
+use clap::{Arg, crate_version, Command};
+
+use tokio::sync::{Mutex, mpsc};
+use std::sync::Arc;
+use log::Level;
+use log::LevelFilter;
+use env_logger::{Builder, Env};
+use std::io::Write;
+use colored::Colorize;
+use std::env;
+
+use log::{trace, debug, info, warn, error};
+
+mod cli {
+    pub mod daemon;
+    pub mod serve;
+    pub mod get;
+    pub mod set;
+    pub mod show;
+    pub mod mount;
+    pub mod call;
+
+    pub use daemon::daemon;
+    pub use serve::serve;
+    pub use get::get;
+    pub use set::set;
+    pub use show::show;
+    pub use mount::mount;
+    pub use call::call;
+}
+
+static APPNAME: &str = "mize";
+static DEFAULT_LOG_LEVEL: LevelFilter = LevelFilter::Warn;
+
+fn main() {
+    // welcome to the mize source code
+    // this is the main entry point for platforms with an os (eg: Linux, MacOS, Windows, BSD, ...)
+    // Also let's try to make the most best documented code ever. (eventually) so it's fun to work on
+
+    let cli_matches = cli_matches();
+
+    init_logger(&cli_matches);
+
+    // log a message of every level, to test loging
+    // test_logger();
+
+
+    // match command
+    match cli_matches.subcommand() {
+        // mi daemon
+        Some(("daemon", sub_matches)) => cli::daemon(sub_matches),
+
+        // mi serve
+        Some(("serve", sub_matches)) => cli::serve(sub_matches),
+
+        // mi mount
+        Some(("mount", sub_matches)) => cli::mount(sub_matches),
+
+        // mi get
+        Some(("get", sub_matches)) => cli::get(sub_matches),
+
+        // mi set
+        Some(("set", sub_matches)) => cli::set(sub_matches),
+
+        // mi show
+        Some(("show", sub_matches)) => cli::show(sub_matches),
+
+        // mi call
+        Some(("call", sub_matches)) => cli::call(sub_matches),
+
+        // some unknown command passed
+        Some((cmd, sub_matches)) => error!("The subcommand: {} is not known. use --help to list availavle commands", cmd),
+
+        None => error!("No subcommand was passed. use --help to list availavle comamnds."),
+    }
+
+
+}
+
+fn init_logger(cli_matches: &ArgMatches) {
+
+    // to save log messages, before the logger is setup
+    let mut log_messages: Vec<(Level, String)> = Vec::new();
+
+    log_messages.push((Level::Info, "Starting Mize".to_owned()));
+
+    // builder with DEFAULT_LOG_LEVEL
+    let mut builder = Builder::new();
+    builder.format_indent(Some(8));
+
+
+
+    // set loglevel
+    let mut log_level = DEFAULT_LOG_LEVEL;
+    log_messages.push((Level::Trace, format!("set log level from DEFAULT_LOG_LEVEL to {}", DEFAULT_LOG_LEVEL)));
+
+    // check env RUST_LOG
+    let mut val_rust_log = level_from_env("RUST_LOG", log_level);
+    if let Some(level_rust_log) = val_rust_log.0 {
+        log_level = level_rust_log;
+        log_messages.push((Level::Trace, format!("set log level from the Variable RUST_LOG to {}", level_rust_log)));
+    }
+    log_messages.append(&mut val_rust_log.1);
+
+    // check env MIZE_LOG
+    let mut val_mize_log = level_from_env("MIZE_LOG", log_level);
+    if let Some(level_mize_log) = val_mize_log.0 {
+        log_level = level_mize_log;
+        log_messages.push((Level::Trace, format!("set log level from the Variable MIZE_LOG to {}", level_mize_log)));
+    }
+    log_messages.append(&mut val_mize_log.1);
+
+    // check -vvvvv flags
+    let v_flag_count = cli_matches.get_count("verbose");
+    let default_log_level_as_num = match DEFAULT_LOG_LEVEL {
+        LevelFilter::Off => 0,
+        LevelFilter::Error => 1,
+        LevelFilter::Warn => 2,
+        LevelFilter::Info => 3,
+        LevelFilter::Debug => 4,
+        LevelFilter::Trace => 5,
+    };
+    let new_log_level_num = default_log_level_as_num + v_flag_count;
+    log_level = match new_log_level_num {
+        0 => LevelFilter::Off,
+        1 => LevelFilter::Error,
+        2 => LevelFilter::Warn,
+        3 => LevelFilter::Info,
+        4 => LevelFilter::Debug,
+        5 => LevelFilter::Trace,
+        _ => {
+            log_messages.push((Level::Trace, format!("the number of verbose (-v) flags passed ({}) exeeds the maximumg log_level", v_flag_count)));
+            LevelFilter::Trace
+        },
+    };
+    log_messages.push((Level::Trace, format!("set log level from the verbose flags to {}", log_level)));
+ 
+    // check --log-level arg
+    if let Some(log_level_arg_string) = cli_matches.get_one::<&str>("log-level") {
+        let log_level_from_arg = match log_level_arg_string.to_lowercase().as_str() {
+            "none" => LevelFilter::Off,
+            "trace" => LevelFilter::Trace,
+            "debug" => LevelFilter::Debug,
+            "info" => LevelFilter::Info,
+            "warn" => LevelFilter::Warn,
+            "error" => LevelFilter::Error,
+            _ => {
+                log_messages.push((Level::Warn, format!(
+                        "The argument to --log-level \"{}\" is not a valid log-level. loglevel stays at: {}"
+                        , log_level_arg_string
+                        , log_level
+                    )));
+                log_level
+            }
+        };
+        log_level = log_level_from_arg;
+        log_messages.push((Level::Trace, format!("set log level from the --log-level arg to {}", log_level_from_arg)));
+    }
+
+    // check --silent flag
+    if cli_matches.get_flag("silent") {
+        if log_level != DEFAULT_LOG_LEVEL {
+            log_messages.push((Level::Warn, format!(
+                    "The loglevel was set by some method (it's not the DEFAULT_LOG_LEVEL) and is now overwritten by the --siltne to be OFF"
+                )))
+        }
+        log_level = LevelFilter::Off;
+    }
+
+    // actually set the log_level on the builder
+    builder.filter_level(log_level);
+    log_messages.push((Level::Info, format!("loglevel is now {}", log_level)));
+
+
+    // set format
+    builder.format(|buf, record| {
+            match record.level() {
+                Level::Trace => writeln!(buf, "[ {} ] {}", format!("{}", record.level()).yellow(), record.args()),
+                Level::Debug => writeln!(buf, "[ {} ] {}", format!("{}", record.level()).green(), record.args()),
+                Level::Info => writeln!(buf, "[ {}  ] {}", format!("{}", record.level()).blue(), record.args()),
+                Level::Warn => writeln!(buf, "[ {}  ] {}", format!("{}", record.level()).truecolor(245, 164, 66), record.args()),
+                Level::Error => writeln!(buf, "[ {} ] {}", format!("{}", record.level()).red(), record.args()),
+            }
+        });
+
+    builder.init();
+
+
+    // write out all log messages from log_messages
+    for (level, msg) in log_messages {
+        match level {
+            Level::Error => error!("{}", msg),
+            Level::Warn => warn!("{}", msg),
+            Level::Info => info!("{}", msg),
+            Level::Debug => debug!("{}", msg),
+            Level::Trace => trace!("{}", msg),
+        }
+    }
+}
+
+fn test_logger() {
+    error!("hi from error");
+    warn!("hi from warn");
+    info!("hi from info");
+    debug!("hi from debug");
+    trace!("hi from trace");
+}
+
+fn level_from_env(var_name: &str, log_level: LevelFilter) -> (Option<log::LevelFilter>, Vec<(Level, String)>) {
+    let mut log_messages: Vec<(Level, String)> = Vec::new();
+    let var_value = env::var(var_name);
+    match var_value {
+        Ok(val) => {
+            // set loglevel acording to RUST_LOG
+            let level = match val.to_lowercase().as_str() {
+                "none" => LevelFilter::Off,
+                "trace" => LevelFilter::Trace,
+                "debug" => LevelFilter::Debug,
+                "info" => LevelFilter::Info,
+                "warn" => LevelFilter::Warn,
+                "error" => LevelFilter::Error,
+                _ => {
+                    log_messages.push((Level::Warn, format!(
+                            "The value of Environment Variable {}: \"{}\" is not a valid log-level. loglevel stays at: {}"
+                            , var_name
+                            , val
+                            , log_level
+                        )));
+                    log_level
+                }
+            };
+            return (Some(level), log_messages)
+        },
+        Err(env::VarError::NotPresent) => {},
+        Err(env::VarError::NotUnicode(os_string)) => {
+
+            let os_string_as_hex : String = os_string.clone().into_encoded_bytes().iter()
+              .map(|b| format!("{:02x}", b).to_string())
+              .collect::<Vec<String>>()
+              .join("");
+            log_messages.push((Level::Debug, format!(
+                "value of Environment Variable {} was not a unicode valid string. Therefore ignoring it.
+                \n value with replacement char U+FFFD: {}
+                \n the WTF-8 raw bytes as hex: {}"
+                , var_name
+                , os_string.to_string_lossy(), os_string_as_hex
+            )));
+        },
+    };
+    return (None, log_messages);
+}
+
+fn cli_matches() -> clap::ArgMatches {
+
+    let store_arg = Arg::new("store")
+        .short('s')
+        .long("store")
+        .help("The folder used as the Itemstore");
+
+    let main = Command::new(APPNAME)
+        .version(crate_version!())
+        .author("Sebastian Moser")
+        .about("The MiZe Command line tool")
+        .arg(Arg::new("verbose")
+            .long("verbose")
+            .short('v')
+            .action(ArgAction::Count)
+        )
+        .arg(Arg::new("log-level")
+            .long("log-level")
+            .value_name("LOGLEVEL")
+            .help("set the log-level to one of OFF, ERROR, WARN, INFO, DEBUG, TRACE")
+        )
+        .arg(Arg::new("silent")
+            .long("silent")
+            .action(ArgAction::SetTrue)
+            .help("set the log-level to OFF")
+        )
+        .subcommand(
+                Command::new("daemon")
+                .aliases(["da"])
+                .arg(&store_arg)
+                .about("Starts the mize daemon. Has subcommands to stop, etc")
+                .subcommand(Command::new("stop")
+                        .about("Stops the mize daemon if one is running")
+                    )
+                .subcommand(Command::new("kill"))
+                        .about("Stops the mize daemon with a signal")
+                        .alias("k")
+                .subcommand(Command::new("status")
+                        .alias("st")
+                    )
+            )
+        .subcommand(
+                Command::new("serve")
+                .aliases(["se"])
+                .arg(&store_arg)
+            )
+        .subcommand(
+                Command::new("mount")
+                .aliases(["m"])
+                .arg(&store_arg)
+            )
+        .subcommand(
+                Command::new("get")
+                .aliases(["g"])
+                .arg(&store_arg)
+                .arg(Arg::new("id").help("The id or path to get"))
+            )
+        .subcommand(
+                Command::new("set")
+                .aliases(["s"])
+                .arg(&store_arg)
+                .arg(Arg::new("id").help("The id or path to set"))
+            )
+        .subcommand(
+                Command::new("show")
+                .aliases(["so"])
+                .arg(&store_arg)
+                .arg(Arg::new("id").help("The id or path to sub to and show"))
+            )
+        .subcommand(
+                Command::new("call")
+                .aliases(["c"])
+                .arg(&store_arg)
+            )
+        .arg_required_else_help(true);
+
+    return main.get_matches();
+}
+
+

src/platform/os/unix-socket.rs

@@ -0,0 +1,61 @@
+
+impl Connection {
+    pub fn from_unix(unix_sock: UnixStream) -> Connection {
+
+    let (recieve_tx, recieve_rx) = tokio::sync::mpsc::channel(crate::instance::UNIX_SOCKET_CHANNEL_SIZE);
+    let (send_tx, send_rx) = tokio::sync::mpsc::channel(crate::instance::UNIX_SOCKET_CHANNEL_SIZE);
+    let (unix_read, unix_write) = unix_sock.into_split();
+
+    // TODO: error handling and close con on error
+    tokio::spawn(unix_send(unix_write, send_rx));
+    tokio::spawn(unix_recieve(unix_read, recieve_tx));
+
+    return Connection { tx: send_tx, rx: recieve_rx }
+
+    }
+}
+
+async fn unix_send(unix_write: OwnedWriteHalf, send_rx: Receiver<MizeMessage>) -> MizeResult<()> {
+    loop {
+        let ready = unix_write.ready(Interest::WRITABLE).await.mize_result_msg("hi")?;
+        if ready.is_writable() {
+            // Try to write data, this may still fail with `WouldBlock`
+            // if the readiness event is a false positive.
+            match unix_write.try_write(b"hello world") {
+                Ok(n) => {
+                    println!("wrote {} bytes to unix sock", n);
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                    continue;
+                }
+                Err(e) => {
+                    return Err(e.into());
+                }
+            }
+        }
+    }
+}
+
+async fn unix_recieve(unix_read: OwnedReadHalf, recieve_tx: Sender<MizeMessage>) -> MizeResult<()> {
+    loop {
+        let ready = unix_read.ready(Interest::READABLE).await.mize_result_msg("hi")?;
+
+        if ready.is_readable() {
+            let mut data = vec![0; 1024];
+            // Try to read data, this may still fail with `WouldBlock`
+            // if the readiness event is a false positive.
+            match unix_read.try_read(&mut data) {
+                Ok(n) => {
+                    println!("read {} bytes from unix sock", n);        
+                }
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                    continue;
+                }
+                Err(e) => {
+                    return Err(e.into());
+                }
+            }
+
+        }
+    }
+}

src/platform/os/web.rs

@@ -0,0 +1,239 @@
+
+static SERVER_PORT: u16 = 3000;
+static API_ENDPOINT: &str = "@mize";
+
+
+pub fn run() -> Result<Web, MizeError> {
+    /*
+     *  - Run a Webserver
+     *
+     *
+     *
+     */
+
+ 
+    // I will deal with renders later. cli only for now
+    // let renders = 
+
+//    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);
+//            },
+//        }
+//    }
+
+
+    // let serve_client = get_service(ServeDir::new("js-client/src")).handle_error(handle_error);
+
+    let mut app = Router::new()
+        .route(format!("/{}/socket", API_ENDPOINT), get(websocket_handler))
+//        .route(format!("/{}/render/:id", API_ENDPOINT), get(get_render_main))
+//        .route(format!("/{}/file", API_ENDPOINT), get(get_file_handler))
+//        .merge(SpaRouter::new(format!("/{}/client", API_ENDPOINT), "js-client/src").index_file("main.html"))
+//        .nest_service(format!("/{}/client", API_ENDPOINT), serve_client)
+        .with_state(main)
+        .fallback(render_item);
+
+    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
+        .expect(format!("Unable to bind axum::Server to {}", addr));
+}
+
+pub struct Web {
+}
+
+impl Web {
+
+}
+
+
+async fn handle_error(_err: std::io::Error) -> impl IntoResponse {
+        MizeError::new(0).msg("internal server error").handle()
+        (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(main): State<&Main>,
+        headers: HeaderMap,
+) -> impl IntoResponse {
+    ws.on_upgrade(|socket| handle_websocket_connection(socket, main, headers))
+}
+
+async fn handle_websocket_connection(
+    socket: WebSocket,
+    main: &Main,
+    headers: HeaderMap,
+){
+    //TODO: auth
+
+    //socket_tx.send() ... send out the socket
+    //socket_rx.read() ... read a message from the socket
+    let (mut socket_tx, mut socket_rx) = socket.split();
+
+    // msg_tx.send() ... send a MizeMessage to a Peer. should be in a Peer struct
+    // msg_rx.read() ... read MizeMessages that should be sent to the Peer that this ws connection is to
+    let (msg_tx, msg_rx): (Sender<proto::MizeMessage>, Receiver<proto::MizeMessage>) = mpsc::channel(SOCKET_CHANNEL_SIZE);
+
+    // why is this actually
+    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 {
+//    };
+
+    // Forwarding messages
+    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;
+            },
+
+        };
+    };
+}