basically fully restored

core/src/instance/connection.rs

@@ -0,0 +1,57 @@
+use flume::{Receiver, Sender};
+use tracing::{debug, error, info, trace, warn};
+
+use crate::error::{IntoMizeResult, MizeError, MizeResult};
+use crate::id::Namespace;
+use crate::mize_err;
+use crate::proto::MizeMessage;
+
+use super::Mize;
+use crate::item::{get_raw_from_cbor, Item, ItemData};
+
+// only created with Instance::new_connection()
+#[derive(Clone, Debug)]
+pub struct Connection {
+    pub tx: Sender<MizeMessage>,
+    pub id: u64,
+    pub ns: Option<Namespace>,
+}
+
+pub trait ConnListener: Send + Sync {
+    fn listen(self, instance: Mize) -> MizeResult<()>;
+}
+
+impl Connection {
+    pub fn send(&self, msg: MizeMessage) -> MizeResult<()> {
+        Ok(self.tx.send(msg)?)
+    }
+}
+
+pub fn value_raw_con_by_id(item: &mut Item) -> MizeResult<ItemData> {
+    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<ItemData> {
+    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.iter();
+    let skipped = new_id_str.skip(4);
+    let new_id = item.instance.new_id(skipped.cloned().collect::<Vec<String>>())?;
+
+    let connection = item.instance.get_connection(conn_id)?;
+
+    let msg = MizeMessage::new_get(new_id.clone(), connection.id);
+
+    connection.tx.send(msg)?;
+
+    let data = item.instance.give_msg_wait(new_id)?;
+
+    return Ok(data);
+}

core/src/instance/mod.rs

@@ -11,7 +11,7 @@ use std::sync::{Arc, Mutex};
 use std::{thread, vec};
 use tracing::{debug, info, trace};
 
-use crate::c_api::set_global_mize;
+use crate::set_global_mize;
 use crate::config::{ConfigOpt, ConfigOptId};
 use crate::error::{MizeError, MizeResult};
 use crate::id::{IntoMizeId, MizeId, Namespace};
@@ -259,7 +259,7 @@ impl Mize {
         set_global_mize(self.clone())?;
 
         // platform specific init code
-        crate::platform::instance_init(self)?;
+        crate::platform::init(self)?;
 
         // end of platform specific init code
 

core/src/instance/msg_thread.rs

@@ -0,0 +1,2 @@
+// Message thread handling for the Mize instance.
+// This module is reserved for future message threading functionality.

core/src/instance/store.rs

@@ -0,0 +1,71 @@
+use dyn_clone::DynClone;
+use std::fmt::Debug;
+use std::iter::Map;
+use std::option::Iter;
+
+use crate::error::MizeResult;
+use crate::id::MizeId;
+use crate::instance::Mize;
+use crate::item::get_raw_from_cbor;
+use crate::item::{Item, ItemData};
+use crate::memstore::MemStore;
+
+dyn_clone::clone_trait_object!(Store);
+
+pub trait Store: DynClone + Send + Sync + Debug {
+    // 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
+    // is implemented, so that there can be a multithreaded implementation
+    //fn get(self, id: MizeId) -> MizeResult<Item<Self>> where Self: Sized;
+
+    fn set(&self, id: MizeId, data: ItemData) -> MizeResult<()>;
+
+    // in the future should implement transactions, ....
+
+    // funcs to do with links, backlinks
+    fn get_links(&self, item: Item) -> MizeResult<Vec<MizeId>>;
+
+    fn get_backlinks(&self, item: Item) -> MizeResult<Vec<MizeId>>;
+
+    fn new_id(&self) -> MizeResult<String>;
+
+    fn get_value_raw(&self, id: MizeId) -> MizeResult<Vec<u8>>;
+
+    fn get_value_data_full(&self, id: MizeId) -> MizeResult<ItemData>;
+
+    fn id_iter(&self) -> MizeResult<IdIter>;
+
+    fn next_id(&self, prev_id: &str) -> MizeResult<Option<String>>;
+
+    fn first_id(&self) -> MizeResult<String>;
+}
+
+pub struct IdIter {
+    cur_id: String,
+    store: Box<dyn Store>,
+}
+
+impl IdIter {
+    pub fn new(store: Box<dyn Store>) -> MizeResult<IdIter> {
+        Ok(IdIter {
+            cur_id: store.first_id()?,
+            store,
+        })
+    }
+}
+
+impl Iterator for IdIter {
+    type Item = MizeResult<String>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.store.next_id(&self.cur_id) {
+            Err(err) => Some(Err(err)),
+            Ok(Some(new_id)) => {
+                self.cur_id = new_id.clone();
+                Some(Ok(new_id))
+            }
+            Ok(None) => None,
+        }
+    }
+}

core/src/instance/subscription.rs

@@ -0,0 +1,68 @@
+use core::fmt;
+use flume::Sender;
+use std::rc::Rc;
+use std::sync::Arc;
+use tracing::trace;
+
+use crate::error::MizeResult;
+use crate::id::MizeId;
+use crate::instance::connection::Connection;
+use crate::item::Item;
+use crate::proto::MizeMessage;
+
+use super::Mize;
+
+#[derive(Clone, Debug)]
+pub struct Update {
+    pub instance: Arc<Mize>,
+    pub id: MizeId,
+}
+
+impl Update {
+    pub fn new_item(&self) -> MizeResult<Item> {
+        self.instance.get(self.id.clone())
+    }
+}
+
+pub enum Subscription {
+    Connection(Connection),
+    Closure(Box<dyn Fn(Update) -> MizeResult<()> + Send>),
+    Channel(Sender<Update>),
+}
+
+impl Subscription {
+    pub fn from_conn(conn: Connection) -> Subscription {
+        Subscription::Connection(conn)
+    }
+    pub fn from_closure(closure: Box<dyn Fn(Update) -> MizeResult<()> + Send>) -> Subscription {
+        Subscription::Closure(closure)
+    }
+    pub fn from_sender(tx: Sender<Update>) -> Subscription {
+        Subscription::Channel(tx)
+    }
+
+    pub fn handle(&mut self, update: Update) -> MizeResult<()> {
+        trace!("handleing update");
+        match &self {
+            Subscription::Connection(conn) => {
+                let msg = MizeMessage::new_update(
+                    update.id.clone(),
+                    update.new_item()?.as_data_full()?,
+                    conn.id,
+                );
+                conn.send(msg)?;
+            }
+            Subscription::Closure(closure) => closure(update)?,
+            Subscription::Channel(tx) => {
+                tx.send(update);
+            }
+        }
+        Ok(())
+    }
+}
+
+impl fmt::Debug for Subscription {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "...Subscription...")
+    }
+}

core/src/instance/tests.rs

@@ -0,0 +1,252 @@
+use crate::item::IntoItemData;
+
+use super::*;
+
+#[test]
+fn test_sort_keys() -> MizeResult<()> {
+    let mut data = ItemData::from_toml(
+        r#"
+        [hi]
+        test = 4
+
+        [config]
+        zzz = "hiiii"
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "inner"
+    "#,
+    )?;
+
+    let data_sorted = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+        test = { inner = "inner" }
+        zzz = "hiiii"
+
+        [hi]
+        test = 4
+    "#,
+    )?;
+
+    data.sort_keys()?;
+
+    assert_eq!(data.to_json()?, data_sorted.to_json()?);
+
+    Ok(())
+}
+
+#[test]
+fn test_get_module_hash() -> MizeResult<()> {
+    let mut instance = Mize::empty()?;
+
+    use crate::platform::os::get_part_hash;
+    let selector_toml = r#"
+        mize_version = "0.0.0"
+        system = "x86_64-unknown-linux-gnu"
+        toolchain_version = "rustc 1.78.0 (9b00956e5 2024-04-29)"
+    "#;
+
+    instance.set_blocking("0/config/selector", ItemData::from_toml(&selector_toml)?)?;
+
+    let (hash, selector) = get_part_hash(&mut instance, "String", ItemData::new())?;
+
+    println!("hash: {}", hash);
+    println!("selector: {}", selector);
+
+    assert_eq!(hash, "e96edf60216688314efc263c1aa8d6ce".to_owned());
+
+    Ok(())
+}
+
+#[test]
+fn test_set_store_path() -> MizeResult<()> {
+    let instance = Mize::empty()?;
+
+    let store_path = "/home/me/.mize".to_owned();
+
+    let config = instance.get("0/config")?.as_data_full()?;
+
+    println!("data at 0/config: {}", config);
+
+    instance.set_blocking("0/config/store_path", store_path.clone().into_item_data())?;
+
+    let store_path_new = instance.get("0/config/store_path")?.value_string()?;
+
+    assert_eq!(store_path_new, store_path);
+
+    Ok(())
+}
+
+#[test]
+fn test_set_sub_path() -> MizeResult<()> {
+    let instance = Mize::empty()?;
+
+    let data = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "inner"
+    "#,
+    )?;
+
+    let data_two = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "new inner"
+    "#,
+    )?;
+
+    let item = instance.new_item()?;
+    instance.set_blocking(item.id(), data.clone())?;
+
+    assert_eq!(item.as_data_full()?, data);
+
+    // try to set a sub path of item
+    instance.set_blocking(
+        vec![item.id().store_part(), "config", "test", "inner"],
+        "new inner".into_item_data(),
+    )?;
+
+    // the contents of item should then be like data_two
+    println!(
+        "item with id '{}': {}",
+        item.id(),
+        instance.get(vec![item.id().store_part()])?.as_data_full()?
+    );
+    assert_eq!(
+        instance
+            .get(vec![item.id().store_part(), "config", "test", "inner"])?
+            .as_data_full()?,
+        ItemData::from_string("new inner")
+    );
+    assert_eq!(item.as_data_full()?, data_two);
+
+    Ok(())
+}
+
+#[test]
+fn test_itemdata_set_path() -> MizeResult<()> {
+    let mut data = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "inner"
+    "#,
+    )?;
+
+    let data_two = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "new inner"
+    "#,
+    )?;
+
+    data.set_path(
+        vec!["config", "test", "inner"],
+        ItemData::from_string("new inner"),
+    )?;
+
+    assert_eq!(data, data_two);
+
+    Ok(())
+}
+
+#[test]
+fn test_itemdata_merge() -> MizeResult<()> {
+    let mut data = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "inner"
+    "#,
+    )?;
+
+    let data_two = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "new inner"
+    "#,
+    )?;
+
+    assert_eq!(
+        data.get_path(vec!["config", "test", "inner"])?,
+        ItemData::from_string("inner")
+    );
+
+    data.merge(data_two.clone());
+
+    assert_eq!(data, data_two);
+
+    assert_eq!(
+        data.get_path(vec!["config", "test", "inner"])?,
+        ItemData::from_string("new inner")
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_item_merge() -> MizeResult<()> {
+    let instance = Mize::empty()?;
+
+    let mut item = instance.new_item()?;
+
+    // here the item should contain "empty" data
+    assert_eq!(item.as_data_full()?, ItemData::new());
+
+    let data = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "inner"
+    "#,
+    )?;
+
+    let data_two = ItemData::from_toml(
+        r#"
+        [config]
+        hi = "hello from config/hi"
+
+        [config.test]
+        inner = "new inner"
+    "#,
+    )?;
+
+    item.merge(data.clone())?;
+
+    // now it should contain data
+    assert_eq!(item.as_data_full()?, data);
+
+    let mut sub_item = instance.get(vec![item.id().store_part(), "config", "test", "inner"])?;
+
+    // the sub item should have the string "inner"
+    assert_eq!(sub_item.as_data_full()?, ItemData::from_string("inner"));
+
+    sub_item.merge(ItemData::from_string("new inner"));
+
+    assert_eq!(sub_item.as_data_full()?, ItemData::from_string("new inner"));
+
+    // also the item should be updated
+    assert_eq!(item.as_data_full()?, data_two);
+
+    Ok(())
+}

core/src/instance/updater.rs

@@ -0,0 +1,236 @@
+use flume::Receiver;
+use std::borrow::BorrowMut;
+use std::sync::Arc;
+use tracing::{error, trace, warn};
+
+use crate::error::{MizeError, MizeResult, MizeResultTrait};
+use crate::id::MizeId;
+use crate::mize_err;
+use crate::proto::{MessageCmd, MizeMessage};
+use crate::{instance::Mize, item::ItemData};
+
+use super::connection::{self, Connection};
+use super::subscription::{Subscription, Update};
+
+#[derive(Debug)]
+pub enum Operation {
+    Set(MizeId, ItemData, Option<Connection>), // bool: is_from_update_msg
+    Msg(MizeMessage),
+}
+
+// console_log macro
+// that can be copied into other files for debugging purposes
+#[cfg(feature = "wasm-target")]
+use wasm_bindgen::prelude::*;
+
+#[cfg(feature = "wasm-target")]
+#[wasm_bindgen]
+extern "C" {
+    #[wasm_bindgen(js_namespace = console)]
+    fn log(s: &str);
+}
+
+#[cfg(feature = "wasm-target")]
+macro_rules! console_log {
+    // Note that this is using the `log` function imported above during
+    // `bare_bones`
+    ($($t:tt)*) => (unsafe { log(&format_args!($($t)*).to_string())})
+}
+
+#[cfg(not(feature = "wasm-target"))]
+macro_rules! console_log {
+    // Note that this is using the `log` function imported above during
+    // `bare_bones`
+    ($($t:tt)*) => {};
+}
+//end of console_log macro
+
+pub async fn updater_thread_async(operation_rx: Receiver<Operation>, instance: Mize) -> () {
+    let mut count = 0;
+    console_log!("inside an updater thread");
+
+    loop {
+        let mut operation = match operation_rx.recv_async().await {
+            Ok(val) => val,
+            Err(e) => {
+                instance.report_err(e.into());
+                continue;
+            }
+        };
+
+        let op_str = match operation {
+            Operation::Set(_, _, _) => "SET",
+            Operation::Msg(_) => "MSG",
+        };
+
+        trace!("OPERATION {} - {}", count, op_str);
+        console_log!("OPERATION {} - {}", count, op_str);
+
+        let result = handle_operation(&mut operation, &instance);
+
+        trace!("OPERATION {} DONE", count);
+        count += 1;
+
+        if let Err(err) = result {
+            error!("OPERATION {} FAILED: {:?}", count, operation);
+            err.log();
+        }
+    }
+}
+
+pub fn updater_thread(operation_rx: Receiver<Operation>, instance: &Mize) -> MizeResult<()> {
+    let mut count = 0;
+
+    loop {
+        let mut operation = operation_rx.recv()?;
+        let op_str = match operation {
+            Operation::Set(_, _, _) => "SET",
+            Operation::Msg(_) => "MSG",
+        };
+
+        trace!("OPERATION {} - {}", count, op_str);
+
+        let result = handle_operation(&mut operation, instance);
+
+        trace!("OPERATION {} DONE", count);
+        count += 1;
+
+        if let Err(err) = result {
+            error!("OPERATION FAILED: {:?}", operation);
+            err.log();
+        }
+    }
+    Ok(())
+}
+
+pub fn handle_operation(operation: &mut Operation, instance: &Mize) -> MizeResult<()> {
+    match operation {
+        Operation::Set(id, value, maybe_conn) => {
+            let item_data: ItemData = value.to_owned();
+            let mut item = instance.get(id.clone())?;
+            item.merge(item_data)?;
+
+            //check subs and handle them
+            let mut subs_inner = instance.subs.lock()?;
+            if let Some(vec) = subs_inner.get_mut(&id) {
+                let update = Update {
+                    instance: Arc::new(instance.to_owned()),
+                    id: id.clone(),
+                };
+                for sub in vec.iter_mut() {
+                    // don't handle sub of type connection, in case the update comes from this
+                    // connection
+                    if let Some(conn) = maybe_conn {
+                        if let Subscription::Connection(conn2) = sub {
+                            if conn.id == conn2.id {
+                                continue;
+                            }
+                        }
+                    }
+
+                    sub.handle(update.clone());
+                }
+            }
+        }
+        Operation::Msg(msg) => handle_msg(msg, instance)?,
+    }
+    Ok(())
+}
+
+fn handle_msg(msg: &mut MizeMessage, instance: &Mize) -> MizeResult<()> {
+    match msg.cmd()? {
+        MessageCmd::Get => {
+            let id = msg.id(instance)?;
+            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)?;
+        }
+
+        MessageCmd::GetSub => {
+            let id = msg.id(instance)?;
+            let mut connection = instance.get_connection(msg.conn_id)?.clone();
+            let item = instance.get(id.clone())?;
+            let msg = MizeMessage::new_give(id.clone(), item.as_data_full()?, msg.conn_id);
+            connection.send(msg)?;
+            let sub = Subscription::from_conn(connection.clone());
+            instance.sub(id, sub)?;
+        }
+
+        MessageCmd::Sub => {
+            let id = msg.id(instance)?;
+            let mut connection = instance.get_connection(msg.conn_id)?.clone();
+            let sub = Subscription::from_conn(connection.clone());
+            instance.sub(id, sub)?;
+        }
+
+        MessageCmd::Update => {
+            let data = msg.data()?;
+            let id = msg.id(instance)?;
+            let connection = instance.get_connection(msg.conn_id)?;
+            instance
+                .op_tx
+                .send(Operation::Set(id.clone(), data, Some(connection)));
+        }
+
+        // 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)?;
+            let connection = instance.get_connection(msg.conn_id)?;
+            instance
+                .op_tx
+                .send(Operation::Set(id.clone(), data, Some(connection)));
+        }
+
+        MessageCmd::Give => {
+            let id = msg.id(instance)?;
+            let data = msg.data()?;
+            let give_msg_wait_inner = instance.give_msg_wait.lock()?;
+            if let Some(vec) = give_msg_wait_inner.get(&id) {
+                for tx in vec {
+                    tx.send(data.clone());
+                }
+            } else {
+                warn!(
+                    "got give msg for id '{}', that has no one waiting for it",
+                    id
+                );
+            }
+        }
+
+        MessageCmd::Create => {
+            println!("instance.store: {:?}", instance.clone().store);
+            let item = instance.new_item()?;
+            let reply_msg = MizeMessage::new_create_reply(item.id(), msg.conn_id);
+            let mut connection = instance.get_connection(msg.conn_id)?;
+            connection.send(reply_msg)?;
+        }
+
+        MessageCmd::CreateReply => {
+            let create_msg_wait_inner = instance.create_msg_wait.lock()?;
+            if let Some(sender) = create_msg_wait_inner.as_ref() {
+                sender.send(msg.id(&mut instance.clone())?)?;
+            }
+            return Ok(());
+        }
+        MessageCmd::Custom(type_val) => {
+            let listeners = instance.listeners.lock()?;
+            let type_str = type_val.value_string()?;
+            let msg = msg.clone();
+            let mut mize = instance.clone();
+            println!("type_str: {}", type_str);
+            println!("listeners: {:?}", listeners.keys());
+            if let Some(listener) = listeners.get(&type_str) {
+                let listener = listener.clone();
+                drop(listeners);
+                listener(msg, &mut mize)?;
+            }
+        }
+        _ => {
+            return Err(mize_err!("got a message, that is not handeled"));
+        }
+    }
+    Ok(())
+}