diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index bd4090c..4e8fe16 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -43,7 +43,7 @@ static MSG_CHANNEL_SIZE: usize = 200; #[derive(Clone)] pub struct Instance { // a bit a lot of Mutexes isn-it??? - pub store: Box, + pub store: Arc>>, connections: Arc>>, next_con_id: Arc>, subs: Arc>>, @@ -76,7 +76,6 @@ pub struct InstanceAsync { impl Instance { pub fn empty() -> MizeResult { - let store = MemStore::new(); let id_pool = Arc::new(Mutex::new(VecStringPool::default())); let namespace_pool_raw = StringPool::default(); let connections = Arc::new(Mutex::new(Vec::new())); @@ -88,7 +87,7 @@ impl Instance { let self_namespace = Arc::new(Mutex::new(Namespace ( namespace_pool_raw.get("mize.default.namespace") ))); let mut instance = Instance { - store: Box::new(store), + store: Arc::new(Mutex::new(Box::new(MemStore::new()))), connections, subs, id_pool, namespace, self_namespace, op_tx, namespace_pool: Arc::new(Mutex::new(namespace_pool_raw)), @@ -102,7 +101,7 @@ impl Instance { }; let instance_clone = instance.clone(); - let closure = move || updater_thread(op_rx, instance_clone); + let closure = move || updater_thread(op_rx, &instance_clone); instance.spawn("updater_thread", closure)?; // will probably move the msg stuff into it's own thread @@ -135,6 +134,7 @@ impl Instance { // end of platform specific init code + debug!("INSTANCE INIT DONE"); Ok(()) } @@ -145,36 +145,30 @@ impl Instance { instance.set("0", config.clone()); instance.init()?; - // set it again, so that the passed config data has presidence over anything the init would set - debug!("overwriting instance config again with the one passed to Instance::with_config()"); - instance.set("0", config); - - debug!("instance inited with conifg: {}", instance.get("0/config")?); - Ok(instance) } - pub fn migrate_to_store(&mut self, new_store: Box) -> MizeResult<()> { + pub fn migrate_to_store(&self, new_store: Box) -> MizeResult<()> { println!("MIGRATING"); - let old_store = &self.store; + let mut old_store = self.store.lock()?; let id = self.id_from_string("0".to_owned())?; - let inst_data = self.store.get_value_data_full(id.clone())?; + 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 = self.store.get_value_data_full(self.id_from_string(id?)?)?; + 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())?; }; - self.store = new_store; + *old_store = new_store; Ok(()) } - pub fn new_item(&mut self) -> MizeResult { + pub fn new_item(&self) -> MizeResult { if self.get_namespace()? != self.get_self_namespace()? { // need to send create msg and wait for it let mut connection = self.get_connection_by_ns(self.get_namespace()?)?; @@ -195,7 +189,8 @@ impl Instance { return Ok(Item::new(id, self)); } - let id = self.id_from_string(self.store.new_id()?)?; + let store_inner = self.store.lock()?; + let id = self.id_from_string(store_inner.new_id()?)?; return Ok(Item::new(id, self)); } @@ -204,7 +199,7 @@ impl Instance { return Ok(Item::new(id, &self)); } - pub fn set>(&mut self, id: I, value: V) -> MizeResult<()> { + 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())); Ok(()) @@ -247,7 +242,7 @@ impl Instance { Ok(namespace) } - pub fn set_namespace(&mut self, ns: Namespace) -> MizeResult<()> { + pub fn set_namespace(&self, ns: Namespace) -> MizeResult<()> { let mut namespace_inner = self.namespace.lock()?; *namespace_inner = ns; @@ -259,7 +254,7 @@ impl Instance { return Ok(namespace_inner.clone()); } - pub fn get_self_namespace(&mut self) -> MizeResult { + pub fn get_self_namespace(&self) -> MizeResult { let mut self_namespace_inner = self.self_namespace.lock()?; return Ok(self_namespace_inner.clone()); } @@ -270,7 +265,7 @@ impl Instance { Ok(()) } - pub fn new_connection(&mut self, tx: Sender) -> MizeResult { + 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; @@ -281,18 +276,18 @@ impl Instance { Ok(old_next_con_id) } - pub fn connection_set_namespace(&mut self, conn_id: u64, namespace: Namespace) -> MizeResult<()> { + 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(crate) fn got_msg(&mut self, msg: MizeMessage) -> MizeResult<()> { + pub(crate) fn got_msg(&self, msg: MizeMessage) -> MizeResult<()> { Ok(self.op_tx.send(Operation::Msg(msg))?) } - fn set_connection(&mut self, conn_id: u64, new_connection: Connection) -> MizeResult<()> { + 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() { @@ -368,7 +363,7 @@ impl Instance { runtime_inner.spawn(func); } #[cfg(feature = "async")] - pub fn async_get_handle(&mut self) -> Handle { + pub fn async_get_handle(&self) -> Handle { let runtime_inner = self.runtime.lock().unwrap(); let handle = runtime_inner.handle().to_owned(); handle diff --git a/src/core/instance/store.rs b/src/core/instance/store.rs index dc4cec4..7e5012f 100644 --- a/src/core/instance/store.rs +++ b/src/core/instance/store.rs @@ -1,3 +1,4 @@ +use std::fmt::Debug; use std::iter::Map; use std::option::Iter; use dyn_clone::DynClone; @@ -11,7 +12,7 @@ use crate::item::get_raw_from_cbor; dyn_clone::clone_trait_object!(Store); -pub trait Store: DynClone + Send + Sync { +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 diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 6b8bd5f..86d92b7 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -16,10 +16,10 @@ pub enum Operation { Msg(MizeMessage), } -pub fn updater_thread(operation_rx : Receiver, mut instance: Instance) -> MizeResult<()> { +pub fn updater_thread(operation_rx : Receiver, instance: &Instance) -> MizeResult<()> { for mut operation in operation_rx { trace!("OPERATION"); - let result = handle_operation(&mut operation, &mut instance); + let result = handle_operation(&mut operation, instance); if let Err(err) = result { error!("OPERATION FAILED: {:?}", operation); @@ -29,7 +29,7 @@ pub fn updater_thread(operation_rx : Receiver, mut instance: Instance Ok(()) } -fn handle_operation(operation: &mut Operation, instance: &mut Instance) -> MizeResult<()> { +fn handle_operation(operation: &mut Operation, instance: &Instance) -> MizeResult<()> { match operation { Operation::Set(id, value) => { let item_data: ItemData = value.to_owned(); @@ -43,8 +43,7 @@ fn handle_operation(operation: &mut Operation, instance: &mut Instance) -> MizeR Ok(()) } -fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> { - println!("got msg: {}", msg); +fn handle_msg(msg: &mut MizeMessage, instance: &Instance) -> MizeResult<()> { match msg.cmd()? { MessageCmd::Get => { let id = msg.id(instance)?; @@ -74,9 +73,10 @@ fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> }, MessageCmd::Create => { + println!("instance.store: {:?}", instance.clone().store); let item = instance.new_item()?; - let reply_msg = MizeMessage::new_create_reply(item.id()); - let connection = instance.get_connection(msg.conn_id)?; + 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)?; } diff --git a/src/core/item.rs b/src/core/item.rs index b4aee67..99bc90e 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -1,4 +1,4 @@ -use tracing::{debug, error, info, instrument, trace, warn}; +use tracing::{debug, error, info, instrument, trace, warn, Instrument}; use serde::Serialize; use core::fmt; use std::{i128, string}; @@ -92,7 +92,8 @@ impl Item<'_> { } } - return self.instance.store.get_value_data_full(self.id()) + let store_inner = self.instance.store.lock()?; + return store_inner.get_value_data_full(self.id()); } else { @@ -105,13 +106,14 @@ impl Item<'_> { } pub fn merge>(&mut self, mut value: V) -> MizeResult<()> { - let mut old_data = self.instance.store.get_value_data_full(self.id())?; + let store_inner = self.instance.store.lock()?; + let mut old_data = store_inner.get_value_data_full(self.id())?; let new_data = value.into(); old_data.merge(new_data); - self.instance.store.set(self.id(), old_data)?; + store_inner.set(self.id(), old_data)?; Ok(()) } } diff --git a/src/core/memstore.rs b/src/core/memstore.rs index 9e5ccf8..a475776 100644 --- a/src/core/memstore.rs +++ b/src/core/memstore.rs @@ -44,6 +44,7 @@ impl Store for MemStore { Ok(Vec::new()) } fn new_id(&self) -> MizeResult { + println!("are we hereeeeeee memstore new_get"); let mut inner = self.inner.lock()?; let id_string = format!("{}", inner.next_id); diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index ed11d03..a402ac4 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -4,7 +4,7 @@ use serde::Serialize; use ciborium::{value::Integer, Value as CborValue}; -use crate::{error::{MizeError, MizeResult}, id::MizeId, instance::{self, connection::Connection, Instance}, item::{IntoItemData, ItemData}}; +use crate::{error::{IntoMizeResult, MizeError, MizeResult}, id::MizeId, instance::{self, connection::Connection, Instance}, item::{IntoItemData, ItemData}}; #[derive(Clone)] pub struct MizeMessage { @@ -60,6 +60,16 @@ impl MizeMessage { MizeMessage::new(value, conn_id) } + pub fn new_create_reply(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_CREATE_REPLY.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(); @@ -167,7 +177,7 @@ impl MizeMessage { Ok(vec_string) } - pub fn id(&mut self, instance: &mut Instance) -> MizeResult { + pub fn id(&mut self, instance: &Instance) -> MizeResult { let id_str = self.id_str()?; return instance.new_id(id_str); } @@ -220,28 +230,52 @@ impl<'a, 'b> io::Write for DisplayWriter<'a, 'b> { impl fmt::Display for MizeMessage { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut err = false; - if let Ok(cmd) = self.to_owned().cmd() { - writeln!(f, "MizeMessage with cmd: {:?}", cmd); - } else { - writeln!(f, "MizeMessage with unparsable cmd"); - }; - - if let Ok(id) = self.to_owned().id_str() { - let id_str = id.join("/"); - writeln!(f, "\tid: {}", id_str); - } else { - writeln!(f, "id: parsing error"); + match self.to_owned().cmd() { + Ok(cmd) => { + writeln!(f, "MizeMessage with cmd: {:?}", cmd); + }, + Err(e) => { + writeln!(f, "MizeMessage with unparsable cmd"); + err = true; + } } - if let Ok(data) = self.to_owned().data() { - let value = data.cbor(); - write!(f, "\tdata: "); + match self.to_owned().id_str() { + Ok(id) => { + let id_str = id.join("/"); + writeln!(f, "\tid: {:?}", id_str); + }, + Err(e) => { + writeln!(f, "\t msg has no id"); + //err = true; + } + } + + match self.to_owned().data() { + Ok(data) => { + let value = data.cbor(); + write!(f, "\tdata: "); + let display_writer = DisplayWriter (f); + if let Err(e) = value.serialize(&mut serde_json::Serializer::pretty(display_writer)) + .map_err(|serde_err| std::fmt::Error) { + writeln!(f, "serialize err: {:?}", e); + err = true; + } + }, + Err(e) => { + writeln!(f, "\t msg has no data"); + //err = true; + }, + } + + if err { + write!(f, "Full msg: "); let display_writer = DisplayWriter (f); - value.serialize(&mut serde_json::Serializer::pretty(display_writer)) - .map_err(|serde_err| std::fmt::Error)? - } else { - writeln!(f, "\tdata: unparsable"); + self.value.serialize(&mut serde_json::Serializer::pretty(display_writer)) + .map_err(|serde_err| std::fmt::Error).mize_result_msg("serialize error") + .inspect_err(|e| { e.clone().log(); }); } Ok(()) diff --git a/src/platform/os/cli/create.rs b/src/platform/os/cli/create.rs index 32f48ff..fc5a74e 100644 --- a/src/platform/os/cli/create.rs +++ b/src/platform/os/cli/create.rs @@ -16,6 +16,7 @@ pub fn create(sub_matches: &ArgMatches) -> MizeResult<()> { let item = instance.new_item()?; println!("id: {}", item.id()); + println!("with namespace: {}", item.id().namespace().as_real_string()); return Ok(()); } diff --git a/src/platform/os/fsstore.rs b/src/platform/os/fsstore.rs index 492a1df..3005804 100644 --- a/src/platform/os/fsstore.rs +++ b/src/platform/os/fsstore.rs @@ -15,7 +15,7 @@ use crate::core::id::MizeId; use crate::item::get_raw_from_cbor; use crate::instance::Instance; -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct FileStore { path: PathBuf, } @@ -61,6 +61,7 @@ impl Store for FileStore { .parse() .mize_result_msg(format!("could not parse next_id at '{}' to u64", self.path.display()))?; + println!("nextid fsstore: {}", next_id); let id_string = format!("{}", next_id); next_id += 1;