realising, that an extra msg_thread will be nececary

src/core/instance/connection.rs

@@ -55,3 +55,5 @@ pub fn value_raw_from_peer(item: &mut Item) -> MizeResult<ItemData> {
 
     return Ok(data);
 }
+
+

src/core/instance/mod.rs

@@ -5,7 +5,7 @@ use std::thread::JoinHandle;
 use std::{thread, vec};
 use crossbeam::channel::{self, bounded, Receiver, Sender};
 use std::sync::{Arc, Mutex};
-use tracing::{trace, debug, info, warn, error};
+use tracing::{debug, error, info, trace, warn, Instrument};
 use uuid::Uuid;
 use std::fs::File;
 use colored::Colorize;
@@ -56,11 +56,12 @@ pub struct Instance {
     // the namespace of the instance itself
     // TODO: set to a random uuid
     pub self_namespace: Arc<Mutex<Namespace>>,
-    op_tx: channel::Sender<Operation>,
+    pub op_tx: channel::Sender<Operation>,
     context: Vec<MizeId>,
     threads: Vec<String>,
     give_msg_wait: Arc<Mutex<HashMap<MizeId, Vec<Sender<ItemData>>>>>,
     create_msg_wait: Arc<Mutex<Option<Sender<MizeId>>>>,
+    update_thread_busy: Arc<Mutex<bool>>,
     #[cfg(feature = "async")]
     pub runtime: Arc<Mutex<Runtime>>,
 }
@@ -95,15 +96,21 @@ impl Instance {
             threads: vec![],
             next_con_id: Arc::new(Mutex::new(1)),
             give_msg_wait, create_msg_wait,
+            update_thread_busy: Arc::new(Mutex::new(false)),
 
             #[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);
+        let op_rx_clone = op_rx.clone();
+        let closure = move || updater_thread(op_rx_clone, &instance_clone);
         instance.spawn("updater_thread", closure)?;
 
+        let instance_clone_two = instance.clone();
+        let closure_two = move || updater_thread(op_rx, &instance_clone_two);
+        instance.spawn("updater_thread", closure_two)?;
+
         // will probably move the msg stuff into it's own thread
         //let msg_instance_clone = instance.clone();
         //let msg_closure = move || msg_thread(op_rx, instance_clone);
@@ -259,6 +266,10 @@ impl Instance {
         return Ok(self_namespace_inner.clone());
     }
 
+    pub fn we_are_namespace(&self) -> MizeResult<bool> {
+        Ok(self.get_namespace()? == self.get_self_namespace()?)
+    }
+
     pub fn add_listener<T: ConnListener +'static>(&mut self, listener: T) -> MizeResult<()> {
         let mut instance_clone = self.clone();
         self.spawn("some_listener", move || listener.listen(instance_clone));
@@ -387,6 +398,17 @@ impl Instance {
             thread::sleep_ms(10000000)
         }
     }
+
+    pub fn wait_for_updaater_thread(&self) -> MizeResult<()> {
+        thread::sleep_ms(10000000);
+        return Ok(());
+        while !self.op_tx.is_empty() {
+            thread::sleep_ms(50);
+        }
+        self.update_thread_busy.lock()?;
+        trace!("wait_for_updaater_thread Thread Idle");
+        Ok(())
+    }
 }
 
 impl fmt::Debug for Instance {

src/core/instance/updater.rs

@@ -17,14 +17,27 @@ pub enum Operation {
 }
 
 pub fn updater_thread(operation_rx : Receiver<Operation>, instance: &Instance) -> MizeResult<()> {
-    for mut operation in operation_rx {
+    let mut count = 0;
-        trace!("OPERATION");
+
+    loop {
+        println!("updater_thread loop beginning: {}", operation_rx.len());
+
+        let mut operation = operation_rx.recv()?;
+        trace!("OPERATION {}", count);
+
+        //let mut busy = instance.update_thread_busy.lock()?;
+
         let result = handle_operation(&mut operation, instance);
 
+        trace!("OPERATION DONE {}", count);
+        count += 1;
+
         if let Err(err) = result {
             error!("OPERATION FAILED: {:?}", operation);
             err.log();
         }
+
+        //drop(busy);
     }
     Ok(())
 }
@@ -59,6 +72,14 @@ fn handle_msg(msg: &mut MizeMessage, instance: &Instance) -> MizeResult<()> {
             instance.set(id, data);
         },
 
+        // this should check, if the update is valid
+        // but for now, we do just always accept it
+        MessageCmd::UpdateRequest => {
+            let data = msg.data()?;
+            let id = msg.id(instance)?;
+            instance.set(id, data);
+        },
+
         MessageCmd::Give => {
             let id = msg.id(instance)?;
             let data = msg.data()?;

src/core/item.rs

@@ -11,7 +11,7 @@ use std::io;
 use tracing::{span, Level};
 
 use crate::error::{MizeError, MizeResult, IntoMizeResult};
-use crate::instance::Instance;
+use crate::instance::{connection, Instance};
 use crate::instance::store::Store;
 use crate::id::MizeId;
 use crate::proto::MizeMessage;
@@ -106,14 +106,26 @@ impl Item<'_> {
     }
 
     pub fn merge<V: Into<ItemData>>(&mut self, mut value: V) -> MizeResult<()> {
-        let store_inner = self.instance.store.lock()?;
+
-        let mut old_data = store_inner.get_value_data_full(self.id())?;
+        println!("before as_data_full: {}", self.instance.op_tx.len());
+        let mut data = self.as_data_full()?;
+        println!("data: {:?}", data);
 
         let new_data = value.into();
 
-        old_data.merge(new_data);
+        data.merge(new_data);
+        println!("new_data: {:?}", data);
+
+        if self.instance.we_are_namespace()? {
+            let store_inner = self.instance.store.lock()?;
+            store_inner.set(self.id(), data);
+        } else {
+            let namespace = self.id().namespace();
+            let mut connection = self.instance.get_connection_by_ns(namespace)?;
+            let msg = MizeMessage::new_update_request(self.id(), data, connection.id);
+            connection.send(msg)?;
+        }
 
-        store_inner.set(self.id(), old_data)?;
         Ok(())
     }
 }
@@ -131,6 +143,10 @@ impl ItemData {
         ItemData(CborValue::Null)
     }
 
+    pub fn empty() -> ItemData {
+        ItemData(CborValue::Null)
+    }
+
     pub fn from_string<S: Into<String>>(into_string: S) -> ItemData {
         let string = into_string.into();
         let data = CborValue::Text(string);
@@ -419,6 +435,16 @@ impl IntoItemData for &str {
         ItemData::parse(self)
     }
 }
+impl IntoItemData for String {
+    fn into_item_data(self) -> ItemData {
+        ItemData(self.into())
+    }
+}
+impl IntoItemData for &String {
+    fn into_item_data(self) -> ItemData {
+        ItemData(self.to_owned().into())
+    }
+}
 impl IntoItemData for CborValue {
     fn into_item_data(self) -> ItemData {
         ItemData ( self )

src/core/proto/mod.rs

@@ -1,5 +1,6 @@
 use core::fmt;
 use std::io;
+use nix::libc::CTRL_CMD_UNSPEC;
 use serde::Serialize;
 
 use ciborium::{value::Integer, Value as CborValue};
@@ -81,6 +82,28 @@ impl MizeMessage {
         MizeMessage::new(value, conn_id)
     }
 
+    pub fn new_update_request(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_UPDATE_REQUEST.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]);
+
+        MizeMessage::new(value, conn_id)
+    }
+
+    pub fn new_update(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_UPDATE.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]);
+
+        MizeMessage::new(value, conn_id)
+    }
+
     pub fn value(self) -> CborValue {
         self.value
     }

src/platform/os/cli/set.rs

@@ -5,6 +5,7 @@ use home::home_dir;
 
 use mize::error::{IntoMizeResult, MizeError, MizeResult, MizeResultTrait};
 use mize::instance::Instance;
+use mize::item::IntoItemData;
 use mize::platform::os::config_from_cli_args;
 
 
@@ -16,19 +17,12 @@ pub fn set(sub_matches: &ArgMatches) -> MizeResult<()> {
     let id = sub_matches.get_one::<String>("id")
         .ok_or(MizeError::new().msg("No id Argument specified"))?;
 
-    let item = instance.get(id)?;
+    let value = sub_matches.get_one::<String>("value")
+        .ok_or(MizeError::new().msg("No value Argument specified"))?;
 
-    let recurse_count = sub_matches.get_count("recurse");
+    instance.set(id, value.into_item_data())?;
 
-    println!("{}", item.as_data_full()?);
+    instance.wait_for_updaater_thread()?;
-    return Ok(());
 
-    if recurse_count > 0 {
+    Ok(())
-        println!("{}", item.id());
-        println!("{}", item.as_data_full()?);
-    } else {
-        println!("{}", item);
-    };
-
-    return Ok(());
 }

src/platform/os/fsstore.rs

@@ -42,7 +42,9 @@ impl FileStore {
         }
 
         // init the store
-        fs::write(path.join("next_id"), "1");
+        if !path.join("next_id").exists() {
+            fs::write(path.join("next_id"), "1");
+        }
 
         Ok(FileStore { path: Path::new(&path).to_owned() })
     }
@@ -110,6 +112,9 @@ impl Store for FileStore {
     fn get_value_data_full(&self, id: MizeId) -> MizeResult<ItemData> {
         let file_path = self.path.join("store").join(id.namespace_str()).join(id.store_part());
         println!("file_path: {:?}", file_path.display());
+        if !file_path.exists() {
+            return Ok(ItemData::empty());
+        }
         let file = OpenOptions::new().read(true).open(file_path)?;
 
         let cbor_value: CborValue = ciborium::from_reader(file)

src/platform/os/main.rs

@@ -164,6 +164,7 @@ fn cli_matches() -> clap::ArgMatches {
                 Command::new("set")
                 .aliases(["s"])
                 .arg(Arg::new("id").help("The id or path to set"))
+                .arg(Arg::new("value").help("The value to set the path to"))
             )
         .subcommand(
                 Command::new("show")

src/platform/os/unix_socket.rs

@@ -10,7 +10,7 @@ use tracing::{info, warn};
 use crate::instance::connection::{ConnListener, Connection};
 use crate::instance::{self, Instance};
 use crate::error::{MizeError, MizeResult, IntoMizeResult};
-use crate::proto::MizeMessage;
+use crate::proto::{self, MizeMessage};
 
 pub struct UnixListener {
     sock_path: PathBuf,
@@ -140,6 +140,7 @@ fn unix_incomming(mut unix_read: OwnedReadHalf, mut instance: Instance, conn_id:
         }
         let msg = MizeMessage::new(value, conn_id);
         println!("unix incoming got msg: {}", msg);
+        println!("op channel len: {}", instance.op_tx.len());
         instance.got_msg(msg);
     }
     Ok(())