use async_trait::async_trait; use colored::Colorize; use core::fmt; use std::time::Duration; use flume::{bounded, unbounded, Sender}; use interner::shared::{StringPool, VecStringPool}; use std::any::Any; use std::collections::HashMap; use std::ops::{Deref, DerefMut}; use std::sync::{Arc, Mutex, RwLock}; use std::{thread, vec}; use crate::set_global_mize; use crate::config::{ConfigOpt, ConfigOptId, MizeFun}; use crate::error::{MizeError, MizeResult}; use crate::id::{IntoMizeId, MizeId, Namespace}; use crate::instance::store::Store; use crate::instance::subscription::Subscription; use crate::instance::updater::Operation; use crate::instance::updater::updater_thread; use crate::item::{Item, ItemData}; use crate::memstore::MemStore; use crate::mize_err; use crate::proto::MizeMessage; use crate::error::IntoMizeResult; use crate::prelude::*; use self::connection::{ConnListener, Connection}; use self::updater::handle_operation; #[cfg(feature = "async")] use tokio::runtime::Handle; #[cfg(feature = "async")] use tokio::runtime::Runtime; use core::future::Future; use std::thread::JoinHandle; pub mod connection; pub mod msg_thread; pub mod store; pub mod subscription; pub mod updater; #[cfg(test)] mod tests; static MSG_CHANNEL_SIZE: usize = 200; static BUILD_TIME_CONFIG: &str = std::env!("MIZE_BUILD_CONFIG"); /// The Instance type is the heart of the mize system #[derive(Clone)] pub struct Mize { // a bit a lot of Mutexes isn't it??? pub(crate) store: Arc>>, connections: Arc>>, next_con_id: Arc>, subs: Arc>>>, pub(crate) id_pool: Arc>, pub(crate) namespace_pool: Arc>, pub(crate) listeners: Arc< Mutex MizeResult<()> + Send + Sync>>>, >, pub parts: Arc>>>>, //pub(crate) platform: Arc>>, // should we do that instead of all the #cfg(faturs=targt-smth) in lib.rs part_names: Arc>>, // exists, so that we know the ordering in which parts were loaded pub(crate) config_opts: Arc>>, // the namespace the instance operates in pub(crate) namespace: Arc>, // the namespace of the instance itself // TODO: set to a random uuid pub(crate) self_namespace: Arc>, pub(crate) op_tx: Sender, threads: Arc>>)>>>, next_thread_id: Arc>, give_msg_wait: Arc>>>>, create_msg_wait: Arc>>>, pub cur_eval_part: Arc>>, pub native_values: Arc>>>, pub next_native_value_id: Arc>, pub fun_values: Arc>>, pub next_fun_values_id: Arc>, // generate mode: when true, part_smth functions record metadata instead of executing pub generate_mode: Arc>, // recorded metadata per part (keyed by part id) while in generate mode pub part_metadata: Arc>>, #[cfg(feature = "async")] pub runtime: Arc>, } /// Metadata recorded about a part while in generate mode #[derive(Clone, Debug, Default)] pub struct PartMetadata { pub part_name: String, pub opts: Vec, pub methods: Vec, pub effects: Vec, pub threads: Vec, pub natives: Vec, } #[derive(Clone, Debug)] pub struct OptMetadata { pub name: String, pub id: MizeId, pub default_val: Option, } #[derive(Clone, Debug)] pub struct MethodMetadata { pub name: String, pub id: MizeId, pub kind: String, // "method", "effect", "thread" /// Raw signature string, e.g. `"String -> ()"` or `"Command, Func -> ()"`. pub signature: String, /// Parsed argument type names (left of `->`). pub arg_types: Vec, /// Parsed return type name (right of `->`). Empty means unspecified / ItemData. pub return_type: String, } /// Parse a method signature like `"String -> ()"` or `"Command, Func -> ()"`. /// Returns `(arg_types, return_type)` where `return_type` is empty if no `->` was given. pub fn parse_method_signature(signature: &str) -> (Vec, String) { let signature = signature.trim(); if signature.is_empty() { return (Vec::new(), String::new()); } let (args_part, ret_part) = match signature.split_once("->") { Some((a, r)) => (a.trim(), r.trim().to_string()), None => (signature, String::new()), }; let arg_types = if args_part.is_empty() { Vec::new() } else { args_part .split(',') .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) .collect() }; (arg_types, ret_part) } #[derive(Clone, Debug)] pub struct NativeMetadata { pub name: String, pub type_name: String, pub associated: Vec, pub methods: Vec, /// Signature strings parallel to `associated`. pub associated_sigs: Vec, /// Signature strings parallel to `methods`. pub method_sigs: Vec, } pub struct InstanceRef { inner: Arc>, } pub struct PartRef { id: MizeId, } impl PartRef { pub fn into_any(self) -> Box { Box::new(self) } } impl MizePart for PartRef { fn id(&self) -> MizeResult { Ok(self.id.clone()) } } pub trait PartToNative { fn from_mize_id(id: MizeId) -> Self; } pub struct InstanceAsync { inner: Arc>, } pub trait MizePart { fn id(&self) -> MizeResult; } impl Mize { pub fn empty() -> MizeResult { let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build()?; let guard = runtime.enter(); std::mem::forget(guard); let id_pool = Arc::new(Mutex::new(VecStringPool::default())); let namespace_pool_raw = StringPool::default(); let connections = Arc::new(Mutex::new(Vec::new())); let subs = Arc::new(Mutex::new(HashMap::new())); let (op_tx, op_rx) = unbounded(); let give_msg_wait = Arc::new(Mutex::new(HashMap::new())); let create_msg_wait = Arc::new(Mutex::new(None)); let namespace = Arc::new(Mutex::new(Namespace( namespace_pool_raw.get("mize.default.namespace"), ))); let self_namespace = Arc::new(Mutex::new(Namespace( namespace_pool_raw.get("mize.default.namespace"), ))); let mut instance = Mize { store: Arc::new(Mutex::new(Box::new(MemStore::new()))), parts: Arc::new(Mutex::new(HashMap::new())), part_names: Arc::new(Mutex::new(Vec::new())), config_opts: Arc::new(Mutex::new(HashMap::new())), listeners: Arc::new(Mutex::new(HashMap::new())), connections, subs, id_pool, namespace, self_namespace, op_tx, namespace_pool: Arc::new(Mutex::new(namespace_pool_raw)), threads: Arc::new(Mutex::new(Vec::new())), next_thread_id: Arc::new(Mutex::new(0)), next_con_id: Arc::new(Mutex::new(1)), give_msg_wait, create_msg_wait, cur_eval_part: Arc::new(Mutex::new(None)), native_values: Arc::new(Mutex::new(HashMap::new())), next_native_value_id: Arc::new(Mutex::new(0)), fun_values: Arc::new(Mutex::new(HashMap::new())), next_fun_values_id: Arc::new(Mutex::new(0)), generate_mode: Arc::new(RwLock::new(false)), part_metadata: Arc::new(Mutex::new(HashMap::new())), #[cfg(feature = "async")] runtime: Arc::new(Mutex::new(runtime)), }; #[cfg(feature = "target-os")] { let instance_clone = instance.clone(); let op_rx_clone = op_rx.clone(); let closure = move || updater_thread(op_rx_clone, &instance_clone); instance.spawn_background("updater_thread", closure)?; let instance_clone_two = instance.clone(); let closure_two = move || updater_thread(op_rx, &instance_clone_two); instance.spawn_background("updater_thread", closure_two)?; } // set up async update "threads" when using wasm #[cfg(feature = "target-wasm ")] { let instance_clone = instance.clone(); let op_rx_clone = op_rx.clone(); wasm_bindgen_futures::spawn_local(updater_thread_async(op_rx_clone, instance_clone)); let instance_clone_two = instance.clone(); wasm_bindgen_futures::spawn_local(updater_thread_async(op_rx, instance_clone_two)); } // will move the msg stuff into it's own thread // like this it can deadlock.... if a msg waits on an operation to complete //let msg_instance_clone = instance.clone(); //let msg_closure = move || msg_thread(op_rx, instance_clone); //instance.spawn("msg_thread", closure)?; #[cfg(feature = "target-wasm ")] console_log!("before loading build time config"); // load the config from build time let config = ItemData::from_json(BUILD_TIME_CONFIG.to_owned())?; instance.set_blocking("0", config); #[cfg(feature = "target-wasm ")] console_log!("after loading build time config"); return Ok(instance); } pub fn new() -> MizeResult { trace!("[ {} ] Instance::new()", "CALL".yellow()); let mut instance = Mize::empty()?; instance.init()?; debug!( "instance inited with config: {}", instance.get("0/config")?.as_data_full()? ); return Ok(instance); } pub fn id_new_dots(&mut self, name: &str) -> MizeResult { let parts = name.split(".").map(|v| v.to_owned()).collect::>(); self.new_id(parts) } pub fn get_part(&mut self, name: &str) -> MizeResult { let id = self.id_new_dots(name)?; Ok(PartRef { id }) } pub fn has_part(&mut self, name: &str) -> MizeResult { let id = self.id_new_dots(name)?; Ok(self.config_opts.lock().unwrap().contains_key(&id)) } pub fn add_listener( &mut self, name: &str, handler: Box MizeResult<()> + Send + Sync>, ) -> MizeResult<()> { let name = name.to_owned(); let mut listeners = self.listeners.lock().unwrap(); listeners.insert(name, Arc::new(handler)); Ok(()) } pub fn init(&mut self) -> MizeResult<()> { // set global mize set_global_mize(self.clone())?; // platform specific init code crate::platform::init(self)?; // end of platform specific init code // load the parts, as specified in the parts config match self .get_config("parts") { Ok(parts_to_load) => { for part_name in parts_to_load.cbor().as_array().ok_or("config parts was not an array")? { let part_name = part_name.as_text().ok_or("part name was not a string")?; if part_name.trim().is_empty() || self.has_part(part_name)? { continue; } self.load_part(part_name)?; } } Err(_err) => { // no parts to load } } debug!("MIZE INIT DONE"); Ok(()) } pub fn with_config(config: ItemData) -> MizeResult { trace!("[ {} ] Instance::with_config()", "CALL".yellow()); trace!("config: {}", config); let mut mize = Mize::empty()?; let mut opts = mize.config_opts.lock().unwrap(); let mut config = config.clone(); for mut path in config.get_paths_recursive()? { if path.starts_with("/") { path.remove(0); } let path_split = path .split("/") .map(|s| s.to_owned()) .collect::>(); let path_string = path_split.join("."); if let Ok(val) = config.get_path(path_split) { let opt = ConfigOpt { name: path.clone(), val: Some(val), fun: None, deps: Vec::new(), }; opts.insert(mize.clone().id_new_dots(&path_string)?, opt); } } drop(opts); mize.init()?; Ok(mize) } pub fn migrate_to_store(&self, new_store: Box) -> MizeResult<()> { info!("MIGRATING"); let mut old_store = self.store.lock()?; let id = self.id_from_string("0".to_owned())?; let inst_data = old_store.get_value_data_full(id.clone())?; new_store.set(id, inst_data.to_owned())?; for id in old_store.id_iter()? { let data = old_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())?; } *old_store = new_store; Ok(()) } pub fn new_item(&self) -> MizeResult { if self.get_namespace()? != self.get_self_namespace()? { // need to send create msg and wait for it let connection = self.get_connection_by_ns(self.get_namespace()?)?; let msg = MizeMessage::new_create(connection.id); connection.send(msg)?; let (tx, rx) = bounded::(1); let mut msg_wait_inner = self.create_msg_wait.lock()?; *msg_wait_inner = Some(tx); drop(msg_wait_inner); let id = rx.recv()?; return Ok(Item::new(id, self)); } let store_inner = self.store.lock()?; let id = self.id_from_string(store_inner.new_id()?)?; return Ok(Item::new(id, self)); } pub fn get(&self, id: I) -> MizeResult { let id = id.to_mize_id(self)?; return Ok(Item::new(id, &self)); } pub fn set>(&self, id: I, value: V) -> MizeResult<()> { let id = id.to_mize_id(self)?; self.op_tx.send(Operation::Set(id, value.into(), None)); Ok(()) } pub fn set_blocking>( &self, id: I, value: V, ) -> MizeResult<()> { handle_operation( &mut Operation::Set(id.to_mize_id(self)?, value.into(), None), self, )?; Ok(()) } pub fn sub(&self, id: I, sub: Subscription) -> MizeResult<()> { let mut subs_inner = self.subs.lock()?; let id = id.to_mize_id(self)?; match subs_inner.get_mut(&id) { Some(vec) => { vec.push(sub); } None => { subs_inner.insert(id.clone(), vec![sub]); } } // if we are not the owner of this item, send a sub msg to them if !self.we_are_namespace()? { let con = self.get_connection_by_ns(id.namespace())?; let msg = MizeMessage::new_sub(id, con.id); con.send(msg)?; } Ok(()) } pub fn new_id(&self, value: T) -> MizeResult { value.to_mize_id(self) } pub fn id_from_string(&self, string: String) -> MizeResult { let vec_string: Vec = string .split("/") .map(|v| v.to_owned()) .filter(|el| el.as_str() != "") .collect(); self.id_from_vec_string(vec_string) } pub fn id_from_vec_string(&self, mut vec_string: Vec) -> MizeResult { let id_pool_inner = self.id_pool.lock()?; let namespace_inner = self.namespace.lock()?; let first_el = vec_string .first_mut() .ok_or(mize_err!("MizeId was empty"))?; let id = if first_el.contains(":") { // first el is a namespace + store_part let new_first_el = first_el.clone(); let vec: Vec<&str> = new_first_el.split(":").collect(); let ns_part = vec .iter() .nth(0) .ok_or(mize_err!("should really not happen"))?; let store_part = vec.iter().nth(1).ok_or(mize_err!( "mizeid was like 'namespace:/hi', why are you doing that" ))?; *first_el = store_part.to_owned().to_owned(); MizeId { path: id_pool_inner.get(vec_string), namespace: self.namespace_from_string(ns_part.to_owned().to_owned())?, } } else { MizeId { path: id_pool_inner.get(vec_string), namespace: namespace_inner.clone(), } }; trace!("new MizeId made: {:?}", id); Ok(id) } pub fn load_part(&mut self, name: &str) -> MizeResult<()> { tracing::debug!("loading part: {name}"); crate::platform::load_part(self, name, None) } pub fn fetch_part(&mut self, name: &str) -> MizeResult { // platform specific way to fetch a part crate::platform::fetch_part(self, name) } pub fn load_part_at(&mut self, name: &str, path: String) -> MizeResult<()> { // platform specific init code crate::platform::load_part(self, name, Some(path)) } // ===== generate mode accessors ===== /// Enable or disable generate mode. When enabled, part_smth functions record /// metadata (methods, opts, natives) instead of performing their normal operations. pub fn set_generate_mode(&self, enabled: bool) -> MizeResult<()> { let mut mode = self.generate_mode.write().map_err(|_| mize_err!("generate_mode lock poisoned"))?; *mode = enabled; Ok(()) } /// Check if generate mode is currently enabled. pub fn is_generate_mode(&self) -> MizeResult { let mode = self.generate_mode.read().map_err(|_| mize_err!("generate_mode lock poisoned"))?; Ok(*mode) } /// Record an opt declaration while in generate mode. pub fn record_opt(&self, name: &str, id: MizeId, default_val: Option) -> MizeResult<()> { let cur_part = self.cur_eval_part.lock()?.clone() .ok_or_else(|| mize_err!("no cur_eval_part set"))?; let mut metadata_map = self.part_metadata.lock()?; let metadata = metadata_map.entry(cur_part.clone()).or_insert_with(|| PartMetadata { part_name: cur_part.dot(), ..Default::default() }); metadata.opts.push(OptMetadata { name: name.to_string(), id, default_val, }); Ok(()) } /// Record a method/effect/thread declaration while in generate mode. pub fn record_method(&self, name: &str, id: MizeId, kind: &str, signature: &str) -> MizeResult<()> { let cur_part = self.cur_eval_part.lock()?.clone() .ok_or_else(|| mize_err!("no cur_eval_part set"))?; let mut metadata_map = self.part_metadata.lock()?; let metadata = metadata_map.entry(cur_part.clone()).or_insert_with(|| PartMetadata { part_name: cur_part.dot(), ..Default::default() }); let (arg_types, return_type) = parse_method_signature(signature); let m = MethodMetadata { name: name.to_string(), id, kind: kind.to_string(), signature: signature.to_string(), arg_types, return_type, }; match kind { "effect" => metadata.effects.push(m), "thread" => metadata.threads.push(m), _ => metadata.methods.push(m), } Ok(()) } /// Record a native type inclusion while in generate mode. pub fn record_native( &self, name: &str, type_name: &str, associated: Vec, methods: Vec, associated_sigs: Vec, method_sigs: Vec, ) -> MizeResult<()> { let cur_part = self.cur_eval_part.lock()?.clone() .ok_or_else(|| mize_err!("no cur_eval_part set"))?; let mut metadata_map = self.part_metadata.lock()?; let metadata = metadata_map.entry(cur_part.clone()).or_insert_with(|| PartMetadata { part_name: cur_part.dot(), ..Default::default() }); metadata.natives.push(NativeMetadata { name: name.to_string(), type_name: type_name.to_string(), associated, methods, associated_sigs, method_sigs, }); Ok(()) } /// Get recorded metadata for a specific part. pub fn get_metadata(&self, part_id: &MizeId) -> MizeResult> { let metadata_map = self.part_metadata.lock()?; Ok(metadata_map.get(part_id).cloned()) } /// Get all recorded metadata (for all parts). pub fn get_all_metadata(&self) -> MizeResult> { let metadata_map = self.part_metadata.lock()?; Ok(metadata_map.values().cloned().collect()) } /// Clear all recorded metadata. pub fn clear_metadata(&self) -> MizeResult<()> { let mut metadata_map = self.part_metadata.lock()?; metadata_map.clear(); Ok(()) } 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 set_namespace(&self, ns: Namespace) -> MizeResult<()> { let mut namespace_inner = self.namespace.lock()?; *namespace_inner = ns; Ok(()) } pub fn get_namespace(&self) -> MizeResult { let namespace_inner = self.namespace.lock()?; return Ok(namespace_inner.clone()); } pub fn get_self_namespace(&self) -> MizeResult { let self_namespace_inner = self.self_namespace.lock()?; return Ok(self_namespace_inner.clone()); } pub fn we_are_namespace(&self) -> MizeResult { Ok(self.get_namespace()? == self.get_self_namespace()?) } pub fn add_conn_listener(&mut self, listener: T) -> MizeResult<()> { let instance_clone = self.clone(); self.spawn_background("some_listener", move || listener.listen(instance_clone)); Ok(()) } pub fn new_connection(&self, tx: Sender) -> MizeResult { let mut conn_inner = self.connections.lock()?; let mut next_con_id = self.next_con_id.lock()?; let old_next_con_id = *next_con_id; let connection = Connection { id: next_con_id.to_owned(), tx, ns: None, }; conn_inner.push(connection); *next_con_id += 1; Ok(old_next_con_id) } pub fn new_connection_join_namespace(&self, tx: Sender) -> MizeResult { let conn_id = self.new_connection(tx)?; let ns_of_peer_str = self .get(format!( "inst/con_by_id/{}/peer/self/config/namespace", conn_id ))? .value_string()?; let ns_of_peer = self.namespace_from_string(ns_of_peer_str)?; self.connection_set_namespace(conn_id, ns_of_peer.clone()); self.set_namespace(ns_of_peer); Ok(conn_id) } pub fn connection_set_namespace(&self, conn_id: u64, namespace: Namespace) -> MizeResult<()> { let mut connection = self.get_connection(conn_id)?; connection.ns = Some(namespace); self.set_connection(conn_id, connection); Ok(()) } pub fn got_msg(&self, msg: MizeMessage) -> MizeResult<()> { Ok(self.op_tx.send(Operation::Msg(msg))?) } pub fn report_err(&self, err: MizeError) { #[cfg(feature = "target-js")] unsafe { use wasm_bindgen::prelude::wasm_bindgen; #[wasm_bindgen] extern "C" { #[wasm_bindgen(js_namespace = console)] fn log(s: &str); } macro_rules! console_log { // Note that this is using the `log` function imported above during // `bare_bones` ($($t:tt)*) => ({ log(&format_args!($($t)*).to_string())}) } console_log!("reported_err: {:?}", err); } err.log(); } fn set_connection(&self, conn_id: u64, new_connection: Connection) -> MizeResult<()> { let mut conn_inner = self.connections.lock()?; for connection in conn_inner.iter_mut() { if connection.id == conn_id { *connection = new_connection; return Ok(()); } } return Err(mize_err!( "Connection with id {} not known to instance", conn_id )); } pub fn get_connection(&self, conn_id: u64) -> MizeResult { let conn_inner = self.connections.lock()?; for connection in conn_inner.iter() { if connection.id == conn_id { return Ok(connection.clone()); } } return Err(mize_err!( "Connection with id {} not known to instance", conn_id )); } pub fn get_connection_by_ns(&self, ns: Namespace) -> MizeResult { let conn_inner = self.connections.lock()?; for connection in conn_inner.iter() { if connection.ns == Some(ns.clone()) { return Ok(connection.clone()); } } return Err(mize_err!( "Connection with namespace {} not known to instance", ns.as_string() )); } pub fn spawn_and_wait( &mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static, ) -> MizeResult<()> { let mize_clone = self.clone(); let mut threads_inner = self.threads.lock()?; let mut next_thread_id = self.next_thread_id.lock()?; let my_thread_id_no_mutex_guard = *next_thread_id; let _thread_mutex = self.threads.clone(); let name_to_move = name.to_owned(); let to_spawn = move || -> MizeResult<()> { debug!("spawning thread: {}", name_to_move); let _my_thread_id = my_thread_id_no_mutex_guard; if let Err(err) = func() { mize_clone.report_err(err); } //let mut threads_inner = thread_mutex.lock()?; /* *threads_inner = threads_inner .clone() .into_iter() .filter(|el| match el { (my_thread_id, _) => false, (_, _) => true, }) .collect(); */ debug!("thread '{}' stopped", name_to_move); Ok(()) }; *next_thread_id += 1; #[cfg(feature = "target-os")] let handle = thread::spawn(move || to_spawn()); #[cfg(feature = "target-os")] threads_inner.push((*next_thread_id, name.to_owned(), Some(handle))); #[cfg(feature = "target-wasm ")] { //console_log!("in instance::spawn with wasm target") } //NOT WELL SUPPORTED //crate::platform::wasm::wasm_spawn(to_spawn)?; Ok(()) } pub fn spawn_background( &mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static, ) -> MizeResult<()> { let mut threads_inner = self.threads.lock()?; let mut next_thread_id = self.next_thread_id.lock()?; let tokio_handle = tokio::runtime::Handle::current(); let my_thread_id_no_mutex_guard = *next_thread_id; let _thread_mutex = self.threads.clone(); let name_to_move = name.to_owned(); let to_spawn = move || -> () { debug!("spawning thread: {}", name_to_move); let guard = tokio_handle.enter(); std::mem::forget(guard); let _my_thread_id = my_thread_id_no_mutex_guard; let res = func(); if let Err(e) = res { e.log(); } //let mut threads_inner = thread_mutex.lock()?; /* *threads_inner = threads_inner .clone() .into_iter() .filter(|el| match el { (my_thread_id, _) => false, (_, _) => true, }) .collect(); */ debug!("thread '{}' stopped", name_to_move); }; *next_thread_id += 1; #[cfg(feature = "target-os")] thread::spawn(move || to_spawn()); threads_inner.push((*next_thread_id, name.to_owned(), None)); #[cfg(feature = "target-js")] { //console_log!("in instance::spawn with wasm target") } //NOT WELL SUPPORTED //crate::platform::wasm::wasm_spawn(to_spawn)?; Ok(()) } pub fn give_msg_wait(&self, id: MizeId) -> MizeResult { let mut give_msg_wait_inner = self.give_msg_wait.lock()?; let (tx, rx) = bounded::(1); let vec = match give_msg_wait_inner.get_mut(&id) { Some(vec) => vec, None => { give_msg_wait_inner.insert(id.clone(), Vec::new()); give_msg_wait_inner.get_mut(&id).unwrap() } }; vec.push(tx); // so that another thread can also give_msg_wait(), while we wait in the recv() of rx drop(vec); drop(give_msg_wait_inner); let data = rx.recv()?; return Ok(data); } #[cfg(feature = "async")] pub fn spawn_async + Send + 'static>( &mut self, name: &str, func: F, ) -> MizeResult<()> { let mut threads_inner = self.threads.lock()?; let mut next_thread_id = self.next_thread_id.lock()?; threads_inner.push((*next_thread_id, name.to_owned(), None)); let runtime = tokio::runtime::Handle::current(); runtime.spawn(func); *next_thread_id += 1; Ok(()) } #[cfg(feature = "async")] pub fn async_get_handle(&self) -> Handle { tokio::runtime::Handle::current() } #[cfg(feature = "async")] pub fn spawn_async_blocking< F: Future + Send + Sync + 'static, >( &mut self, name: &str, func: F, ) -> F::Output { let mut threads_inner = self .threads .lock() .expect("mutex lock failed in spawn_async_blocking"); let mut next_thread_id = self .next_thread_id .lock() .expect("mutex lock failed in spawn_async_blocking"); threads_inner.push((*next_thread_id, name.to_owned(), None)); let runtime = tokio::runtime::Handle::current(); *next_thread_id += 1; drop(threads_inner); drop(next_thread_id); let result = runtime.block_on(func); return result; } pub fn wait(&self) { info!("Instance main thread waiting"); loop { thread::sleep(Duration::from_secs(100000)) } } pub fn report_error(err: MizeError) { err.log(); } pub fn get_config(&mut self, name: &str) -> MizeResult { let mut config_opts = self.config_opts.lock().unwrap(); let mut id; let mut split = name.split("."); if split.nth(0).unwrap() == "self" { let cur_part = self.cur_eval_part.lock()?.clone().ok_or(mize_err!("get_config called but no curr eval part set"))?; id = cur_part.clone(); for el in split { id = id.join(el)?; } } else { id = self.clone().id_new_dots(name)?; } let mut opt = config_opts .get_mut(&id) .ok_or_else(|| mize_err!("Config opt for {id} not in config_opts"))?; if let Some(val) = opt.val.clone() { return Ok(val); } // in case it is not cached run the fun if it exists if opt.fun.is_some() { let fun = opt.fun.take().unwrap(); drop(config_opts); let res = fun()?; let mut config_opts = self.config_opts.lock().unwrap(); let mut opt = config_opts.get_mut(&id).unwrap(); opt.fun = Some(fun); opt.val = Some(res.clone()); return Ok(res); } Err(mize_err!("ConfigOpt {id} has no fun and no cached value")) } pub fn dyn_call(&mut self, part_id: MizeId, method: &str, args: ItemData) -> MizeResult { debug!("dyn call {}/{} with args: {}", part_id, method, args); let mut fun_values = self.fun_values.lock()?; let id = part_id.clone().join(method)?; //let part = opts.get_mut(&id).ok_or(mize_err!("part {} is not loaded", part_id.dot()))?; let fun = fun_values.remove(&id).ok_or(mize_err!("fun value for {id} not found"))?; drop(fun_values); let calling_part = self.clone().cur_eval_part()?; self.clone().set_cur_eval_part(Some(part_id)); println!("before call"); let res = fun(args); println!("after call"); self.clone().set_cur_eval_part(calling_part); let mut fun_values = self.fun_values.lock()?; fun_values.insert(id, fun); return res; } pub fn cur_eval_part(&mut self) -> MizeResult> { Ok(self.cur_eval_part.lock()?.clone()) } pub fn set_cur_eval_part(&mut self, part: Option) -> MizeResult<()> { let mut guard = self.cur_eval_part.lock()?; *guard = part; Ok(()) } pub fn get_part_gen( &mut self, name: &str, ) -> MizeResult { let dyn_part = self.get_part(name)?; let concrete_part = dyn_part .into_any() .downcast::() .map_err(|_| mize_err!("failed to downcast"))?; Ok(*concrete_part) } pub fn id_path(&self, id: &MizeId) -> MizeResult { Ok(id.path().join("/")) } } impl fmt::Debug for Mize { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "Mize Instance with subs: {:?}", self.subs,) } } impl Default for Mize { fn default() -> Self { Self::empty().unwrap() } }