cargo check

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<String> {
         instance.id_from_vec_string(self)
     }
 }
+impl IntoMizeId for Vec<&String> {
+    fn to_mize_id(self, instance: &Instance) -> MizeResult<MizeId> {
+        let owned = self.into_iter().map(|s| s.to_owned()).collect::<Vec<String>>();
+        instance.id_from_vec_string(owned)
+    }
+}
 impl IntoMizeId for &[&str] {
     fn to_mize_id(self, instance: &Instance) -> MizeResult<MizeId> {
         let mut vec: Vec<String> = Vec::new();

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<MizeMessage>,
     pub tx: Sender<MizeMessage>,
     pub id: u64,
+    pub ns: Option<Namespace>,
 }
 
 pub trait ConnListener : Send + Sync {
@@ -23,5 +28,33 @@ impl Connection {
     }
 }
 
+pub fn value_raw_con_by_id(item: &mut Item) -> MizeResult<Vec<u8>> {
+    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<Vec<u8>> {
+    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::<Vec<&String>>())?;
+
+    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());
+
+}

src/core/instance/mod.rs

@@ -56,6 +56,7 @@ pub struct Instance {
     threads: Vec<String>,
     #[cfg(feature = "async")]
     pub runtime: Arc<Mutex<Runtime>>,
+    give_msg_wait: Arc<Mutex<HashMap<MizeId, (Sender<ItemData>, Receiver<ItemData>)>>>,
 }
 
 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<Connection> {
+    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<Connection> {
         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<ItemData> {
+
+        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::<ItemData>(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<F: Future<Output = impl Send + Sync + 'static> + Send + Sync + 'static>(&mut self, name: &str, func: F) {
         self.threads.push(name.to_owned());

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)?;

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<Vec<u8>> {
+        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())
     }

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]);

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<bool> {
+        Ok(valid_pid_file(Path::new(&store_path))?.is_some())
+    }
 }
 
 impl Store for FileStore {

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))?;
 

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<UnixListener> {
         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::<MizeMessage>();
+    let (send_tx, send_rx) = unbounded::<MizeMessage>();
+    
+    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()));