diff --git a/flake.nix b/flake.nix index 4fa7adc..0817d9b 100644 --- a/flake.nix +++ b/flake.nix @@ -90,6 +90,7 @@ in { (fenix.packages.${system}.combine [ wasmToolchain osToolchain ]) lldb gdb ]; + shellHook = '' export MIZE_CONFIG_FILE=${self}/test-config.toml ''; diff --git a/src/core/id.rs b/src/core/id.rs index 5597bf9..6da91df 100644 --- a/src/core/id.rs +++ b/src/core/id.rs @@ -4,7 +4,7 @@ use interner::{shared::{SharedPool, SharedString, SharedVecString}, Pooled}; use std::collections::hash_map::RandomState; use std::path::Path; -use crate::instance::{store::Store, Instance}; +use crate::{instance::{store::Store, Instance}, mize_err}; use crate::error::{MizeResult, MizeError, MizeResultTrait}; #[derive(Debug, Clone, Hash, Eq, PartialEq)] @@ -35,6 +35,13 @@ impl MizeId { .expect("an empty MizeId found, that should absolutely not be possible!!!!").as_str() } + pub fn nth_part(&self, n: usize) -> MizeResult<&str> { + match self.path.iter().nth(n) { + Some(val) => Ok(val.as_str()), + None => Err(mize_err!("id does not have {} parts", n)), + } + } + pub fn path(&self) -> SharedVecString { self.path.clone() } @@ -75,6 +82,12 @@ impl IntoMizeId for Vec { instance.id_from_vec_string(self) } } +impl IntoMizeId for Vec<&String> { + fn to_mize_id(self, instance: &Instance) -> MizeResult { + let owned = self.into_iter().map(|s| s.to_owned()).collect::>(); + instance.id_from_vec_string(owned) + } +} impl IntoMizeId for &[&str] { fn to_mize_id(self, instance: &Instance) -> MizeResult { let mut vec: Vec = Vec::new(); diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs index e9a013f..f429958 100644 --- a/src/core/instance/connection.rs +++ b/src/core/instance/connection.rs @@ -1,16 +1,21 @@ use crossbeam::channel::{Sender, Receiver}; use tracing::{trace, debug, info, warn, error}; +use crate::id::Namespace; +use crate::mize_err; use crate::proto::MizeMessage; use crate::error::{MizeError, MizeResult, IntoMizeResult}; use super::Instance; +use crate::item::{get_raw_from_cbor, Item}; +// only created with Instance::new_connection() #[derive(Clone)] pub struct Connection { pub rx: Receiver, pub tx: Sender, pub id: u64, + pub ns: Option, } pub trait ConnListener : Send + Sync { @@ -23,5 +28,33 @@ impl Connection { } } +pub fn value_raw_con_by_id(item: &mut Item) -> MizeResult> { + if item.id().nth_part(3)? == "peer" { + return value_raw_from_peer(item); + } else { + return Err(mize_err!("id with path: '{}' is not supported", item.id())); + } +} +pub fn value_raw_from_peer(item: &mut Item) -> MizeResult> { + let id = item.id(); + let conn_id_str = id.nth_part(2)?; + let conn_id : u64 = conn_id_str.parse()?; + let tmp_path = id.path(); + let mut new_id_str = tmp_path.into_iter(); + let skipped = new_id_str.skip(4); + let new_id = item.instance.new_id(skipped.collect::>())?; + + let connection = item.instance.get_connection(conn_id)?; + + let msg = MizeMessage::new_get(new_id, connection.id); + + connection.tx.send(msg)?; + + let data = item.instance.give_msg_wait(item.id())?; + + let raw_data = get_raw_from_cbor(data.cbor(), vec![])?; + return Ok(raw_data.to_owned()); + +} diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 8beb82c..7e7a995 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -56,6 +56,7 @@ pub struct Instance { threads: Vec, #[cfg(feature = "async")] pub runtime: Arc>, + give_msg_wait: Arc, Receiver)>>>, } pub struct InstanceRef { @@ -76,6 +77,7 @@ impl Instance { let subs = Arc::new(Mutex::new(HashMap::new())); let namespace = Namespace ( namespace_pool_raw.get("mize.default.namespace") ); let (op_tx, op_rx) = channel::unbounded(); + let give_msg_wait = Arc::new(Mutex::new(HashMap::new())); let mut instance = Instance { store: Box::new(store), @@ -85,6 +87,7 @@ impl Instance { context: vec![], threads: vec![], next_con_id: Arc::new(Mutex::new(1)), + give_msg_wait, #[cfg(feature = "async")] runtime: Arc::new(Mutex::new(Runtime::new().mize_result_msg("Could not create async runtime")?)), @@ -214,13 +217,34 @@ impl Instance { let mut conn_inner = self.connections.lock()?; let mut next_con_id = self.next_con_id.lock()?; - let connection = Connection { id: next_con_id.to_owned(), rx, tx}; + let connection = Connection { id: next_con_id.to_owned(), rx, tx, ns: None}; conn_inner.push(connection); *next_con_id += 1; Ok(next_con_id.to_owned()) } - pub fn get_connection(&mut self, conn_id: u64) -> MizeResult { + pub fn connection_set_namespace(&mut 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(()) + } + + fn set_connection(&mut 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)); + Ok(()) + } + + pub fn get_connection(&self, conn_id: u64) -> MizeResult { let mut conn_inner = self.connections.lock()?; for connection in conn_inner.iter() { @@ -238,6 +262,29 @@ impl Instance { Ok(()) } + pub fn give_msg_wait(&self, id: MizeId) -> MizeResult { + + let mut give_msg_wait_inner = self.give_msg_wait.lock()?; + + let rx = match give_msg_wait_inner.get(&id) { + // if this id is already in the process of being requested, just use the rx already there + Some((tx, rx)) => rx.to_owned(), + + None => { + let (tx, rx) = crossbeam::channel::bounded::(1); + give_msg_wait_inner.insert(id, (tx, rx.clone())); + rx + }, + }; + + // so that another thread can also give_msg_wait(), while we wait in the recv() of rx + drop(give_msg_wait_inner); + + let data = rx.recv()?; + + return Ok(data); + } + #[cfg(feature = "async")] pub fn spawn_async + Send + Sync + 'static>(&mut self, name: &str, func: F) { self.threads.push(name.to_owned()); diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 195d5ae..6c81eeb 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -47,7 +47,7 @@ fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> match msg.cmd()? { MessageCmd::Get => { let id = msg.id(instance)?; - let mut connection = instance.get_connection(msg.conn_id)?; + let mut connection = instance.get_connection(msg.conn_id)?.clone(); let item = instance.get(id.clone())?; let msg = MizeMessage::new_give(id, item.as_data_full()?, msg.conn_id); connection.send(msg)?; diff --git a/src/core/item.rs b/src/core/item.rs index f5a9b0b..7da9935 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -15,10 +15,11 @@ use crate::instance::Instance; use crate::instance::store::Store; use crate::id::MizeId; use ciborium::Value as CborValue; +use crate::instance::connection::value_raw_con_by_id; // a item always has to do with a Instance, which takes care of how it is updated -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct Item<'a> { id: MizeId, pub instance: &'a Instance @@ -39,6 +40,18 @@ impl Item<'_> { } pub fn value_raw(&self) -> MizeResult> { + let id = self.id(); + + // some paths like inst/con_by_id/x/ are not stored in the store, but gotten from the instance + // itself (the data for that lifes in memory in the instance struct) + // so here we set handlers for certain paths + if id.store_part() == "inst" { + match id.nth_part(1)? { + "con_by_id" => { return value_raw_con_by_id(&mut self.clone()); }, + _ => {}, + } + } + // this will call from the instance which gets the value from the store self.instance.store.get_value_raw(self.id()) } diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index 4f4146b..8bb29d2 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -23,9 +23,20 @@ impl MizeMessage { MizeMessage { value, conn_id } } - pub fn new_give(id: MizeId, data: ItemData, conn_id: u64) -> MizeMessage { - let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_GIVE.into())); + pub fn new_get(id: MizeId, conn_id: u64) -> MizeMessage { let id_path = id.path().into_iter().map(|string| CborValue::Text(string.to_owned())).collect(); + + let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_GET.into())); + let id = (CborValue::Integer(MSG_ID.into()), CborValue::Array(id_path)); + let value = CborValue::Map(vec![cmd, id]); + + MizeMessage::new(value, conn_id) + } + + pub fn new_give(id: MizeId, data: ItemData, conn_id: u64) -> MizeMessage { + let id_path = id.path().into_iter().map(|string| CborValue::Text(string.to_owned())).collect(); + + let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_GIVE.into())); let id = (CborValue::Integer(MSG_ID.into()), CborValue::Array(id_path)); let data = (CborValue::Integer(MSG_DATA.into()), data.cbor().to_owned()); let value = CborValue::Map(vec![cmd, id, data]); diff --git a/src/platform/os/fsstore.rs b/src/platform/os/fsstore.rs index 9568b2c..392169c 100644 --- a/src/platform/os/fsstore.rs +++ b/src/platform/os/fsstore.rs @@ -45,6 +45,10 @@ impl FileStore { Ok(FileStore { path: Path::new(&path).to_owned() }) } + + pub fn store_is_opened(store_path: String) -> MizeResult { + Ok(valid_pid_file(Path::new(&store_path))?.is_some()) + } } impl Store for FileStore { diff --git a/src/platform/os/mod.rs b/src/platform/os/mod.rs index f004fd6..f7b75fa 100644 --- a/src/platform/os/mod.rs +++ b/src/platform/os/mod.rs @@ -2,7 +2,7 @@ use ciborium::Value as CborValue; use serde::Deserialize; use std::env::VarError; use std::fs; -use tracing::{debug, error, warn, trace}; +use tracing::{debug, error, info, trace, warn}; use clap::ArgMatches; use std::path::Path; @@ -12,6 +12,7 @@ use crate::instance::Instance; use crate::error::{IntoMizeResult, MizeError, MizeResult}; use crate::item::{ItemData, IntoItemData}; use crate::memstore::MemStore; +use crate::mize_err; use crate::platform::os::unix_socket::UnixListener; use self::fsstore::FileStore; @@ -60,10 +61,27 @@ pub fn os_instance_init(instance: &mut Instance) -> MizeResult<()> { } }; - ////// if a config.store is set, upgrade to the filestore there - let store_path = instance.get("0/config/store_path")?.value_string()?; + ////// if a config.store_path is set, upgrade to the filestore there + let mut store_path = instance.get("0/config/store_path")?.value_string()?; - if store_path != "" { + if store_path == "" { + let home_dir = env!("HOME"); + if home_dir == "" { + return Err(mize_err!("env var $HOME empty")); + } + + // the default store_path: $HOME/.mize + store_path = home_dir.to_owned() + "/.mize"; + } + + if FileStore::store_is_opened(store_path.to_owned())? { + // if the store is already opened, connect to the instance, that opened it and join + // it's namespace + info!("CONNECTING"); + return Ok(()); + + } else { + // else open it ourselves let file_store = FileStore::new(store_path.as_str())?; instance.migrate_to_store(Box::new(file_store))?; diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index 7fa0ae4..67eb689 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -1,11 +1,11 @@ use std::fs; use std::path::{Path, PathBuf}; -use tokio::net::UnixListener as TokioUnixListener; +use tokio::net::{UnixListener as TokioUnixListener, UnixStream}; use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; use tokio::io::{Interest, AsyncReadExt, AsyncWriteExt}; use crossbeam::channel::{Receiver, Sender, unbounded}; use ciborium::Value as CborValue; -use tracing::warn; +use tracing::{info, warn}; use crate::instance::connection::{ConnListener, Connection}; use crate::instance::{self, Instance}; @@ -20,8 +20,35 @@ impl UnixListener { pub fn new(store_path: PathBuf) -> MizeResult { Ok(UnixListener { sock_path: store_path.join("sock").to_owned() }) } + } + +pub fn connect(instance: &mut Instance, store_path: PathBuf) -> MizeResult<()> { + instance.spawn_async("unix_connection", connect_async(instance.clone(), store_path) ); + + Ok(()) +} + + +async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResult<()> { + let mut stream = UnixStream::connect(store_path).await?; + let (read_half, write_half) = stream.split(); + + let (recieve_tx, recieve_rx) = unbounded::(); + let (send_tx, send_rx) = unbounded::(); + + let conn_id = instance.new_connection(recieve_rx, send_tx)?; + let ns_of_peer = instance.get(format!("inst/con_by_id/{}/peer/0/config/namespace", conn_id))?.value_string()?; + + info!("connect_async ... ns_of_peer: {}", ns_of_peer); + + //instance.connection_set_namespace(conn_id, namespace) + + Ok(()) +} + + impl ConnListener for UnixListener { fn listen(self, mut instance: Instance) -> MizeResult<()> { instance.spawn_async("unix listen async", unix_listen(self, instance.clone()));