diff --git a/src/core/instance/connection.rs b/src/core/instance/connection.rs index 64c9ec9..3a1ec81 100644 --- a/src/core/instance/connection.rs +++ b/src/core/instance/connection.rs @@ -10,7 +10,7 @@ use super::Instance; use crate::item::{get_raw_from_cbor, Item, ItemData}; // only created with Instance::new_connection() -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct Connection { pub tx: Sender, pub id: u64, @@ -22,7 +22,7 @@ pub trait ConnListener : Send + Sync { } impl Connection { - pub fn send(&mut self, msg: MizeMessage) -> MizeResult<()> { + pub fn send(&self, msg: MizeMessage) -> MizeResult<()> { Ok(self.tx.send(msg)?) } } diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 7ca1911..f68df8e 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -26,6 +26,7 @@ use crate::proto::MizeMessage; use tokio::runtime::Handle; use self::connection::{ConnListener, Connection}; +use self::updater::handle_operation; #[cfg(feature = "async")] use tokio::runtime::Runtime; @@ -46,7 +47,7 @@ pub struct Instance { pub store: Arc>>, connections: Arc>>, next_con_id: Arc>, - subs: Arc>>, + subs: Arc>>>, pub id_pool: Arc>, pub namespace_pool: Arc>, @@ -148,8 +149,11 @@ impl Instance { pub fn with_config(config: ItemData) -> MizeResult { trace!("[ {} ] Instance::with_config()", "CALL".yellow()); trace!("config: {}", config); + let mut instance = Instance::empty()?; - instance.set("0", config.clone()); + + instance.set_blocking("0", config.clone()); + instance.init()?; Ok(instance) @@ -211,6 +215,25 @@ impl Instance { self.op_tx.send(Operation::Set(id, value.into())); Ok(()) } + + pub fn set_blocking>(&self, id: I, value: V) -> MizeResult<()> { + handle_operation(&mut Operation::Set(id.to_mize_id(self)?, value.into()), self)?; + Ok(()) + } + + pub fn sub(&self, id: I, sub: Subscription) -> MizeResult<()> { + let mut subs_inner = self.subs.lock()?; + let id = id.to_mize_id(self)?; + match subs_inner.get_mut(&id) { + Some(vec) => { + vec.push(sub); + }, + None => { + subs_inner.insert(id, vec![sub]); + } + } + Ok(()) + } pub fn new_id(&self, value: T) -> MizeResult { value.to_mize_id(self) diff --git a/src/core/instance/subscription.rs b/src/core/instance/subscription.rs index 57efe31..679a675 100644 --- a/src/core/instance/subscription.rs +++ b/src/core/instance/subscription.rs @@ -1,4 +1,66 @@ +use core::fmt; +use std::rc::Rc; +use crossbeam::channel::Sender; +use std::sync::Arc; +use tracing::trace; +use crate::id::MizeId; +use crate::instance::connection::Connection; +use crate::item::Item; +use crate::error::MizeResult; +use crate::proto::MizeMessage; -#[derive(Debug)] -pub struct Subscription; +use super::Instance; + +#[derive(Clone, Debug)] +pub struct Update { + pub instance: Arc, + pub id: MizeId, +} + +impl Update { + pub fn new_item(&self) -> MizeResult { + self.instance.get(self.id.clone()) + } +} + +pub enum Subscription { + Connection(Connection), + Closure(Box MizeResult<()> + Send>), + Channel(Sender), +} + +impl Subscription { + pub fn from_conn(conn: Connection) -> Subscription { + Subscription::Connection(conn) + } + pub fn from_closure(closure: Box MizeResult<()> + Send>) -> Subscription { + Subscription::Closure(closure) + } + pub fn from_sender(tx: Sender) -> 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...") + } +} diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index 4b23bec..8c51722 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -1,5 +1,7 @@ use crossbeam::channel::Receiver; use tracing::{error, trace, warn}; +use std::borrow::BorrowMut; +use std::sync::Arc; use crate::mize_err; use crate::{instance::Instance, item::ItemData}; @@ -8,6 +10,7 @@ use crate:: error::{MizeResult, MizeError, MizeResultTrait}; use crate::proto::{MessageCmd, MizeMessage}; use super::connection::{self, Connection}; +use super::subscription::{Subscription, Update}; #[derive(Debug)] @@ -21,13 +24,18 @@ pub fn updater_thread(operation_rx : Receiver, instance: &Instance) - loop { let mut operation = operation_rx.recv()?; - trace!("OPERATION {}", count); + let op_str = match operation { + Operation::Set(_, _) => "SET", + Operation::Msg(_) => "MSG", + }; + + trace!("OPERATION {} - {}", count, op_str); //let mut busy = instance.update_thread_busy.lock()?; let result = handle_operation(&mut operation, instance); - trace!("OPERATION DONE {}", count); + trace!("OPERATION {} DONE", count); count += 1; if let Err(err) = result { @@ -40,12 +48,26 @@ pub fn updater_thread(operation_rx : Receiver, instance: &Instance) - Ok(()) } -fn handle_operation(operation: &mut Operation, instance: &Instance) -> MizeResult<()> { +pub fn handle_operation(operation: &mut Operation, instance: &Instance) -> MizeResult<()> { match operation { Operation::Set(id, value) => { 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()?; + println!("goooooooooooooooooot some subs: {:?}", subs_inner); + println!("some subs id: {:?}", id); + 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() { + sub.handle(update.clone()); + } + } }, Operation::Msg(msg) => { handle_msg(msg, instance)? @@ -64,10 +86,20 @@ fn handle_msg(msg: &mut MizeMessage, instance: &Instance) -> MizeResult<()> { 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::Update => { let data = msg.data()?; let id = msg.id(instance)?; - instance.set(id, data); + instance.set(id.clone(), data); }, // this should check, if the update is valid @@ -75,7 +107,7 @@ fn handle_msg(msg: &mut MizeMessage, instance: &Instance) -> MizeResult<()> { MessageCmd::UpdateRequest => { let data = msg.data()?; let id = msg.id(instance)?; - instance.set(id, data); + instance.set(id.clone(), data); }, MessageCmd::Give => { diff --git a/src/core/item.rs b/src/core/item.rs index 376b23e..7609386 100644 --- a/src/core/item.rs +++ b/src/core/item.rs @@ -107,14 +107,14 @@ impl Item<'_> { pub fn merge>(&mut self, mut value: V) -> MizeResult<()> { - println!("before as_data_full: {}", self.instance.op_tx.len()); let mut data = self.as_data_full()?; - println!("data: {:?}", data); + println!("item::merge data: {:?}", data); + println!("item::merge id: {:?}", self.id()); let new_data = value.into(); data.merge(new_data); - println!("new_data: {:?}", data); + println!("item::merge new_data: {:?}", data); if self.instance.we_are_namespace()? { let store_inner = self.instance.store.lock()?; diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index 5a4f08e..2e9b49c 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -25,6 +25,7 @@ static CMD_GIVE: u16 = 3; static CMD_CREATE: u16 = 4; static CMD_CREATE_REPLY: u16 = 5; static CMD_UPDATE_REQUEST: u16 = 6; +static CMD_GET_SUB: u16 = 7; #[derive(Debug)] pub enum MessageCmd { @@ -34,6 +35,7 @@ pub enum MessageCmd { Create, CreateReply, UpdateRequest, + GetSub, } @@ -54,6 +56,16 @@ impl MizeMessage { MizeMessage::new(value, conn_id) } + pub fn new_get_sub(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_GET_SUB.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_create(conn_id: u64) -> MizeMessage { let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_CREATE.into())); let value = CborValue::Map(vec![cmd]); diff --git a/src/platform/os/cli/show.rs b/src/platform/os/cli/show.rs index 039e1ac..a037c84 100644 --- a/src/platform/os/cli/show.rs +++ b/src/platform/os/cli/show.rs @@ -1,12 +1,29 @@ use clap::ArgMatches; -use mize::error::MizeResult; +use crossbeam::channel::{bounded, unbounded}; +use mize::error::{MizeResult, MizeError}; +use mize::instance::subscription::{Subscription, Update}; use mize::instance::Instance; +use mize::platform::os::config_from_cli_args; pub fn show(sub_matches: &ArgMatches) -> MizeResult<()> { - let instance = Instance::new()?; + let instance = Instance::with_config(config_from_cli_args(sub_matches)?)?; - println!("0/config/test is: {}", instance.get("0/config/test")?); + let id = sub_matches.get_one::("id") + .ok_or(MizeError::new().msg("No id Argument specified"))?; + + let item = instance.get(id)?; + + let (tx, rx) = bounded::(4); + let sub = Subscription::from_sender(tx); + instance.sub(id, sub)?; + + println!("ItemData: {}", item.as_data_full()?); + + for update in rx { + println!("####### GOT UPDATE #######"); + println!("ItemData: {}", update.new_item()?.as_data_full()?); + } Ok(()) } diff --git a/src/platform/os/mod.rs b/src/platform/os/mod.rs index c9f2efd..d31c162 100644 --- a/src/platform/os/mod.rs +++ b/src/platform/os/mod.rs @@ -30,7 +30,7 @@ pub fn os_instance_init(instance: &mut Instance) -> MizeResult<()> { Ok(config_file_path) => { let config = config_from_file(config_file_path)?; debug!("env var MIZE_CONFIG_FILE present"); - instance.set("0", config)?; + instance.set_blocking("0", config)?; trace!("config after MIZE_CONFIG_FILE env var: {}", instance.get("0/config")?); } Err(var_err) => match var_err { @@ -48,7 +48,7 @@ pub fn os_instance_init(instance: &mut Instance) -> MizeResult<()> { Ok(config_string) => { let config = config_from_string(config_string)?; debug!("env var MIZE_CONFIG present"); - instance.set("0", config)?; + instance.set_blocking("0", config)?; trace!("config after MIZE_CONFIG env var: {}", instance.get("0/config")?); } Err(var_err) => match var_err { @@ -62,6 +62,8 @@ pub fn os_instance_init(instance: &mut Instance) -> MizeResult<()> { }; ////// if a config.store_path is set, upgrade to the filestore there + let mut test = instance.get("0")?.as_data_full()?; + println!("os::Mod test: {}", test); let mut store_path = instance.get("0/config/store_path")?.value_string()?; if store_path == "" {