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

new_item works via connection

Browse Source
This commit is contained in:
Sebastian Moser
2024-08-19 09:58:35 +02:00
parent f529fbd625
commit d0a1e26e96
8 changed files with 93 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<dyn Store>,
pub store: Arc<Mutex<Box<dyn Store>>>,
connections: Arc<Mutex<Vec<Connection>>>,
next_con_id: Arc<Mutex<u64>>,
subs: Arc<Mutex<HashMap<MizeId, Subscription>>>,
@@ -76,7 +76,6 @@ pub struct InstanceAsync {
impl Instance {
pub fn empty() -> MizeResult<Instance> {
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<dyn Store>) -> MizeResult<()> {
pub fn migrate_to_store(&self, new_store: Box<dyn Store>) -> 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<Item> {
pub fn new_item(&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()?)?;
@@ -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<I: IntoMizeId, V: Into<ItemData>>(&mut self, id: I, value: V) -> MizeResult<()> {
pub fn set<I: IntoMizeId, V: Into<ItemData>>(&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<Namespace> {
pub fn get_self_namespace(&self) -> MizeResult<Namespace> {
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<MizeMessage>) -> MizeResult<u64> {
pub fn new_connection(&self, tx: Sender<MizeMessage>) -> MizeResult<u64> {
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

3
src/core/instance/store.rs
View File

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

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

@@ -16,10 +16,10 @@ pub enum Operation {
Msg(MizeMessage),
}
pub fn updater_thread(operation_rx : Receiver<Operation>, mut instance: Instance) -> MizeResult<()> {
pub fn updater_thread(operation_rx : Receiver<Operation>, 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<Operation>, 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)?;
}

10
src/core/item.rs
View File

@@ -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<V: Into<ItemData>>(&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(())
}
}

1
src/core/memstore.rs
View File

@@ -44,6 +44,7 @@ impl Store for MemStore {
Ok(Vec::new())
}
fn new_id(&self) -> MizeResult<String> {
println!("are we hereeeeeee memstore new_get");
let mut inner = self.inner.lock()?;
let id_string = format!("{}", inner.next_id);

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

@@ -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<MizeId> {
pub fn id(&mut self, instance: &Instance) -> MizeResult<MizeId> {
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(())