ppc/mize
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

set msg

Browse Source
This commit is contained in:
Sebastian Moser
2024-08-18 12:12:18 +02:00
parent b60c9c2f69
commit f529fbd625
11 changed files with 269 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/core/instance/mod.rs
View File

@@ -60,6 +60,7 @@ pub struct Instance {
context: Vec<MizeId>,
threads: Vec<String>,
give_msg_wait: Arc<Mutex<HashMap<MizeId, Vec<Sender<ItemData>>>>>,
create_msg_wait: Arc<Mutex<Option<Sender<MizeId>>>>,
#[cfg(feature = "async")]
pub runtime: Arc<Mutex<Runtime>>,
}
@@ -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<Item> {
pub fn new_item(&mut self) -> MizeResult<Item> {
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::<MizeId>(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<I: IntoMizeId>(&self, id: I) -> MizeResult<Item> {
@@ -232,6 +254,16 @@ impl Instance {
Ok(())
}
pub fn get_namespace(&self) -> MizeResult<Namespace> {
let mut namespace_inner = self.namespace.lock()?;
return Ok(namespace_inner.clone());
}
pub fn get_self_namespace(&mut self) -> MizeResult<Namespace> {
let mut self_namespace_inner = self.self_namespace.lock()?;
return Ok(self_namespace_inner.clone());
}
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));

23
src/core/instance/updater.rs
View File

@@ -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(())
}

36
src/core/item.rs
View File

@@ -55,23 +55,6 @@ impl Item<'_> {
pub fn as_data_full(&self) -> MizeResult<ItemData> {
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);
}
}

109
src/core/proto/mod.rs
View File

@@ -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<MizeId> {
pub fn id_str(&mut self) -> MizeResult<Vec<String>> {
// 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<MizeId> {
let id_str = self.id_str()?;
return instance.new_id(id_str);
}
pub fn data(&mut self) -> MizeResult<ItemData> {
@@ -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<usize, std::io::Error> {
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)
}
}

22
src/platform/os/cli/create.rs Normal file
View File

@@ -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(());
}

27
src/platform/os/cli/is_running.rs Normal file
View File

@@ -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(())
}

32
src/platform/os/cli/set.rs
View File

@@ -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::<String>("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(());
}

1
src/platform/os/fsstore.rs
View File

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

18
src/platform/os/main.rs
View File

@@ -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();

2
src/platform/os/mod.rs
View File

@@ -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;

4
src/platform/os/unix_socket.rs
View File

@@ -122,7 +122,7 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu
fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver<MizeMessage>, 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(())