From f529fbd625ec6a80b35a310be1019b45ced9cd30 Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Sun, 18 Aug 2024 12:12:18 +0200 Subject: [PATCH] set msg --- src/core/instance/mod.rs | 38 ++++++++++- src/core/instance/updater.rs | 23 ++++++- src/core/item.rs | 36 +++++----- src/core/proto/mod.rs | 109 ++++++++++++++++++++++++++---- src/platform/os/cli/create.rs | 22 ++++++ src/platform/os/cli/is_running.rs | 27 ++++++++ src/platform/os/cli/set.rs | 32 ++++++++- src/platform/os/fsstore.rs | 1 + src/platform/os/main.rs | 18 +++++ src/platform/os/mod.rs | 2 +- src/platform/os/unix_socket.rs | 4 +- 11 files changed, 269 insertions(+), 43 deletions(-) create mode 100644 src/platform/os/cli/create.rs create mode 100644 src/platform/os/cli/is_running.rs diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 327b7d2..bd4090c 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -60,6 +60,7 @@ pub struct Instance { context: Vec, threads: Vec, give_msg_wait: Arc>>>>, + create_msg_wait: Arc>>>, #[cfg(feature = "async")] pub runtime: Arc>, } @@ -82,6 +83,7 @@ impl Instance { let subs = Arc::new(Mutex::new(HashMap::new())); let (op_tx, op_rx) = channel::unbounded(); let give_msg_wait = Arc::new(Mutex::new(HashMap::new())); + let create_msg_wait = Arc::new(Mutex::new(None)); let namespace = Arc::new(Mutex::new(Namespace ( namespace_pool_raw.get("mize.default.namespace") ))); let self_namespace = Arc::new(Mutex::new(Namespace ( namespace_pool_raw.get("mize.default.namespace") ))); @@ -93,7 +95,7 @@ impl Instance { context: vec![], threads: vec![], next_con_id: Arc::new(Mutex::new(1)), - give_msg_wait, + give_msg_wait, create_msg_wait, #[cfg(feature = "async")] runtime: Arc::new(Mutex::new(Runtime::new().mize_result_msg("Could not create async runtime")?)), @@ -172,9 +174,29 @@ impl Instance { Ok(()) } - pub fn new_item(&self) -> MizeResult { + pub fn new_item(&mut 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()?)?; + + let msg = MizeMessage::new_create(connection.id); + + connection.send(msg)?; + + let (tx, rx) = bounded::(1); + + let mut msg_wait_inner = self.create_msg_wait.lock()?; + *msg_wait_inner = Some(tx); + drop(msg_wait_inner); + + let id = rx.recv()?; + println!("new_item namespace: {:?}", id.namespace()); + + return Ok(Item::new(id, self)); + } + let id = self.id_from_string(self.store.new_id()?)?; - return Ok(Item::new(id, &self)); + return Ok(Item::new(id, self)); } pub fn get(&self, id: I) -> MizeResult { @@ -232,6 +254,16 @@ impl Instance { Ok(()) } + pub fn get_namespace(&self) -> MizeResult { + let mut namespace_inner = self.namespace.lock()?; + return Ok(namespace_inner.clone()); + } + + pub fn get_self_namespace(&mut self) -> MizeResult { + let mut self_namespace_inner = self.self_namespace.lock()?; + return Ok(self_namespace_inner.clone()); + } + pub fn add_listener(&mut self, listener: T) -> MizeResult<()> { let mut instance_clone = self.clone(); self.spawn("some_listener", move || listener.listen(instance_clone)); diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 8437e14..6b8bd5f 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -44,7 +44,7 @@ fn handle_operation(operation: &mut Operation, instance: &mut Instance) -> MizeR } fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> { - println!("got msg: {:?}", msg); + println!("got msg: {}", msg); match msg.cmd()? { MessageCmd::Get => { let id = msg.id(instance)?; @@ -53,11 +53,13 @@ fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> let msg = MizeMessage::new_give(id, item.as_data_full()?, msg.conn_id); connection.send(msg)?; }, - MessageCmd::Set => { + + MessageCmd::Update => { let data = msg.data()?; let id = msg.id(instance)?; instance.set(id, data); }, + MessageCmd::Give => { let id = msg.id(instance)?; let data = msg.data()?; @@ -70,9 +72,24 @@ fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> warn!("got give msg for id '{}', that has no one waiting for it", id); } }, + + MessageCmd::Create => { + let item = instance.new_item()?; + let reply_msg = MizeMessage::new_create_reply(item.id()); + let 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(()); + } _ => { return Err(mize_err!("got a message, that is not handeled")); - } + }, } Ok(()) } diff --git a/src/core/item.rs b/src/core/item.rs index 36b4456..b4aee67 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -55,23 +55,6 @@ impl Item<'_> { pub fn as_data_full(&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()); }, - "namespace" => { - let namespace_inner = self.instance.namespace.lock()?; - return Ok(ItemData::from_string(namespace_inner.as_real_string())); - }, - "self_namespace" => { - let namespace_inner = self.instance.self_namespace.lock()?; - return Ok(ItemData::from_string(namespace_inner.as_real_string())); - }, - _ => {}, - } - } // the instance we are if id.store_part() == "self" { @@ -93,16 +76,31 @@ impl Item<'_> { if self.id().namespace() == self_namespace_inner.to_owned() { debug!("getting item from store"); + + if id.store_part() == "inst" { + match id.nth_part(1)? { + "con_by_id" => { return value_raw_con_by_id(&mut self.clone()); }, + "namespace" => { + let namespace_inner = self.instance.namespace.lock()?; + return Ok(ItemData::from_string(namespace_inner.as_real_string())); + }, + "self_namespace" => { + let namespace_inner = self.instance.self_namespace.lock()?; + return Ok(ItemData::from_string(namespace_inner.as_real_string())); + }, + _ => {}, + } + } + return self.instance.store.get_value_data_full(self.id()) - } else { + } else { let mut connection = self.instance.get_connection_by_ns(self.id().namespace())?; let msg = MizeMessage::new_get(self.id(), connection.id); connection.send(msg); let data = self.instance.give_msg_wait(self.id())?; return Ok(data); - } } diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index babe30f..ed11d03 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -1,8 +1,12 @@ +use core::fmt; +use std::io; +use serde::Serialize; + use ciborium::{value::Integer, Value as CborValue}; -use crate::{error::{MizeError, MizeResult}, id::MizeId, instance::{connection::Connection, Instance}, item::{IntoItemData, ItemData}}; +use crate::{error::{MizeError, MizeResult}, id::MizeId, instance::{self, connection::Connection, Instance}, item::{IntoItemData, ItemData}}; -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct MizeMessage { value: CborValue, pub conn_id: u64, @@ -15,8 +19,24 @@ static MSG_DATA: u16 = 3; // cmds static CMD_GET: u16 = 1; -static CMD_SET: u16 = 2; +static CMD_UPDATE: u16 = 2; static CMD_GIVE: u16 = 3; +static CMD_CREATE: u16 = 4; +static CMD_CREATE_REPLY: u16 = 5; +static CMD_UPDATE_REQUEST: u16 = 6; + +#[derive(Debug)] +pub enum MessageCmd { + Get, + Update, + Give, + Create, + CreateReply, + UpdateRequest, +} + + + impl MizeMessage { pub fn new(value: CborValue, conn_id: u64) -> MizeMessage { @@ -33,6 +53,13 @@ impl MizeMessage { MizeMessage::new(value, conn_id) } + pub fn new_create(conn_id: u64) -> MizeMessage { + let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_CREATE.into())); + let value = CborValue::Map(vec![cmd]); + + 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(); @@ -80,8 +107,11 @@ impl MizeMessage { // match on value of c let cmd = match val_of_c { 1 => MessageCmd::Get, - 2 => MessageCmd::Set, + 2 => MessageCmd::Update, 3 => MessageCmd::Give, + 4 => MessageCmd::Create, + 5 => MessageCmd::CreateReply, + 6 => MessageCmd::UpdateRequest, _ => { return Err(MizeError::new().msg("error cmd of msg was not a valid command")); }, @@ -91,7 +121,7 @@ impl MizeMessage { } - pub fn id(&mut self, instance: &mut Instance) -> MizeResult { + pub fn id_str(&mut self) -> MizeResult> { // return err, if msg is not a map let msg_as_map = match &self.value { CborValue::Map(val) => val, @@ -134,8 +164,12 @@ impl MizeMessage { } } - let id = instance.id_from_vec_string(vec_string)?; - Ok(id) + Ok(vec_string) + } + + pub fn id(&mut self, instance: &mut Instance) -> MizeResult { + let id_str = self.id_str()?; + return instance.new_id(id_str); } pub fn data(&mut self) -> MizeResult { @@ -168,12 +202,61 @@ impl MizeMessage { } } -pub enum MessageCmd { - Get, - Set, - Give, + +// thanks to: https://stackoverflow.com/a/61768916 +struct DisplayWriter<'a, 'b>(&'a mut fmt::Formatter<'b>); + +impl<'a, 'b> io::Write for DisplayWriter<'a, 'b> { + fn write(&mut self, bytes: &[u8]) -> std::result::Result { + + self.0.write_str(&String::from_utf8_lossy(bytes)) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err))?; + + Ok(bytes.len()) + } + fn flush(&mut self) -> std::result::Result<(), std::io::Error> { todo!() } +} + +impl fmt::Display for MizeMessage { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + + + 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"); + } + + if let Ok(data) = self.to_owned().data() { + let value = data.cbor(); + write!(f, "\tdata: "); + 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"); + } + + Ok(()) + + } +} + +impl fmt::Debug for MizeMessage { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "\nMizeMessage: "); + let display_writer = DisplayWriter (f); + let value = self.to_owned().value(); + value.serialize(&mut serde_json::Serializer::pretty(display_writer)) + .map_err(|serde_err| std::fmt::Error) + } } - - diff --git a/src/platform/os/cli/create.rs b/src/platform/os/cli/create.rs new file mode 100644 index 0000000..32f48ff --- /dev/null +++ b/src/platform/os/cli/create.rs @@ -0,0 +1,22 @@ +use std::path::PathBuf; + +use clap::ArgMatches; +use home::home_dir; + +use mize::error::{IntoMizeResult, MizeError, MizeResult, MizeResultTrait}; +use mize::instance::Instance; +use mize::platform::os::config_from_cli_args; + + + +pub fn create(sub_matches: &ArgMatches) -> MizeResult<()> { + + let mut instance = Instance::with_config(config_from_cli_args(sub_matches)?)?; + + let item = instance.new_item()?; + + println!("id: {}", item.id()); + + return Ok(()); +} + diff --git a/src/platform/os/cli/is_running.rs b/src/platform/os/cli/is_running.rs new file mode 100644 index 0000000..6d54999 --- /dev/null +++ b/src/platform/os/cli/is_running.rs @@ -0,0 +1,27 @@ +use std::path::PathBuf; + +use clap::ArgMatches; +use home::home_dir; + +use mize::error::{IntoMizeResult, MizeError, MizeResult, MizeResultTrait}; +use mize::instance::Instance; +use mize::platform::os::config_from_cli_args; + +use mize::platform::os::fsstore::FileStore; +//use self::FileStore; + + +pub fn is_running(sub_matches: &ArgMatches) -> MizeResult<()> { + + let home_dir = env!("HOME"); + + if FileStore::store_is_opened(home_dir.to_owned() + "/.mize")? { + println!("true"); + } else { + println!("false"); + } + + Ok(()) + +} + diff --git a/src/platform/os/cli/set.rs b/src/platform/os/cli/set.rs index daf51fb..e1131ba 100644 --- a/src/platform/os/cli/set.rs +++ b/src/platform/os/cli/set.rs @@ -1,6 +1,34 @@ +use std::path::PathBuf; + use clap::ArgMatches; -use mize::error::MizeResult; +use home::home_dir; + +use mize::error::{IntoMizeResult, MizeError, MizeResult, MizeResultTrait}; +use mize::instance::Instance; +use mize::platform::os::config_from_cli_args; + + pub fn set(sub_matches: &ArgMatches) -> MizeResult<()> { - Ok(()) + + let instance = Instance::with_config(config_from_cli_args(sub_matches)?)?; + + let id = sub_matches.get_one::("id") + .ok_or(MizeError::new().msg("No id Argument specified"))?; + + let item = instance.get(id)?; + + let recurse_count = sub_matches.get_count("recurse"); + + println!("{}", item.as_data_full()?); + return Ok(()); + + if recurse_count > 0 { + println!("{}", item.id()); + println!("{}", item.as_data_full()?); + } else { + println!("{}", item); + }; + + return Ok(()); } diff --git a/src/platform/os/fsstore.rs b/src/platform/os/fsstore.rs index 8b12d8c..492a1df 100644 --- a/src/platform/os/fsstore.rs +++ b/src/platform/os/fsstore.rs @@ -35,6 +35,7 @@ impl FileStore { } else { // write our own pid file + fs::remove_file(path.join("pid")); let pid = std::process::id(); let mut file = OpenOptions::new().write(true).create(true).open(path.join("pid"))?; write!(file, "{}", pid)?; diff --git a/src/platform/os/main.rs b/src/platform/os/main.rs index 5ee95cc..0a8e106 100644 --- a/src/platform/os/main.rs +++ b/src/platform/os/main.rs @@ -28,14 +28,19 @@ mod cli { pub mod show; pub mod mount; pub mod call; + pub mod create; + pub mod is_running; + pub use run::run; pub use stop::stop; pub use get::get; pub use set::set; pub use show::show; + pub use create::create; pub use mount::mount; pub use call::call; + pub use is_running::is_running; } mod logging; @@ -57,6 +62,8 @@ fn main() { // mi daemon Some(("run", sub_matches)) => cli::run(sub_matches), + Some(("is-running", sub_matches)) => cli::is_running(sub_matches), + // mi serve Some(("serve", sub_matches)) => cli::stop(sub_matches), @@ -75,6 +82,9 @@ fn main() { // mi call Some(("call", sub_matches)) => cli::call(sub_matches), + // mi create + Some(("create", sub_matches)) => cli::create(sub_matches), + // some unknown command passed Some((cmd, sub_matches)) => Err(MizeError::new().msg(format!("The subcommand: {} is not known. use --help to list availavle commands", cmd))), @@ -164,6 +174,14 @@ fn cli_matches() -> clap::ArgMatches { Command::new("call") .aliases(["c"]) ) + .subcommand( + Command::new("create") + .aliases(["cr"]) + ) + .subcommand( + Command::new("is-running") + .aliases(["isr"]) + ) .arg_required_else_help(true); return main.get_matches(); diff --git a/src/platform/os/mod.rs b/src/platform/os/mod.rs index a511683..c9f2efd 100644 --- a/src/platform/os/mod.rs +++ b/src/platform/os/mod.rs @@ -17,7 +17,7 @@ use crate::platform::os::unix_socket::UnixListener; use self::fsstore::FileStore; -mod fsstore; +pub mod fsstore; mod unix_socket; //mod web; diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index ccfef9d..9cd8eb3 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -122,7 +122,7 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver, mut instance: Instance, conn_id: u64) -> MizeResult<()> { for msg in send_rx { - println!("unix outgoing got msg: {:?}", msg); + println!("unix outgoing got msg: {}", msg); let adapter = MyCiboriumWriter { inner: &mut unix_write, instance: &mut instance }; let value = msg.value(); ciborium::into_writer(&value, adapter)? @@ -139,7 +139,7 @@ fn unix_incomming(mut unix_read: OwnedReadHalf, mut instance: Instance, conn_id: break; } let msg = MizeMessage::new(value, conn_id); - println!("unix incoming got msg: {:?}", msg); + println!("unix incoming got msg: {}", msg); instance.got_msg(msg); } Ok(())