diff --git a/Cargo.lock b/Cargo.lock index b924521..43f39a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -455,6 +455,28 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8" +dependencies = [ + "crossbeam-channel", + "crossbeam-deque", + "crossbeam-epoch", + "crossbeam-queue", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-channel" +version = "0.5.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33480d6946193aa8033910124896ca395333cae7e2d1113d1fef6c3272217df2" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-deque" version = "0.8.5" @@ -474,6 +496,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-queue" +version = "0.3.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df0346b5d5e76ac2fe4e327c5fd1118d6be7c51dfb18f9b7922923f287471e35" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.20" @@ -542,6 +573,12 @@ dependencies = [ "strsim 0.10.0", ] +[[package]] +name = "dyn-clone" +version = "1.0.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d6ef0072f8a535281e4876be788938b528e9a1d43900b82c2569af7da799125" + [[package]] name = "either" version = "1.12.0" @@ -561,19 +598,6 @@ dependencies = [ "termcolor", ] -[[package]] -name = "env_logger" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4cd405aab171cb85d6735e5c8d9db038c17d3ca007a4d2c25f337935c3d90580" -dependencies = [ - "humantime", - "is-terminal", - "log", - "regex", - "termcolor", -] - [[package]] name = "equivalent" version = "1.0.1" @@ -939,17 +963,6 @@ name = "interner" version = "0.2.1" source = "git+https://github.com/c2vi/interner#746e7586bb65e2e50ac02140c6414574efb2eb4e" -[[package]] -name = "is-terminal" -version = "0.4.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f23ff5ef2b80d608d61efee834934d862cd92461afc0560dedf493e4c033738b" -dependencies = [ - "hermit-abi 0.3.9", - "libc", - "windows-sys 0.52.0", -] - [[package]] name = "is_terminal_polyfill" version = "1.70.0" @@ -1001,6 +1014,15 @@ version = "0.4.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "90ed8c1e510134f979dbc4f070f87d4313098b704861a105fe34231c70a3901c" +[[package]] +name = "matchers" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8263075bb86c5a1b1427b5ae862e8889656f126e9f77c484496e8b47cf5c5558" +dependencies = [ + "regex-automata 0.1.10", +] + [[package]] name = "matchit" version = "0.7.3" @@ -1060,17 +1082,19 @@ dependencies = [ "clap", "colored", "console_error_panic_hook", + "crossbeam", "daemonize", - "env_logger 0.10.2", + "dyn-clone", "home", "interner", - "log", "nix", "serde", "serde_json", "sysinfo", "tokio", "toml", + "tracing", + "tracing-subscriber", "uuid", "wasm-bindgen", "wasm-bindgen-cli", @@ -1115,6 +1139,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "nu-ansi-term" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84" +dependencies = [ + "overload", + "winapi", +] + [[package]] name = "num-conv" version = "0.1.0" @@ -1164,6 +1198,12 @@ version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92" +[[package]] +name = "overload" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" + [[package]] name = "percent-encoding" version = "2.3.1" @@ -1305,8 +1345,17 @@ checksum = "c117dbdfde9c8308975b6a18d71f3f385c89461f7b3fb054288ecf2a2058ba4c" dependencies = [ "aho-corasick", "memchr", - "regex-automata", - "regex-syntax", + "regex-automata 0.4.6", + "regex-syntax 0.8.3", +] + +[[package]] +name = "regex-automata" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132" +dependencies = [ + "regex-syntax 0.6.29", ] [[package]] @@ -1317,9 +1366,15 @@ checksum = "86b83b8b9847f9bf95ef68afb0b8e6cdb80f498442f5179a29fad448fcc1eaea" dependencies = [ "aho-corasick", "memchr", - "regex-syntax", + "regex-syntax 0.8.3", ] +[[package]] +name = "regex-syntax" +version = "0.6.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1" + [[package]] name = "regex-syntax" version = "0.8.3" @@ -1516,6 +1571,15 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae1a47186c03a32177042e55dbc5fd5aee900b8e0069a8d70fba96a9375cd012" +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "slab" version = "0.4.9" @@ -1525,6 +1589,12 @@ dependencies = [ "autocfg", ] +[[package]] +name = "smallvec" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" + [[package]] name = "socket2" version = "0.5.7" @@ -1649,6 +1719,16 @@ dependencies = [ "syn 2.0.66", ] +[[package]] +name = "thread_local" +version = "1.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b9ef9bad013ada3808854ceac7b46812a6465ba368859a37e2100283d2d719c" +dependencies = [ + "cfg-if", + "once_cell", +] + [[package]] name = "threadpool" version = "1.8.1" @@ -1716,6 +1796,7 @@ dependencies = [ "bytes", "libc", "mio", + "num_cpus", "pin-project-lite", "socket2", "windows-sys 0.48.0", @@ -1840,9 +1921,21 @@ checksum = "c3523ab5a71916ccf420eebdf5521fcef02141234bbc0b8a49f2fdc4544364ef" dependencies = [ "log", "pin-project-lite", + "tracing-attributes", "tracing-core", ] +[[package]] +name = "tracing-attributes" +version = "0.1.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.66", +] + [[package]] name = "tracing-core" version = "0.1.32" @@ -1850,6 +1943,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c06d3da6113f116aaee68e4d601191614c9053067f9ab7f6edbcb161237daa54" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad0f048c97dbd9faa9b7df56362b8ebcaa52adb06b498c050d2f4e32f90a7a8b" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -1981,6 +2104,12 @@ version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a183cf7feeba97b4dd1c0d46788634f6221d87fa961b305bed08c851829efcc0" +[[package]] +name = "valuable" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d" + [[package]] name = "version_check" version = "0.9.4" @@ -2064,7 +2193,7 @@ checksum = "08f61e21873f51e3059a8c7c3eef81ede7513d161cfc60751c7b2ffa6ed28270" dependencies = [ "anyhow", "docopt", - "env_logger 0.8.4", + "env_logger", "log", "rouille", "serde", diff --git a/Cargo.toml b/Cargo.toml index 982430a..ab0a29b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,8 @@ path = "src/platform/os/main.rs" required-features = [ "os-target" ] [features] -os-target = [ "clap", "home", "env_logger", "daemonize", "nix", "toml", "serde_json", "sysinfo", "ciborium/default"] +os-target = [ "clap", "home", "daemonize", "nix", "toml", "serde_json", "sysinfo", "ciborium/default", "tracing-subscriber", "async", "tracing-subscriber/env-filter"] +async = ["tokio/net", "tokio", "tokio/rt-multi-thread"] wasm-target = [ "wasm-bindgen", "console_error_panic_hook", "wasm-bindgen-cli" ] @@ -25,11 +26,10 @@ wasm-target = [ "wasm-bindgen", "console_error_panic_hook", "wasm-bindgen-cli" ] # core dependencies colored = "2.0.0" uuid = "1.3.0" -log = "0.4.20" -tokio = { version = "1.21.0", features = ["sync"] } ciborium = { version = "0.2.2", default-features = false } serde = "1.0.203" interner = { git = "https://github.com/c2vi/interner" } +tracing = "0.1.40" # wasm dependencies wasm-bindgen = { version = "0.2.84", optional = true } @@ -39,12 +39,13 @@ wasm-bindgen-cli = { version = "0.2.92", optional = true } # os dependencies clap = { version = "4.4.11", features = ["cargo"], optional = true } home = { version = "0.5.9", optional = true } -env_logger = { version = "0.10.1", optional = true } daemonize = { version = "0.5.0", optional = true } nix = { version = "0.27.1", features = ["signal"], optional = true } toml = { version = "0.8.13", features = ["parse"], optional = true } serde_json = { version = "1.0.117", optional = true } sysinfo = { version = "0.30.12", optional = true } +tracing-subscriber = { version = "0.3.18", optional = true } +tokio = { version = "1.21.0", features = ["sync"], optional = true } # bare dependencies @@ -52,6 +53,8 @@ sysinfo = { version = "0.30.12", optional = true } axum = { version = "0.6.1", features = ["ws", "headers"], optional = true } axum-extra = { version = "0.4.2", features = ["spa"], optional = true } axum-macros = { version = "0.3.0", optional = true } +crossbeam = "0.8.4" +dyn-clone = "1.0.17" #tokio-stream = { version = "0.1.9", optional = true } #derive_more = "0.99.17" diff --git a/src/core/error.rs b/src/core/error.rs index 56cdf4d..fc11f51 100644 --- a/src/core/error.rs +++ b/src/core/error.rs @@ -5,7 +5,7 @@ use std::fmt::format; use std::string::FromUtf8Error; use std::io; use colored::Colorize; -use log::{trace, debug, info, warn, error}; +use tracing::{trace, debug, info, warn, error}; #[macro_export] macro_rules! mize_err { diff --git a/src/core/id.rs b/src/core/id.rs index 7a3a090..5597bf9 100644 --- a/src/core/id.rs +++ b/src/core/id.rs @@ -5,6 +5,7 @@ use std::collections::hash_map::RandomState; use std::path::Path; use crate::instance::{store::Store, Instance}; +use crate::error::{MizeResult, MizeError, MizeResultTrait}; #[derive(Debug, Clone, Hash, Eq, PartialEq)] pub struct MizeId { @@ -22,7 +23,7 @@ impl Namespace { } pub trait IntoMizeId { - fn to_mize_id(self, instance: &Instance) -> MizeId; + fn to_mize_id(self, instance: &Instance) -> MizeResult; } @@ -54,28 +55,28 @@ impl fmt::Display for MizeId { } impl IntoMizeId for &str { - fn to_mize_id(self, instance: &Instance) -> MizeId { + fn to_mize_id(self, instance: &Instance) -> MizeResult { instance.id_from_string(self.to_owned()) } } impl IntoMizeId for String { - fn to_mize_id(self, instance: &Instance) -> MizeId { + fn to_mize_id(self, instance: &Instance) -> MizeResult { instance.id_from_string(self) } } impl IntoMizeId for &String { - fn to_mize_id(self, instance: &Instance) -> MizeId { + fn to_mize_id(self, instance: &Instance) -> MizeResult { instance.id_from_string(self.to_owned()) } } impl IntoMizeId for Vec { - fn to_mize_id(self, instance: &Instance) -> MizeId { + fn to_mize_id(self, instance: &Instance) -> MizeResult { instance.id_from_vec_string(self) } } impl IntoMizeId for &[&str] { - fn to_mize_id(self, instance: &Instance) -> MizeId { + fn to_mize_id(self, instance: &Instance) -> MizeResult { let mut vec: Vec = Vec::new(); for i in self { vec.push((*i).to_owned()) @@ -84,7 +85,7 @@ impl IntoMizeId for &[&str] { } } impl IntoMizeId for Vec<&str> { - fn to_mize_id(self, instance: &Instance) -> MizeId { + fn to_mize_id(self, instance: &Instance) -> MizeResult { let mut vec: Vec = Vec::new(); for i in self { vec.push((*i).to_owned()) @@ -93,8 +94,8 @@ impl IntoMizeId for Vec<&str> { } } impl IntoMizeId for MizeId { - fn to_mize_id(self, instance: &Instance) -> MizeId { - self + fn to_mize_id(self, instance: &Instance) -> MizeResult { + Ok(self) } } diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs deleted file mode 100644 index 93db5e7..0000000 --- a/src/core/instance/connection.rs +++ /dev/null @@ -1,16 +0,0 @@ -use log::{trace, debug, info, warn, error}; -use tokio::sync::mpsc::{Sender, Receiver}; - -use crate::proto::MizeMessage; -use crate::error::{MizeError, MizeResult, IntoMizeResult}; - -pub trait Connection { -} - - -//#[derive(Debug)] -//pub struct Connection { - //tx: Sender, - //rx: Receiver, -//} - diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 14a7239..be240e3 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -1,59 +1,102 @@ use core::fmt; use std::collections::HashMap; use std::fs::create_dir; +use std::{thread, vec}; +use crossbeam::channel; use std::sync::{Arc, Mutex}; -use log::{trace, debug, info, warn, error}; +use tracing::{trace, debug, info, warn, error}; use uuid::Uuid; use std::fs::File; use colored::Colorize; use std::path::Path; use interner::shared::{VecStringPool, StringPool}; + use crate::error::{MizeError, MizeResult, IntoMizeResult, MizeResultTrait}; -use crate::instance; use crate::instance::store::Store; -use crate::instance::connection::Connection; +use crate::instance::updater::updater_thread; +use crate::instance::peer::Peer; use crate::id::{IntoMizeId, MizeId, Namespace}; use crate::instance::subscription::Subscription; use crate::item::{Item, ItemData}; use crate::memstore::MemStore; +use crate::instance::updater::Operation; -pub mod connection; +use self::peer::PeerListener; + +#[cfg(feature = "async")] +use tokio::runtime::Runtime; +use core::future::Future; + +pub mod peer; pub mod store; pub mod subscription; +pub mod updater; static MSG_CHANNEL_SIZE: usize = 200; /// The Instance type is the heart of the mize system +#[derive(Clone)] pub struct Instance { // question vec of stores??? // would mean replication logic and such things is handeled by the instance // i think it would be better thogh to then implement a ReplicatedStore pub store: Box, - peers: Arc>>>, + peers: Arc>>, subs: Arc>>, - pub id_pool: VecStringPool, - pub namespace_pool: StringPool, + pub id_pool: Arc>, + pub namespace_pool: Arc>, pub namespace: Namespace, + op_tx: channel::Sender, + context: Vec, + threads: Vec, + #[cfg(feature = "async")] + runtime: Arc>, +} + +pub struct InstanceRef { + inner: Arc>, +} + +pub struct InstanceAsync { + inner: Arc>, } impl Instance { - pub fn empty() -> Instance { + pub fn empty() -> MizeResult { let store = MemStore::new(); - let id_pool = VecStringPool::default(); - let namespace_pool = StringPool::default(); + let id_pool = Arc::new(Mutex::new(VecStringPool::default())); + let namespace_pool_raw = StringPool::default(); let peers = Arc::new(Mutex::new(Vec::new())); let subs = Arc::new(Mutex::new(HashMap::new())); - let namespace = Namespace ( namespace_pool.get("mize.default.namespace") ); - let mut instance = Instance { store: Box::new(store), peers, subs, id_pool, namespace_pool, namespace }; - return instance; + let namespace = Namespace ( namespace_pool_raw.get("mize.default.namespace") ); + let (op_tx, op_rx) = channel::unbounded(); + + let mut instance = Instance { + store: Box::new(store), + peers, subs, id_pool, + namespace, op_tx, + namespace_pool: Arc::new(Mutex::new(namespace_pool_raw)), + context: vec![], + threads: vec![], + + #[cfg(feature = "async")] + runtime: Arc::new(Mutex::new(Runtime::new().mize_result_msg("Could not create async runtime")?)), + }; + + let instance_clone = instance.clone(); + let closure = move || updater_thread(op_rx, instance_clone); + instance.spawn("updater_thread", closure)?; + + + return Ok(instance); } pub fn new() -> MizeResult { trace!("[ {} ] Instance::new()", "CALL".yellow()); - let mut instance = Instance::empty(); + let mut instance = Instance::empty()?; instance.init(); @@ -77,7 +120,7 @@ impl Instance { pub fn with_config(config: ItemData) -> MizeResult { trace!("[ {} ] Instance::with_config()", "CALL".yellow()); trace!("config: {}", config); - let mut instance = Instance::empty(); + let mut instance = Instance::empty()?; instance.set("0", config.clone()); instance.init()?; @@ -94,56 +137,86 @@ impl Instance { println!("MIGRATING"); let old_store = &self.store; - let inst_data = self.store.get_value_data_full(self.id_from_string("0".to_owned()))?; - new_store.set(self.id_from_string("0".to_owned()), inst_data.to_owned())?; + let id = self.id_from_string("0".to_owned())?; + let inst_data = self.store.get_value_data_full(id.clone())?; + new_store.set(id, inst_data.to_owned())?; for id in old_store.id_iter()? { - let data = self.store.get_value_data_full(self.id_from_string(id?))?; + let data = self.store.get_value_data_full(self.id_from_string(id?)?)?; let id_of_new_store = new_store.new_id()?; - new_store.set(self.id_from_string(id_of_new_store), data.to_owned())?; + new_store.set(self.id_from_string(id_of_new_store)?, data.to_owned())?; }; self.store = new_store; Ok(()) } -} -impl Instance { pub fn new_item(&self) -> MizeResult { - let id = self.id_from_string(self.store.new_id()?); + let id = self.id_from_string(self.store.new_id()?)?; return Ok(Item::new(id, &self)); } pub fn get(&self, id: I) -> MizeResult { - let id = id.to_mize_id(self); + let id = id.to_mize_id(self)?; return Ok(Item::new(id, &self)); } pub fn set>(&mut self, id: I, value: V) -> MizeResult<()> { - let id = id.to_mize_id(self); - let item_data = value.into(); - let mut item = self.get(id.clone())?; - item.merge(item_data)?; + let id = id.to_mize_id(self)?; + self.op_tx.send(Operation::Set(id, value.into())); Ok(()) } - pub fn new_id(&self, value: T) -> MizeId { + pub fn new_id(&self, value: T) -> MizeResult { value.to_mize_id(self) } - pub fn id_from_string(&self, string: String) -> MizeId { + pub fn id_from_string(&self, string: String) -> MizeResult { let vec_string: Vec = string.split("/").map(|v| v.to_owned()).collect(); - MizeId { path: self.id_pool.get(vec_string), namespace: self.namespace.clone() } + let id_pool_inner = self.id_pool.lock()?; + + let id = MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace.clone() }; + + Ok(id) } - pub fn id_from_vec_string(&self, vec_string: Vec) -> MizeId { - MizeId { path: self.id_pool.get(vec_string), namespace: self.namespace.clone() } + pub fn id_from_vec_string(&self, vec_string: Vec) -> MizeResult { + let id_pool_inner = self.id_pool.lock()?; + let id = MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace.clone() }; + Ok(id) } - pub fn namespace_from_string(&self, ns_str: String) -> Namespace { - Namespace ( self.namespace_pool.get(ns_str) ) + pub fn namespace_from_string(&self, ns_str: String) -> MizeResult { + let namespace_pool_inner = self.namespace_pool.lock()?; + let namespace = Namespace ( namespace_pool_inner.get(ns_str) ); + Ok(namespace) + } + + pub fn add_listener(&mut self, listener: T) -> MizeResult<()> { + let mut instance_clone = self.clone(); + self.spawn("some_listener", move || listener.listen(instance_clone)); + Ok(()) + } + + pub fn add_peer(&mut self, peer: Peer) -> MizeResult<()> { + let mut peer_inner = self.peers.lock()?; + peer_inner.push(peer); + Ok(()) + } + + pub fn spawn(&mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static) -> MizeResult<()> { + self.threads.push(name.to_owned()); + thread::spawn(move || func()); + Ok(()) + } + + #[cfg(feature = "async")] + pub fn spawn_async + Send + Sync + 'static>(&mut self, name: &str, func: F) { + self.threads.push(name.to_owned()); + let runtime_inner = self.runtime.lock().unwrap(); + runtime_inner.spawn(func); } } diff --git a/src/core/instance/peer.rs b/src/core/instance/peer.rs new file mode 100644 index 0000000..fb9582e --- /dev/null +++ b/src/core/instance/peer.rs @@ -0,0 +1,25 @@ +use crossbeam::channel::{Sender, Receiver}; +use tracing::{trace, debug, info, warn, error}; + +use crate::proto::MizeMessage; +use crate::error::{MizeError, MizeResult, IntoMizeResult}; + +use super::Instance; + +pub struct Peer { + rx: Receiver, + tx: Sender, +} + +pub trait PeerListener : Send + Sync { + fn listen(self, instance: Instance) -> MizeResult<()>; +} + +impl Peer { + pub fn new(rx: Receiver, tx: Sender) -> Peer { + Peer { rx , tx } + } +} + + + diff --git a/src/core/instance/store.rs b/src/core/instance/store.rs index bd816cc..dc4cec4 100644 --- a/src/core/instance/store.rs +++ b/src/core/instance/store.rs @@ -1,14 +1,17 @@ - use std::iter::Map; use std::option::Iter; +use dyn_clone::DynClone; use crate::error::MizeResult; use crate::item::{Item, ItemData}; use crate::id::MizeId; use crate::memstore::MemStore; use crate::instance::Instance; +use crate::item::get_raw_from_cbor; -pub trait Store { +dyn_clone::clone_trait_object!(Store); + +pub trait Store: DynClone + Send + Sync { // should have the ability to be multi threaded, if the underlying implementation supports // multithreaded IO operations // for now all refs to a store hold a Mutex but this mutex is part of where the MizeStore trait diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs new file mode 100644 index 0000000..e8072e4 --- /dev/null +++ b/src/core/instance/updater.rs @@ -0,0 +1,35 @@ +use crossbeam::channel::Receiver; +use tracing::{trace, error}; + +use crate::{instance::Instance, item::ItemData}; +use crate::id::MizeId; +use crate:: error::{MizeResult, MizeError, MizeResultTrait}; + + +#[derive(Debug)] +pub enum Operation { + Set(MizeId, ItemData), +} + +pub fn updater_thread(operation_rx : Receiver, mut instance: Instance) -> MizeResult<()> { + for mut operation in operation_rx { + trace!("OPERATION"); + let result = handle_operation(&mut operation, &mut instance); + + if let Err(err) = result { + error!("OPERATION FAILED: {:?}", operation); + } + } + Ok(() ) +} + +fn handle_operation(operation: &mut Operation, instance: &mut Instance) -> MizeResult<()> { + match operation { + Operation::Set(id, value) => { + let item_data: ItemData = value.to_owned(); + let mut item = instance.get(id.clone())?; + item.merge(item_data)?; + } + } + Ok(()) +} diff --git a/src/core/item.rs b/src/core/item.rs index a12f185..f5a9b0b 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -1,4 +1,4 @@ -use log::{trace, debug, info, warn, error}; +use tracing::{debug, error, info, instrument, trace, warn}; use serde::Serialize; use core::fmt; use std::{i128, string}; @@ -8,6 +8,7 @@ use tokio::sync::Mutex; use std::rc::Rc; use colored::Colorize; use std::io; +use tracing::{span, Level}; use crate::error::{MizeError, MizeResult, IntoMizeResult}; use crate::instance::Instance; @@ -256,6 +257,74 @@ pub fn item_data_merge(merge_into: &mut CborValue, other: &CborValue){ } } +#[instrument] +pub fn get_raw_from_cbor<'a>(value: &'a CborValue, path: Vec<&String>) -> MizeResult<&'a [u8]> { + trace!("[ {} ] value: {:?}", "ARG".yellow(), value); + trace!("[ {} ] path: {:?}", "ARG".yellow(), path); + + let mut path_iter = path.clone().into_iter(); + let path_el = match path_iter.nth(0) { + Some(val) => val, + None => { + // our base case + let ret_val = match value { + CborValue::Bytes(vec) => { + trace!("[ {} ] ret value: {:?}", "RET".yellow(), &vec[..]); + return Ok(&vec[..]); + }, + CborValue::Text(string) => { + trace!("[ {} ] ret value: {:?}", "RET".yellow(), string.as_bytes()); + return Ok(string.as_bytes()); + }, + other => { + return Err(MizeError::new() + .msg("path is empty and the value is neither Bytes nor Text")); + }, + }; + }, + }; + + match value { + CborValue::Bytes(vec) => { + trace!("[ {} ] ret value: {:?}", "RET".yellow(), &vec[..]); + return Ok(&vec[..]); + }, + CborValue::Text(string) => { + trace!("[ {} ] ret value: {:?}", "RET".yellow(), string.as_bytes()); + return Ok(string.as_bytes()); + }, + CborValue::Map(map) => { + let mut inner_val: &CborValue = &CborValue::Null; + let mut inner_val_found = false; + + // find the inner_val at path + for (key, val) in map.into_iter() { + if let CborValue::Text(key_text) = key { + if key_text == path_el { + inner_val = val; + inner_val_found = true; + } + } + } + + // if there is no value at that path + if inner_val_found == false { + return Err(MizeError::new() + // jesus christ, this is a pfusch... + .msg(format!("get_raw_from_cbor: Path '{}' not Found", path.clone().into_iter().map(|v| v.to_owned()).collect::>().join("/")))) + } + + //path_iter.next(); + let inner_path = path_iter.collect(); + let ret_value = get_raw_from_cbor(inner_val, inner_path); + trace!("[ {} ] ret value: {:?}", "RET".yellow(), ret_value); + return ret_value; + }, + _ => Err(MizeError::new() + .msg(format!("get_raw_from_cbor: value is not a map, text or bytes ... value: {:?}", value))), + } +} + trait IntoPath { fn into_path(self) -> Vec; } diff --git a/src/core/memstore.rs b/src/core/memstore.rs index d8dfdf0..9e5ccf8 100644 --- a/src/core/memstore.rs +++ b/src/core/memstore.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use std::iter::Map; use std::sync::{Arc, Mutex}; use ciborium::Value as CborValue; -use log::trace; +use tracing::{instrument, trace}; use colored::Colorize; use crate::id::MizeId; @@ -11,6 +11,7 @@ use crate::instance::store::{Store, IdIter}; use crate::error::{MizeError, MizeResult, IntoMizeResult}; use crate::item::{Item, ItemData}; use crate::instance::Instance; +use crate::item::get_raw_from_cbor; #[derive(Clone, Debug)] pub struct MemStore { @@ -49,7 +50,9 @@ impl Store for MemStore { inner.next_id += 1; return Ok(id_string); } + #[instrument(name="MemStore::get_value_raw" skip(self))] fn get_value_raw(&self, id: MizeId) -> MizeResult> { + trace!("[ {} ] id: {:?}", "ARG".yellow(), id); let inner = self.inner.lock()?; let cbor_val = inner.map.get(&id_to_u64(id.clone())?) @@ -131,74 +134,6 @@ fn str_to_u64(id_str: &str) -> MizeResult { .mize_result_msg(format!("Could not parse the store_part of mizeid {} into a u64 for the MemStore", id_str)) } -pub fn get_raw_from_cbor<'a>(value: &'a CborValue, path: Vec<&String>) -> MizeResult<&'a [u8]> { - trace!("[ {} ] get_raw_from_cbor()", "CALL".yellow()); - trace!("[ {} ] value: {:?}", "ARG".yellow(), value); - trace!("[ {} ] path: {:?}", "ARG".yellow(), path); - - let mut path_iter = path.clone().into_iter(); - let path_el = match path_iter.nth(0) { - Some(val) => val, - None => { - // our base case - let ret_val = match value { - CborValue::Bytes(vec) => { - trace!("[ {} ] ret value: {:?}", "RET".yellow(), &vec[..]); - return Ok(&vec[..]); - }, - CborValue::Text(string) => { - trace!("[ {} ] ret value: {:?}", "RET".yellow(), string.as_bytes()); - return Ok(string.as_bytes()); - }, - other => { - return Err(MizeError::new() - .msg("path is empty and the value is neither Bytes nor Text")); - }, - }; - }, - }; - trace!("hoooooooo: {:?}", path_el); - - match value { - CborValue::Bytes(vec) => { - trace!("[ {} ] ret value: {:?}", "RET".yellow(), &vec[..]); - return Ok(&vec[..]); - }, - CborValue::Text(string) => { - trace!("[ {} ] ret value: {:?}", "RET".yellow(), string.as_bytes()); - return Ok(string.as_bytes()); - }, - CborValue::Map(map) => { - let mut inner_val: &CborValue = &CborValue::Null; - let mut inner_val_found = false; - - // find the inner_val at path - for (key, val) in map.into_iter() { - if let CborValue::Text(key_text) = key { - if key_text == path_el { - inner_val = val; - inner_val_found = true; - } - } - } - - // if there is no value at that path - if inner_val_found == false { - return Err(MizeError::new() - // jesus christ, this is a pfusch... - .msg(format!("get_raw_from_cbor: Path '{}' not Found", path.clone().into_iter().map(|v| v.to_owned()).collect::>().join("/")))) - } - - //path_iter.next(); - let inner_path = path_iter.collect(); - trace!("hiiiiii: {:?}", inner_path); - let ret_value = get_raw_from_cbor(inner_val, inner_path); - trace!("[ {} ] ret value: {:?}", "RET".yellow(), ret_value); - return ret_value; - }, - _ => Err(MizeError::new().msg("get_raw_from_cbor: value is not a map, text or bytes")), - } -} diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index a02f24f..080076d 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -4,3 +4,7 @@ use ciborium::Value as CborValue; pub struct MizeMessage { value: CborValue, } + + + + diff --git a/src/platform/os/cli/run.rs b/src/platform/os/cli/run.rs index a2c1d84..dd20ade 100644 --- a/src/platform/os/cli/run.rs +++ b/src/platform/os/cli/run.rs @@ -3,7 +3,7 @@ use std::path::PathBuf; use clap::ArgMatches; use home::home_dir; use std::fs::File; -use log::{trace, debug, info, warn, error}; +use tracing::{trace, debug, info, warn, error}; use nix::sys::signal::{self, Signal}; use mize::error::{MizeResultTrait, IntoMizeResult, MizeError, MizeResult}; diff --git a/src/platform/os/fsstore.rs b/src/platform/os/fsstore.rs index 74b8c2f..9568b2c 100644 --- a/src/platform/os/fsstore.rs +++ b/src/platform/os/fsstore.rs @@ -12,9 +12,10 @@ use crate::{core::instance::store::Store, mize_err}; use crate::error::{IntoMizeResult, MizeResult, MizeError}; use crate::core::item::{ItemData, Item}; use crate::core::id::MizeId; -use crate::core::memstore::get_raw_from_cbor; +use crate::item::get_raw_from_cbor; use crate::instance::Instance; +#[derive(Clone)] pub struct FileStore { path: PathBuf, } diff --git a/src/platform/os/logging.rs b/src/platform/os/logging.rs new file mode 100644 index 0000000..849b3c7 --- /dev/null +++ b/src/platform/os/logging.rs @@ -0,0 +1,257 @@ +use std::path::PathBuf; +use clap::{ArgAction, ArgMatches}; +use clap::{Arg, crate_version, Command}; + +use mize::error::{MizeError, MizeResult}; +use mize::item::ItemData; +use tokio::sync::{Mutex, mpsc}; +use std::sync::Arc; +use std::io::Write; +use colored::Colorize; +use std::env; +use tracing_subscriber::fmt::Subscriber; +use tracing::Level; +use tracing::level_filters::LevelFilter; +use tracing_subscriber::filter::EnvFilter; +use tracing_subscriber::fmt::FormatEvent; +use tracing_subscriber::registry::LookupSpan; +use tracing_subscriber::fmt::FormatFields; +use tracing_subscriber::fmt::FmtContext; +use tracing_subscriber::fmt::format; +use tracing::Event; +use core::fmt; +use tracing_subscriber::fmt::FormattedFields; + +use tracing::{trace, debug, info, warn, error}; + +use crate::DEFAULT_LOG_LEVEL; + +pub 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 = 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 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 + 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")) + ; + + builder + .with_env_filter(my_env_filter) + .without_time() + .event_format(MyFormatter) + //.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); + 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) + } +} diff --git a/src/platform/os/main.rs b/src/platform/os/main.rs index a079f58..58a3e44 100644 --- a/src/platform/os/main.rs +++ b/src/platform/os/main.rs @@ -9,14 +9,16 @@ use mize::error::{MizeError, MizeResult}; use mize::item::ItemData; 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 tracing_subscriber::fmt::Subscriber; +use tracing::Level; +use tracing::level_filters::LevelFilter; +use tracing_subscriber::filter::EnvFilter; +use crate::logging::init_logger; -use log::{trace, debug, info, warn, error}; +use tracing::{trace, debug, info, warn, error}; mod cli { pub mod run; @@ -36,8 +38,10 @@ mod cli { pub use call::call; } +mod logging; + static APPNAME: &str = "mize"; -static DEFAULT_LOG_LEVEL: LevelFilter = LevelFilter::Warn; +static DEFAULT_LOG_LEVEL: LevelFilter = LevelFilter::WARN; fn main() { // welcome to the mize source code @@ -48,10 +52,6 @@ fn main() { init_logger(&cli_matches); - // log a message of every level, to test loging - // test_logger(); - - // match command let result = match cli_matches.subcommand() { // mi daemon @@ -86,179 +86,6 @@ fn main() { } } -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, 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::>() - .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 { diff --git a/src/platform/os/mod.rs b/src/platform/os/mod.rs index 455dfc2..f004fd6 100644 --- a/src/platform/os/mod.rs +++ b/src/platform/os/mod.rs @@ -2,8 +2,9 @@ use ciborium::Value as CborValue; use serde::Deserialize; use std::env::VarError; use std::fs; -use log::{debug, error, warn, trace}; +use tracing::{debug, error, warn, trace}; use clap::ArgMatches; +use std::path::Path; use crate::id::MizeId; use crate::instance::store::Store; @@ -11,12 +12,13 @@ use crate::instance::Instance; use crate::error::{IntoMizeResult, MizeError, MizeResult}; use crate::item::{ItemData, IntoItemData}; use crate::memstore::MemStore; +use crate::platform::os::unix_socket::UnixListener; use self::fsstore::FileStore; mod fsstore; +mod unix_socket; //mod web; -//mod unix-socket; pub fn os_instance_init(instance: &mut Instance) -> MizeResult<()> { @@ -64,6 +66,9 @@ pub fn os_instance_init(instance: &mut Instance) -> MizeResult<()> { if store_path != "" { let file_store = FileStore::new(store_path.as_str())?; instance.migrate_to_store(Box::new(file_store))?; + + let path = Path::new(&store_path).to_owned(); + instance.add_listener(UnixListener::new(path)?)?; } Ok(()) diff --git a/src/platform/os/unix-socket.rs b/src/platform/os/unix-socket.rs deleted file mode 100644 index afbd656..0000000 --- a/src/platform/os/unix-socket.rs +++ /dev/null @@ -1,61 +0,0 @@ - -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) -> 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) -> 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()); - } - } - - } - } -} diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs new file mode 100644 index 0000000..6c8813d --- /dev/null +++ b/src/platform/os/unix_socket.rs @@ -0,0 +1,94 @@ +use std::path::{Path, PathBuf}; +use tokio::net::UnixListener as TokioUnixListener; +use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::io::Interest; +use crossbeam::channel::{Receiver, Sender, unbounded}; + +use crate::instance::peer::{PeerListener, Peer}; +use crate::instance::{self, Instance}; +use crate::error::{MizeError, MizeResult, IntoMizeResult}; +use crate::proto::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() }) + } +} + +impl PeerListener for UnixListener { + fn listen(self, mut instance: Instance) -> MizeResult<()> { + instance.spawn_async("unix listen async", unix_listen(self, instance.clone())); + Ok(()) + } +} + + +async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResult<()> { + let listener = TokioUnixListener::bind(&listener.sock_path) + .mize_result_msg(format!("Could not bind to unix socket at '{}'", listener.sock_path.display()))?; + + loop { + let (unix_sock, addr) = listener.accept().await + .mize_result_msg("Error while accepting Unix sock connection")?; + + let (recieve_tx, recieve_rx) = unbounded::(); + let (send_tx, send_rx) = unbounded::(); + let (unix_read, unix_write) = unix_sock.into_split(); + + let peer = Peer::new(recieve_rx, send_tx); + + instance.add_peer(peer); + instance.spawn_async("incomming", unix_incomming(unix_read, recieve_tx)); + instance.spawn_async("outgoing", unix_outgoing(unix_write, send_rx)); + } +} + + +async fn unix_outgoing(unix_write: OwnedWriteHalf, send_rx: Receiver) -> 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() == tokio::io::ErrorKind::WouldBlock => { + continue; + } + Err(e) => { + return Err(e.into()); + } + } + } + } +} + +async fn unix_incomming(unix_read: OwnedReadHalf, recieve_tx: Sender) -> MizeResult<()> { + loop { + let ready = unix_read.ready(Interest::READABLE).await.mize_result_msg("unix_read_half.ready() failed")?; + + 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() == tokio::io::ErrorKind::WouldBlock => { + continue; + } + Err(e) => { + return Err(e.into()); + } + } + + } + } +}