diff --git a/core/src/c_api.rs b/core/src/c_api.rs index e69de29..e18f65a 100644 --- a/core/src/c_api.rs +++ b/core/src/c_api.rs @@ -0,0 +1,3 @@ + + +//pub extern "C" fn diff --git a/core/src/lib.rs b/core/src/lib.rs index 74372d7..8d2548b 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -55,6 +55,46 @@ pub use instance::{ pub use mize_macros::*; use std::path::PathBuf; +use std::sync::{Arc, Mutex, OnceLock}; + +use crate::error::MizeResult; +use crate::instance::Mize; +use crate::mize_err; + +/// Global Mize instance storage. +/// This is set once when the Mize instance is created and can be accessed from anywhere. +pub static GLOBAL_MIZE: OnceLock>> = OnceLock::new(); + +/// Get the global Mize instance. +/// Returns a clone of the Mize instance (which is cheap as it only clones Arcs internally). +pub fn get_global_mize() -> MizeResult { + let arc = GLOBAL_MIZE + .get() + .ok_or_else(|| mize_err!("No global Mize instance set"))?; + let mize = arc + .lock() + .map_err(|_| mize_err!("Failed to lock global Mize"))?; + Ok(mize.clone()) +} + +/// Set the global Mize instance. +pub fn set_global_mize(mize: Mize) -> MizeResult<()> { + let arc = Arc::new(Mutex::new(mize)); + if let Errr(_) = GLOBAL_MIZE.set(arc) { + let a = GLOBAL_MIZE.get().unwrap(); + let mut b = a.lock(); + *b = mize; + } + Ok(()) +} + +/// C API function to set the global Mize instance from a dynamically loaded part. +/// This is a hack to allow Rust parts loaded as .so files to access the global Mize instance. +#[unsafe(no_mangle)] +pub extern "C" fn mize_rust_hack_set_global(mize: Arc>) { + let _ = GLOBAL_MIZE.set(mize); +} + // platform specific stuff pub mod platform { use crate::{Mize, MizeResult}; diff --git a/core/src/platform/js/deno.json b/core/src/platform/js/deno.json new file mode 100644 index 0000000..5b320c2 --- /dev/null +++ b/core/src/platform/js/deno.json @@ -0,0 +1,8 @@ +{ + "tasks": { + "dev": "deno run --watch main.ts" + }, + "imports": { + "@std/assert": "jsr:@std/assert@1" + } +} diff --git a/core/src/platform/js/init.js b/core/src/platform/js/init.js new file mode 100644 index 0000000..1726094 --- /dev/null +++ b/core/src/platform/js/init.js @@ -0,0 +1,30 @@ + + +export async function init_mize(config) { + + if ("mize" in window) { + window.mize.set("self/config", config) + return window.mize + } + + let config_str = ""; + if (typeof config == "string") { + config_str = config + + } else if (typeof config == "object") { + config_str = JSON.stringify(config) + + } else { + throw "Error: config passed to init_mize() is not of type 'object' or 'string'" + } + + const { new_js_instance } = wasm_bindgen; + + await wasm_bindgen(config.module_dir.mize + "/mize_bg.wasm") + + window.mize = await new_js_instance(config_str) + window.mize.mod = {} + window.mize.init() + + return window.mize +} diff --git a/core/src/platform/js/mod.rs b/core/src/platform/js/mod.rs new file mode 100644 index 0000000..5bcddd6 --- /dev/null +++ b/core/src/platform/js/mod.rs @@ -0,0 +1,328 @@ +use crate::id::MizeId; +use crate::item::{Item, ItemData}; +use crate::platform::js::js_sys::Function; +use core::convert::From; +use core::option::Option; +use std::panic; +use std::ptr::NonNull; +use web_sys::js_sys::{self, eval, Promise}; +use web_sys::Worker; +use web_sys::{WorkerOptions, WorkerType}; + +use crate::core::item::IntoItemData; +use crate::error::MizeResult; +use crate::Mize; +use crate::MizeError; +use crate::mize_err; + +// console_log macro +// that can be copied into other files for debugging purposes +#[cfg(feature = "target-js")] +use wasm_bindgen::prelude::*; + +#[cfg(feature = "target-js")] +#[wasm_bindgen] +extern "C" { + #[wasm_bindgen(js_namespace = console)] + fn log(s: &str); +} + +#[cfg(feature = "target-js")] +macro_rules! console_log { + // Note that this is using the `log` function imported above during + // `bare_bones` + ($($t:tt)*) => (unsafe { log(&format_args!($($t)*).to_string())}) +} + +#[cfg(not(feature = "target-js"))] +macro_rules! console_log { + // Note that this is using the `log` function imported above during + // `bare_bones` + ($($t:tt)*) => {}; +} +//end of console_log macro + +#[global_allocator] +static ALLOC: wee_alloc::WeeAlloc = wee_alloc::WeeAlloc::INIT; + +pub fn wasm_instance_init(mize: &mut Mize) -> MizeResult<()> { + console_log!("Hello world from wasm_instance_init!!!!!!!!!!"); + + #[cfg(feature = "target-obsidian")] { + ////// populate values from $MIZE_CONFIG_FILES + if let Ok(config_file_paths) = obsidian_get_env_var("MIZE_CONFIG_FILES") { + for config_file_path in config_file_paths.split(":") { + println!("reading config file: {config_file_path}"); + let content = std::fs::read_to_string(config_file_path)?; + let mut data = ItemData::from_toml(content.as_str())?; + crate::core::item::set_config_from_data(mize, data)?; + } + } + + ////// load config from MIZE_CONFIG env var + match std::env::var("MIZE_CONFIG") { + Ok(config_string) => { + let data = data_from_string(config_string)?; + crate::core::item::set_config_from_data(mize, data)?; + } + Err(var_err) => match var_err { + VarError::NotPresent => { + debug!("env var MIZE_CONFIG NOT present") + } + VarError::NotUnicode(_) => { + warn!("env var MIZE_CONFIG is not Unicode, so not reading it") + } + }, + }; + + } + + Ok(()) +} + +#[cfg(feature = "target-obsidian")] +pub fn obsidian_get_env_var(name: &str) -> Option { + use js_sys::{Reflect, Object}; + // 1. Get the global `process` object + let global = js_sys::global(); + let process = Reflect::get(&global, &JsValue::from_str("process")).ok()?; + + if process.is_undefined() { + return None; + } + + // 2. Get the `env` object from `process` + let env = Reflect::get(&process, &JsValue::from_str("env")).ok()?; + if env.is_undefined() { + return None; + } + + // 3. Get the `MIZE_STORE` value from `env` + let mize_store = Reflect::get(&env, &JsValue::from_str("MIZE_STORE")).ok()?; + + // 4. Convert the JsValue to a Rust String if it exists + mize_store.as_string() +} + + +pub fn load_module(mize: &mut Mize, name: &str, path: Option) -> MizeResult<()> { + console_log!("loading module NOT IMPLEMENTED: {}", name); + + //wasm_bindgen_futures::spawn_local(load_module_async(mize.clone(), name.to_owned(), path)); + + Ok(()) +} + +/* +async fn load_module_async(mut mize: Mize, name: String, path: Option) -> () { + let name = name.as_str(); + let module_url = fetch_module(&mut mize, name).unwrap(); + let memory = wasm_bindgen::memory(); + + let mut empty_module: Box = Box::new(EmptyModule {}); + let mut mize = mize.clone(); + + let empty_module_ptr = Box::into_raw(Box::new(empty_module)); + let empty_module_usize = empty_module_ptr as usize; + let mize_usize = Box::into_raw(Box::new(mize)) as usize; + + console_log!("testttttttttttttt in load_module_async..."); + + let function_str = format!( + r#" + const promise = new Promise((resolve, reject) => {{ + import("{module_url}/mize_module_{name}.js").then( module => {{ + const wasm_bindgen = module.get_wasm_bindgen(); + + const {{ wasm_get_mize_module_mme }} = wasm_bindgen; + + wasm_bindgen("{module_url}/mize_module_{name}_bg.wasm", memory).then( () => {{ + console.log("empty_module_usize:", empty_module_usize) + console.log("mize_usize:", mize_usize) + //resolve() + resolve(wasm_get_mize_module_mme(empty_module_usize, mize_usize)) + }}) + }}) + }}) + return promise; + "# + ); + + let function = Function::new_with_args("memory, empty_module_usize, mize_usize", &function_str); + + let ret_value: JsValue = match function.call3( + &JsValue::null(), + &memory, + &empty_module_usize.into(), + &mize_usize.into(), + ) { + Ok(val) => val, + Err(err) => { + console_log!("failed to load the wasm mize module...."); + console_log!("{:?}", err); + return; + } + }; + + console_log!("got ret_value: {:?}", ret_value); + let ret_promise = Promise::from(ret_value); + let ret_result = match wasm_bindgen_futures::JsFuture::from(ret_promise).await { + Ok(val) => val, + Err(err) => { + console_log!("err awaiting ret_promise"); + console_log!("{:?}", err); + return; + } + }; + + let mut filled_module = unsafe { Box::from_raw(empty_module_ptr) }; + + console_log!("before...."); + //let a = mize.get("0/config"); + console_log!("after...."); + + //filled_module.init(&instance); + + /* + console_log!("got ret_result: {:?}", ret_result); + let ret_f64 = ret_result.as_f64().unwrap(); + console_log!("got ret_f64: {:?}", ret_f64); + + let ret_usize = ret_f64.floor() as usize; + + let mut module: Box = Box::new(EmptyModule {}); + + unsafe { + let get_mize_module_fn_ptr = ret_usize as *const fn(&mut Box, Instance) -> (); + let get_mize_module_fn: fn(&mut Box, Instance) -> () = unsafe { std::mem::transmute(get_mize_module_fn_ptr) }; + get_mize_module_fn(&mut module, instance.clone()); + } + */ +} + */ + + pub fn fetch_module(mize: &mut Mize, name: &str) -> MizeResult { + let module_name_with_slashes = name.replace(".", "/"); + if let Ok(module_path) = mize + .get(format!( + "self/config/module_dir/{}", + module_name_with_slashes + ))? + .value_string() + { + return Ok(module_path); + } else { + let module_url = mize.get("self/config/module_url")?.value_string()?; + return Ok(format!("{}/{}", module_url, module_name_with_slashes)); + } +} + +async fn download_module(mize: &mut Mize, name: &str) -> MizeResult { + use wasm_bindgen_futures::JsFuture; + use web_sys::{Request, RequestInit, RequestMode, Response}; + + let opts = RequestInit::new(); + opts.set_method("GET"); + opts.set_mode(RequestMode::Cors); + + let url = format!("{}/mize_module_{}_bg.wasm", fetch_module(mize, name)?, name); + + let request = Request::new_with_str_and_init(&url, &opts).unwrap(); + + let window = web_sys::window().unwrap(); + let resp_value = JsFuture::from(window.fetch_with_request(&request)) + .await + .unwrap(); + + let resp: Response = resp_value.dyn_into().unwrap(); + + let wasm_bytes: js_sys::ArrayBuffer = JsFuture::from(resp.array_buffer().unwrap()) + .await + .unwrap() + .dyn_into() + .unwrap(); + + Ok(wasm_bytes) +} + +#[wasm_bindgen] +#[derive(Clone)] +pub struct JsMize { + inner: NonNull, +} + +#[wasm_bindgen] +pub struct JsItem { + mize: NonNull, + id: MizeId, +} + +#[wasm_bindgen] +pub async fn new_js_instance(config_json_str: String) -> JsMize { + panic::set_hook(Box::new(console_error_panic_hook::hook)); + + let config = ItemData::from_json(config_json_str).expect("parsing of json config failed"); + + let mut mize = Mize::with_config(config).expect("Mize::new() failed"); + + let mut js_instance = JsMize { + inner: NonNull::from(Box::leak(Box::new(mize))), + }; + + return js_instance; +} + +#[wasm_bindgen] +impl JsMize { + #[wasm_bindgen] + pub unsafe fn init(&mut self) -> MizeResult<()> { + self.inner.as_mut().init() + } + + #[wasm_bindgen] + pub unsafe fn set(&mut self, id: String, value: String) -> () { + let data = value.into_item_data(); + console_log!("data in set: {}", data); + self.inner.as_mut().set(id, data); + } + + #[wasm_bindgen] + pub unsafe fn get_handle(&mut self, id: String) -> MizeResult { + let item = self.inner.as_mut().get(id)?; + Ok(JsItem { + mize: self.inner, + id: item.id(), + }) + } +} + +impl JsMize { + pub fn inner(&mut self) -> &mut Mize { + unsafe { self.inner.as_mut() } + } +} + +#[wasm_bindgen] +impl JsItem { + #[wasm_bindgen] + pub unsafe fn value_string(&mut self) -> MizeResult { + let item = self.mize.as_mut().get(self.id.clone())?; + let string = item.value_string()?; + Ok(string) + } + + #[wasm_bindgen] + pub unsafe fn as_data_full(&mut self) -> MizeResult { + let item = self.mize.as_mut().get(self.id.clone())?; + let data_raw = item.as_data_full()?; + let jsvalue = serde_wasm_bindgen::to_value(data_raw.cbor())?; + Ok(jsvalue) + } +} + +impl From for JsValue { + fn from(value: MizeError) -> Self { + let string = format!("{:?}", value); + string.into() + } +} diff --git a/core/src/platform/js/utils.rs b/core/src/platform/js/utils.rs new file mode 100644 index 0000000..e69de29 diff --git a/core/src/platform/os/fsstore.rs b/core/src/platform/os/fsstore.rs new file mode 100644 index 0000000..0e7f653 --- /dev/null +++ b/core/src/platform/os/fsstore.rs @@ -0,0 +1,218 @@ +use ciborium::Value as CborValue; +use std::fs; +use std::fs::OpenOptions; +use std::io::Write; +use std::path::{Path, PathBuf}; +use sysinfo::{Pid, ProcessRefreshKind, RefreshKind, System}; +use tracing::debug; + +use crate::core::id::MizeId; +use crate::core::item::{Item, ItemData}; +use crate::error::{IntoMizeResult, MizeError, MizeResult}; +use crate::instance::store::IdIter; +use crate::instance::{self, Mize}; +use crate::item::get_raw_from_cbor; +use crate::item::IntoItemData; +use crate::memstore::MemStore; +use crate::{core::instance::store::Store, mize_err}; +use tokio::net::UnixStream; + +#[derive(Clone, Debug)] +pub struct FileStore { + path: PathBuf, +} + +impl FileStore { + pub fn new(path_str: &str) -> MizeResult { + let path = Path::new(path_str); + + // create that path if it does not exist + fs::create_dir_all(Path::new(&path).join("store"))?; + + // check for valid pid file + /* + if let Some(pid) = valid_pid_file(path)? { + // store already opened + return Err(mize_err!("MizeStore at path {} is already opened by process with pid {}", path.display(), pid)); + + } else { + // write our own pid file + fs::remove_file(path.join("pid")); + let pid = std::process::id(); + let mut file = OpenOptions::new().write(true).create(true).open(path.join("pid"))?; + write!(file, "{}", pid)?; + } + */ + + // init the store + if !path.join("next_id").exists() { + fs::write(path.join("next_id"), "1"); + } + + Ok(FileStore { + path: Path::new(&path).to_owned(), + }) + } + + pub fn store_is_opened(store_path: String, instance: &mut Mize) -> MizeResult { + // this was the old method.... + //let valid_pid_file = valid_pid_file(Path::new(&store_path))?.is_some(); + + let can_connect_to_sock = match instance + .spawn_async_blocking("con_test_thread", try_connect_to_sock(store_path.clone())) + { + Ok(_) => true, + Err(_) => false, + }; + debug!( + "MiZe Store at {} opened: {}", + store_path, can_connect_to_sock + ); + Ok(can_connect_to_sock) + } +} + +async fn try_connect_to_sock(store_path: String) -> MizeResult<()> { + match UnixStream::connect(PathBuf::from(store_path).join("sock")).await { + Err(_) => Err(mize_err!("")), + Ok(_) => Ok(()), + } +} + +impl Store for FileStore { + fn new_id(&self) -> MizeResult { + let mut next_id: u64 = + String::from_utf8(fs::read(self.path.join("next_id")).mres_msg(format!( + "could not read next_id at '{}'", + self.path.display() + ))?)? + .parse() + .mres_msg(format!( + "could not parse next_id at '{}' to u64", + self.path.display() + ))?; + + let id_string = format!("{}", next_id); + + next_id += 1; + + fs::write(self.path.join("next_id"), format!("{}", next_id))?; + + return Ok(id_string); + } + + fn set(&self, id: MizeId, data: ItemData) -> MizeResult<()> { + let path = self + .path + .join("store") + .join(id.namespace_str()) + .join(id.store_part()); + fs::create_dir_all(self.path.join("store").join(id.namespace_str()))?; + + let file = OpenOptions::new().write(true).create(true).open(path)?; + + ciborium::into_writer(data.cbor(), file)?; + Ok(()) + } + + fn get_links(&self, item: Item) -> MizeResult> { + Ok(Vec::new()) + } + + fn get_backlinks(&self, item: Item) -> MizeResult> { + Ok(Vec::new()) + } + + fn get_value_raw(&self, id: MizeId) -> MizeResult> { + let path = self + .path + .join("store") + .join(id.namespace_str()) + .join(id.store_part()); + + let file = OpenOptions::new().read(true).create(true).open(path)?; + + let cbor_val: CborValue = ciborium::from_reader(file).mres_msg(format!( + "could not read file '{}' from FileStore", + self.path.join("store").join(id.store_part()).display() + ))?; + + let path = id.path(); + let mut id_iter = path.iter(); + id_iter.next(); + let id_without_first = id_iter.collect(); + + Ok(get_raw_from_cbor(&cbor_val, id_without_first)?.to_owned()) + } + + fn get_value_data_full(&self, id: MizeId) -> MizeResult { + let file_path = self + .path + .join("store") + .join(id.namespace_str()) + .join(id.store_part()); + if !file_path.exists() {} + let file = OpenOptions::new() + .read(true) + .open(file_path.clone()) + .map_err(|e| { + mize_err!( + "fsstoree: Unable to open '{}', err: {}", + file_path.display(), + e + ) + })?; + + let cbor_value: CborValue = ciborium::from_reader(file).mres_msg(format!( + "could not read file '{}' from FileStore", + self.path.join("store").join(id.store_part()).display() + ))?; + let data: ItemData = cbor_value.into_item_data(); + + let tmp = id.path(); + let mut path_iter = tmp.into_iter(); + path_iter.next(); + let new_path: Vec = path_iter.map(|v| v.to_owned()).collect(); + //println!("data: {}", data); + //println!("new_path: {:?}", new_path); + let ret_data = data.get_path(new_path)?; + //println!("ret_data: {}", ret_data); + + return Ok(ret_data); + } + fn id_iter(&self) -> MizeResult { + todo!() + } + fn next_id(&self, prev_id: &str) -> MizeResult> { + todo!() + } + fn first_id(&self) -> MizeResult { + Ok("0".to_owned()) + } +} + +fn valid_pid_file(path: &Path) -> MizeResult> { + let pid_file_path = path.join("pid"); + + // if the pid file does not exist, there is no pid... + if !pid_file_path.exists() { + return Ok(None); + } + + // read pid file + let pid: u32 = String::from_utf8(fs::read(&pid_file_path).mres_msg(format!( + "Could not read contents of pid file at '{}'", + pid_file_path.display() + ))?)? + .parse() + .mres_msg(format!( + "Could not parse contents of pid file at '{}' to u32", + pid_file_path.display() + ))?; + + if Path::new("/proc").join(format!("{}", pid)).exists() { + return Ok(Some(pid)); + } else { + return Ok(None); + } +} diff --git a/core/src/platform/os/logging.rs b/core/src/platform/os/logging.rs new file mode 100644 index 0000000..a862318 --- /dev/null +++ b/core/src/platform/os/logging.rs @@ -0,0 +1,353 @@ +use clap::{crate_version, Arg, Command}; +use clap::{ArgAction, ArgMatches}; +use std::collections::HashMap; +use std::path::PathBuf; + +use colored::Colorize; +use core::fmt; +use std::env; +use std::fmt::Write; +use std::sync::atomic::AtomicUsize; +use std::sync::Arc; +use std::sync::Mutex; +use tokio::sync::mpsc; +use tracing::level_filters::LevelFilter; +use tracing::Event; +use tracing::{debug, error, info, trace, warn}; +use tracing::{field::Visit, Id, Subscriber}; +use tracing::{span_enabled, Instrument, Level}; +use tracing_core::Field; +use tracing_subscriber::filter::EnvFilter; +use tracing_subscriber::fmt::format; +use tracing_subscriber::fmt::FmtContext; +use tracing_subscriber::fmt::FormatEvent; +use tracing_subscriber::fmt::FormatFields; +use tracing_subscriber::fmt::FormattedFields; +use tracing_subscriber::registry::LookupSpan; + +use crate::id::MizeId; +use crate::item::ItemData; +use crate::MizeError; +use crate::MizeResult; + +static DEFAULT_LOG_LEVEL: LevelFilter = LevelFilter::WARN; + +pub fn init_logger(cli_matches: &ArgMatches) { + println!("init_logger called"); + // 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 = tracing_subscriber::fmt::Subscriber::builder(); + + // 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 LOG + let mut val_mize_log = level_from_env("LOG", log_level); + if let Some(level_mize_log) = val_mize_log.0 { + println!("set log level from the Variable LOG to {}", level_mize_log); + 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 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"); + if v_flag_count > 0 { + 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::("log-level") { + let log_level_from_arg = match log_level_arg_string.to_owned().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 + ), + )); + } + + // actually set the log_level on the builder + log_messages.push((Level::INFO, format!("loglevel is now {}", log_level))); + + // filter loging + //let mut my_env_filter = tracing_subscriber::filter::EnvFilter::new("") + //.add_directive("[get_raw_from_cbor]=debug".parse().expect("error parsing log directive")) + //.add_directive("[MemStore::get_value_raw]=trace".parse().expect("error parsing log directive")) + //.add_directive("trace".parse().unwrap()) + //.add_directive("[get_raw_from_cbor]=debug".parse().expect("error parsing log directive")) + //; + + //MinimalTracer::register(); + + println!("log_level: {}", log_level); + let field_formatter = tracing_subscriber::fmt::format::debug_fn(|writer, field, value| { + if field.name() == "message" { + write!(writer, "{:?}", value)?; + } else { + write!(writer, "{}={:?}", field, value)?; + } + Ok(()) + }); + builder + //.with_env_filter(my_env_filter) + .without_time() + .event_format(MyFormatter) + .fmt_fields(field_formatter) + .with_max_level(log_level) + .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 level_from_env( + var_name: &str, + log_level: LevelFilter, +) -> (Option, Vec<(Level, String)>) { + let mut log_messages: Vec<(Level, String)> = Vec::new(); + let var_value = env::var(var_name); + println!("checking env var: {}={:?}", var_name, var_value); + + 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::>() + .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); +} + +struct MyFormatter; + +impl FormatEvent for MyFormatter +where + S: tracing::Subscriber + for<'a> LookupSpan<'a>, + N: for<'a> FormatFields<'a> + 'static, +{ + fn format_event( + &self, + ctx: &FmtContext<'_, S, N>, + mut writer: format::Writer<'_>, + event: &Event<'_>, + ) -> fmt::Result { + // Format values from the event's's metadata: + + let metadata = event.metadata(); + + // set format + //builder.format(|buf, record| match record.level() {}); + + let level_str = match metadata.level() { + &Level::TRACE => format!("[ {} ]", "TRACE".yellow()), + &Level::DEBUG => format!("[ {} ]", "DEBUG".green()), + &Level::INFO => format!("[ {} ]", "INFO".blue()), + &Level::WARN => format!("[ {} ]", "WARN".truecolor(245, 164, 66)), + &Level::ERROR => format!("[ {} ]", "ERROR".red()), + }; + + // with target + // write!(&mut writer, "{} {}: ", level_str, metadata.target())?; + + write!(&mut writer, "{} ", level_str)?; + + // Format all the spans in the event's span context. + if let Some(scope) = ctx.event_scope() { + for span in scope.from_root() { + write!(writer, "{}", span.name())?; + + // `FormattedFields` is a formatted representation of the span's + // fields, which is stored in its extensions by the `fmt` layer's + // `new_span` method. The fields will have been formatted + // by the same field formatter that's provided to the event + // formatter in the `FmtContext`. + let ext = span.extensions(); + let fields = &ext + .get::>() + .expect("will never be `None`"); + + // Skip formatting the fields if the span had no fields. + if !fields.is_empty() { + //write!(writer, "{{{}}}", fields)?; + } + write!(writer, ": ")?; + } + } + + // Write fields on the event + ctx.field_format().format_fields(writer.by_ref(), event)?; + + writeln!(writer); + Ok(()) + } +} + +// thanks: https://stackoverflow.com/a/75546824 +pub struct StringVisitor<'a> { + string: &'a mut String, +} +impl<'a> StringVisitor<'a> { + pub(crate) fn new(string: &'a mut String) -> Self { + StringVisitor { string } + } +} + +impl<'a> Visit for StringVisitor<'a> { + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + if field.name() == "message" { + write!(self.string, "{value:?} ").unwrap(); + } else { + write!(self.string, "{} = {:?}; ", field.name(), value).unwrap(); + } + } +} + +struct LogFilter { + target: Option, + level: Option, +} + +fn string_to_level(string: &str) -> Option { + match string.to_lowercase().as_str() { + "info" => Some(Level::INFO), + "debug" => Some(Level::DEBUG), + "warn" | "warning" => Some(Level::WARN), + "trace" => Some(Level::TRACE), + "error" => Some(Level::ERROR), + _ => None, + } +} diff --git a/core/src/platform/os/unix_socket.rs b/core/src/platform/os/unix_socket.rs new file mode 100644 index 0000000..03d264a --- /dev/null +++ b/core/src/platform/os/unix_socket.rs @@ -0,0 +1,233 @@ +use ciborium::Value as CborValue; +use flume::{unbounded, Receiver, Sender}; +use std::fs; +use std::path::{Path, PathBuf}; +use tokio::io::{AsyncReadExt, AsyncWriteExt, Interest}; +use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::net::{UnixListener as TokioUnixListener, UnixStream}; +use tracing::{debug, info, warn}; + +use crate::error::{IntoMizeResult, MizeError, MizeResult}; +use crate::instance::connection::{ConnListener, Connection}; +use crate::instance::{self, Mize}; +use crate::proto::{self, MizeMessage}; + +pub struct UnixListener { + sock_path: PathBuf, +} + +impl UnixListener { + pub fn new(store_path: PathBuf) -> MizeResult { + Ok(UnixListener { + sock_path: store_path.join("sock").to_owned(), + }) + } +} + +pub fn connect(instance: &mut Mize, store_path: PathBuf) -> MizeResult<()> { + let conn_id = instance.spawn_async_blocking( + "unix_connection", + connect_async(instance.clone(), store_path), + )?; + + let ns_of_peer_str = instance + .get(format!( + "inst/con_by_id/{}/peer/self/config/namespace", + conn_id + ))? + .value_string()?; + let ns_of_peer = instance.namespace_from_string(ns_of_peer_str)?; + + info!( + "connect_async ... ns_of_peer: {}", + ns_of_peer.clone().as_string() + ); + + instance.connection_set_namespace(conn_id, ns_of_peer.clone()); + instance.set_namespace(ns_of_peer); + + Ok(()) +} + +async fn connect_async(mut instance: Mize, store_path: PathBuf) -> MizeResult { + let mut stream = UnixStream::connect(store_path.join("sock")).await?; + let (unix_read, unix_write) = stream.into_split(); + + let (send_tx, send_rx) = unbounded::(); + + let conn_id = instance.new_connection(send_tx)?; + + let cloned_instance = instance.clone(); + instance.spawn_background("incomming", move || { + let result = unix_incomming(unix_read, cloned_instance, conn_id); + // if unix incomming fails, close the connection + if let Err(err) = result { + warn!("Connection closing because of EOF"); + } + Ok(()) + }); + + let outgoing_cloned_instance = instance.clone(); + instance.spawn_background("outgoing", move || { + let result = unix_outgoing(unix_write, send_rx, outgoing_cloned_instance, conn_id); + // if writing fails, close this connection + if let Err(err) = result { + warn!("Connection closing because of EOF"); + } + Ok(()) + }); + + return Ok(conn_id); +} + +impl ConnListener for UnixListener { + fn listen(self, mut instance: Mize) -> MizeResult<()> { + instance.spawn_async("unix listen async", unix_listen(self, instance.clone())); + Ok(()) + } +} + +async fn unix_listen(listener: UnixListener, mut instance: Mize) -> MizeResult<()> { + // remove the file at sock_path if it already exists + fs::remove_file(&listener.sock_path); + + let listener = TokioUnixListener::bind(&listener.sock_path).mres_msg(format!( + "Could not bind to unix socket at '{}'", + listener.sock_path.display() + ))?; + + loop { + let (mut unix_sock, addr) = listener + .accept() + .await + .mres_msg("Error while accepting Unix sock connection")?; + info!("new connection"); + + let (send_tx, send_rx) = unbounded::(); + let (unix_read, unix_write) = unix_sock.into_split(); + + let conn_id = instance.new_connection(send_tx)?; + let cloned_instance = instance.clone(); + instance.spawn_background("incomming", move || { + let result = unix_incomming(unix_read, cloned_instance, conn_id); + // if unix incomming fails, close the connection + if let Err(err) = result { + warn!("Connection closing because of EOF"); + } + Ok(()) + }); + + let outgoing_cloned_instance = instance.clone(); + instance.spawn_background("outgoing", move || { + let result = unix_outgoing(unix_write, send_rx, outgoing_cloned_instance, conn_id); + // if writing fails, close this connection + if let Err(err) = result { + warn!("Connection closing because of EOF"); + } + Ok(()) + }); + } +} + +fn unix_outgoing( + mut unix_write: OwnedWriteHalf, + send_rx: Receiver, + mut instance: Mize, + conn_id: u64, +) -> MizeResult<()> { + for msg in send_rx { + debug!("unix outgoing got msg: {}", msg); + let adapter = MyCiboriumWriter { + inner: &mut unix_write, + instance: &mut instance, + }; + let value = msg.value(); + ciborium::into_writer(&value, adapter)? + } + Ok(()) +} + +pub fn unix_incomming( + mut unix_read: OwnedReadHalf, + mut instance: Mize, + conn_id: u64, +) -> MizeResult<()> { + loop { + //println!("unix incoming loop..."); + let adapter = MyCiboriumReader { + inner: &mut unix_read, + instance: &mut instance, + }; + let value: CborValue = ciborium::from_reader(adapter)?; + if let CborValue::Integer(_) = value { + break; + } + let msg = MizeMessage::new(value, conn_id); + debug!("unix incoming got msg: {}", msg); + instance.got_msg(msg); + } + Ok(()) +} + +struct MyCiboriumWriter<'a> { + inner: &'a mut OwnedWriteHalf, + instance: &'a mut Mize, +} + +impl<'a> ciborium_io::Write for MyCiboriumWriter<'a> { + type Error = MizeError; + fn write_all(&mut self, data: &[u8]) -> Result<(), Self::Error> { + //println!("CALL: write_all"); + let runtime = self.instance.runtime.lock()?; + //println!("after runtime"); + let num_read = runtime.block_on(self.inner.write_all(data))?; + Ok(()) + } + fn flush(&mut self) -> Result<(), Self::Error> { + //println!("CALL flush"); + let handle = self.instance.async_get_handle(); + let result = handle.block_on(self.inner.flush()); + //println!("flush result: {:?}", result); + Ok(()) + } +} + +struct MyCiboriumReader<'a> { + inner: &'a mut OwnedReadHalf, + instance: &'a mut Mize, +} + +async fn test(test: &mut OwnedReadHalf, buf: &mut [u8]) { + //println!("hello from async..."); + //println!("len: {}", buf.len()); + let num_read = test.read_u8().await; + //println!("num_read: {:?}", num_read); + //test.read_exact(buf).await; + //println!("hello from async... twoooooooooooooo"); +} + +impl<'a> ciborium_io::Read for MyCiboriumReader<'a> { + type Error = MizeError; + fn read_exact(&mut self, data: &mut [u8]) -> Result<(), Self::Error> { + //println!("data before: {:?}", data); + //let runtime = self.instance.runtime.lock()?; + //println!("are we here after runtime lock()"); + //runtime.block_on(test(self.inner, data)); + //println!("are we hereeeeeeeeeeeeeeeeeeee"); + + //return Ok(()); + + //println!("incoming before block_on"); + let handle = self.instance.async_get_handle(); + let num_read = handle.block_on(self.inner.read_exact(data))?; + //println!("incoming num_read: {:?}", num_read); + //println!("data after: {:?}", data); + + // if we read 0 bytes, that means the reader stream closed and we should terminate the + // connection + //if num_read == 0 { + //return Err(MizeError::new().msg("ReceiverStream closed")); + //} + Ok(()) + } +} diff --git a/core/src/platform/os/web.rs b/core/src/platform/os/web.rs new file mode 100644 index 0000000..8587e0a --- /dev/null +++ b/core/src/platform/os/web.rs @@ -0,0 +1,239 @@ + +static SERVER_PORT: u16 = 3000; +static API_ENDPOINT: &str = "@mize"; + + +pub fn run() -> Result { + /* + * - 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, State(mutexes): State) -> http::Response { +// 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, Receiver) = 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; + }, + + }; + }; +}