diff --git a/src/core/instance/peer.rs b/src/core/instance/connection.rs similarity index 50% rename from src/core/instance/peer.rs rename to src/core/instance/connection.rs index fb9582e..e9a013f 100644 --- a/src/core/instance/peer.rs +++ b/src/core/instance/connection.rs @@ -6,18 +6,20 @@ use crate::error::{MizeError, MizeResult, IntoMizeResult}; use super::Instance; -pub struct Peer { - rx: Receiver, - tx: Sender, +#[derive(Clone)] +pub struct Connection { + pub rx: Receiver, + pub tx: Sender, + pub id: u64, } -pub trait PeerListener : Send + Sync { +pub trait ConnListener : Send + Sync { fn listen(self, instance: Instance) -> MizeResult<()>; } -impl Peer { - pub fn new(rx: Receiver, tx: Sender) -> Peer { - Peer { rx , tx } +impl Connection { + pub fn send(&mut self, msg: MizeMessage) -> MizeResult<()> { + Ok(self.tx.send(msg)?) } } diff --git a/src/core/instance/mod.rs b/src/core/instance/mod.rs index 6891d2e..8beb82c 100644 --- a/src/core/instance/mod.rs +++ b/src/core/instance/mod.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use std::fs::create_dir; use std::thread::JoinHandle; use std::{thread, vec}; -use crossbeam::channel; +use crossbeam::channel::{self, Receiver, Sender}; use std::sync::{Arc, Mutex}; use tracing::{trace, debug, info, warn, error}; use uuid::Uuid; @@ -16,23 +16,25 @@ use interner::shared::{VecStringPool, StringPool}; use crate::error::{MizeError, MizeResult, IntoMizeResult, MizeResultTrait}; use crate::instance::store::Store; use crate::instance::updater::updater_thread; -use crate::instance::peer::Peer; use crate::id::{IntoMizeId, MizeId, Namespace}; use crate::instance::subscription::Subscription; use crate::item::{Item, ItemData}; use crate::memstore::MemStore; use crate::instance::updater::Operation; +use crate::mize_err; +use crate::proto::MizeMessage; -use self::peer::PeerListener; +use self::connection::{ConnListener, Connection}; #[cfg(feature = "async")] use tokio::runtime::Runtime; use core::future::Future; -pub mod peer; +pub mod connection; pub mod store; pub mod subscription; pub mod updater; +pub mod msg_thread; static MSG_CHANNEL_SIZE: usize = 200; @@ -43,7 +45,8 @@ pub struct Instance { // would mean replication logic and such things is handeled by the instance // i think it would be better thogh to then implement a ReplicatedStore pub store: Box, - peers: Arc>>, + connections: Arc>>, + next_con_id: Arc>, subs: Arc>>, pub id_pool: Arc>, pub namespace_pool: Arc>, @@ -69,18 +72,19 @@ impl Instance { let store = MemStore::new(); let id_pool = Arc::new(Mutex::new(VecStringPool::default())); let namespace_pool_raw = StringPool::default(); - let peers = Arc::new(Mutex::new(Vec::new())); + let connections = Arc::new(Mutex::new(Vec::new())); let subs = Arc::new(Mutex::new(HashMap::new())); let namespace = Namespace ( namespace_pool_raw.get("mize.default.namespace") ); let (op_tx, op_rx) = channel::unbounded(); let mut instance = Instance { store: Box::new(store), - peers, subs, id_pool, + connections, subs, id_pool, namespace, op_tx, namespace_pool: Arc::new(Mutex::new(namespace_pool_raw)), context: vec![], threads: vec![], + next_con_id: Arc::new(Mutex::new(1)), #[cfg(feature = "async")] runtime: Arc::new(Mutex::new(Runtime::new().mize_result_msg("Could not create async runtime")?)), @@ -90,6 +94,11 @@ impl Instance { 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 + //let msg_instance_clone = instance.clone(); + //let msg_closure = move || msg_thread(op_rx, instance_clone); + //instance.spawn("msg_thread", closure)?; + return Ok(instance); } @@ -195,16 +204,32 @@ impl Instance { Ok(namespace) } - pub fn add_listener(&mut self, listener: T) -> MizeResult<()> { + pub fn add_listener(&mut self, listener: T) -> MizeResult<()> { let mut instance_clone = self.clone(); self.spawn("some_listener", move || listener.listen(instance_clone)); Ok(()) } - pub fn add_peer(&mut self, peer: Peer) -> MizeResult<()> { - let mut peer_inner = self.peers.lock()?; - peer_inner.push(peer); - Ok(()) + pub fn new_connection(&mut self, rx: Receiver, tx: Sender) -> MizeResult { + let mut conn_inner = self.connections.lock()?; + let mut next_con_id = self.next_con_id.lock()?; + + let connection = Connection { id: next_con_id.to_owned(), rx, tx}; + conn_inner.push(connection); + *next_con_id += 1; + Ok(next_con_id.to_owned()) + } + + pub fn get_connection(&mut self, conn_id: u64) -> MizeResult { + let mut conn_inner = self.connections.lock()?; + + for connection in conn_inner.iter() { + if connection.id == conn_id { + return Ok(connection.clone()); + } + } + + return Err(mize_err!("Connection with id {} not known to instance", conn_id)); } pub fn spawn(&mut self, name: &str, func: impl FnOnce() -> MizeResult<()> + Send + 'static) -> MizeResult<()> { diff --git a/src/core/instance/msg_thread.rs b/src/core/instance/msg_thread.rs new file mode 100644 index 0000000..139597f --- /dev/null +++ b/src/core/instance/msg_thread.rs @@ -0,0 +1,2 @@ + + diff --git a/src/core/instance/updater.rs b/src/core/instance/updater.rs index e8072e4..195d5ae 100644 --- a/src/core/instance/updater.rs +++ b/src/core/instance/updater.rs @@ -1,14 +1,19 @@ use crossbeam::channel::Receiver; use tracing::{trace, error}; +use crate::mize_err; use crate::{instance::Instance, item::ItemData}; use crate::id::MizeId; use crate:: error::{MizeResult, MizeError, MizeResultTrait}; +use crate::proto::{MessageCmd, MizeMessage}; + +use super::connection::{self, Connection}; #[derive(Debug)] pub enum Operation { Set(MizeId, ItemData), + Msg(MizeMessage), } pub fn updater_thread(operation_rx : Receiver, mut instance: Instance) -> MizeResult<()> { @@ -20,7 +25,7 @@ pub fn updater_thread(operation_rx : Receiver, mut instance: Instance error!("OPERATION FAILED: {:?}", operation); } } - Ok(() ) + Ok(()) } fn handle_operation(operation: &mut Operation, instance: &mut Instance) -> MizeResult<()> { @@ -29,7 +34,35 @@ fn handle_operation(operation: &mut Operation, instance: &mut Instance) -> MizeR let item_data: ItemData = value.to_owned(); let mut item = instance.get(id.clone())?; item.merge(item_data)?; + }, + Operation::Msg(msg) => { + handle_msg(msg, instance)? + }, + } + Ok(()) +} + +fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()> { + println!("got msg: {:?}", msg); + match msg.cmd()? { + MessageCmd::Get => { + let id = msg.id(instance)?; + let mut connection = instance.get_connection(msg.conn_id)?; + let item = instance.get(id.clone())?; + let msg = MizeMessage::new_give(id, item.as_data_full()?, msg.conn_id); + connection.send(msg)?; + }, + MessageCmd::Set => { + let data = msg.data()?; + let id = msg.id(instance)?; + instance.set(id, data); + }, + _ => { + return Err(mize_err!("got a message, that is not handeled")); } } Ok(()) } + + + diff --git a/src/core/proto/mod.rs b/src/core/proto/mod.rs index 080076d..4f4146b 100644 --- a/src/core/proto/mod.rs +++ b/src/core/proto/mod.rs @@ -1,8 +1,164 @@ -use ciborium::Value as CborValue; +use ciborium::{value::Integer, Value as CborValue}; + +use crate::{error::{MizeError, MizeResult}, id::MizeId, instance::{connection::Connection, Instance}, item::{IntoItemData, ItemData}}; #[derive(Debug, Clone)] pub struct MizeMessage { value: CborValue, + pub conn_id: u64, +} + +// msg items +static MSG_CMD: u16 = 1; +static MSG_ID: u16 = 2; +static MSG_DATA: u16 = 3; + +// cmds +static CMD_GET: u16 = 1; +static CMD_SET: u16 = 2; +static CMD_GIVE: u16 = 3; + +impl MizeMessage { + pub fn new(value: CborValue, conn_id: u64) -> MizeMessage { + MizeMessage { value, conn_id } + } + + pub fn new_give(id: MizeId, data: ItemData, conn_id: u64) -> MizeMessage { + let cmd = (CborValue::Integer(MSG_CMD.into()), CborValue::Integer(CMD_GIVE.into())); + let id_path = id.path().into_iter().map(|string| CborValue::Text(string.to_owned())).collect(); + let id = (CborValue::Integer(MSG_ID.into()), CborValue::Array(id_path)); + let data = (CborValue::Integer(MSG_DATA.into()), data.cbor().to_owned()); + let value = CborValue::Map(vec![cmd, id, data]); + + MizeMessage::new(value, conn_id) + } + + pub fn value(self) -> CborValue { + self.value + } + + pub fn cmd(&self) -> MizeResult { + // return err, if msg is not a map + let msg_as_map = match &self.value { + CborValue::Map(val) => val, + _ => { + return Err(MizeError::new().msg("Message was not a map")); + }, + }; + + //check if in this map there is a c, otherwise return err + let val_of_c = { + let mut tmp_val = 0; + for (key, val) in msg_as_map { + if let CborValue::Integer(key_int) = key { + // number one indicates the cmd field + let one: Integer = 1.into(); + if key_int == &one { + if let CborValue::Integer(val_int) = val { + tmp_val = val_int.to_owned().into(); + } + } + } + } + if tmp_val == 0 { + return Err(MizeError::new().msg("error getting the cmd value form a msg")); + } + tmp_val + }; + + // match on value of c + let cmd = match val_of_c { + 1 => MessageCmd::Get, + 2 => MessageCmd::Set, + _ => { + return Err(MizeError::new().msg("error cmd of msg was not a valid command")); + }, + }; + + return Ok(cmd); + } + + + pub fn id(&mut self, instance: &mut Instance) -> MizeResult { + // return err, if msg is not a map + let msg_as_map = match &self.value { + CborValue::Map(val) => val, + _ => { + return Err(MizeError::new().msg("Message was not a map")); + }, + }; + + let vec_value = { + let mut data = &CborValue::Null; + for (key, val) in msg_as_map { + if let CborValue::Integer(key_int) = key { + // number 2 indicates the id field + let two: Integer = 1.into(); + if key_int == &two { + data = val; + } + } + } + if data == &CborValue::Null { + return Err(MizeError::new().msg("error getting the data value form a msg")); + } + match data { + CborValue::Array(inner) => inner, + _ => { + return Err(MizeError::new().msg("Id is not a vector")); + }, + } + }; + + let mut vec_string: Vec = Vec::new(); + for val in vec_value { + match val { + CborValue::Text(text) => { + vec_string.push(text.to_owned()) + }, + _ => { + return Err(MizeError::new().msg("one in the vec of the id in a msg is not of type Text")); + }, + } + } + + let id = instance.id_from_vec_string(vec_string)?; + Ok(id) + } + + pub fn data(&mut self) -> MizeResult { + // return err, if msg is not a map + let msg_as_map = match &self.value { + CborValue::Map(val) => val, + _ => { + return Err(MizeError::new().msg("Message was not a map")); + }, + }; + + let data = { + let mut tmp_val = &CborValue::Null; + for (key, val) in msg_as_map { + if let CborValue::Integer(key_int) = key { + // number three indicates the data field + let three: Integer = 3.into(); + if key_int == &three { + tmp_val = val; + } + } + } + if tmp_val == &CborValue::Null { + return Err(MizeError::new().msg("error getting the data value form a msg")); + } + tmp_val + }; + + return Ok(data.to_owned().into_item_data()); + } +} + +pub enum MessageCmd { + Get, + Set, } diff --git a/src/platform/os/unix_socket.rs b/src/platform/os/unix_socket.rs index 97fb1eb..7fa0ae4 100644 --- a/src/platform/os/unix_socket.rs +++ b/src/platform/os/unix_socket.rs @@ -2,11 +2,12 @@ use std::fs; use std::path::{Path, PathBuf}; use tokio::net::UnixListener as TokioUnixListener; use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; -use tokio::io::{Interest, AsyncReadExt}; +use tokio::io::{Interest, AsyncReadExt, AsyncWriteExt}; use crossbeam::channel::{Receiver, Sender, unbounded}; use ciborium::Value as CborValue; +use tracing::warn; -use crate::instance::peer::{PeerListener, Peer}; +use crate::instance::connection::{ConnListener, Connection}; use crate::instance::{self, Instance}; use crate::error::{MizeError, MizeResult, IntoMizeResult}; use crate::proto::MizeMessage; @@ -21,7 +22,7 @@ impl UnixListener { } } -impl PeerListener for UnixListener { +impl ConnListener for UnixListener { fn listen(self, mut instance: Instance) -> MizeResult<()> { instance.spawn_async("unix listen async", unix_listen(self, instance.clone())); Ok(()) @@ -30,7 +31,9 @@ impl PeerListener for UnixListener { async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResult<()> { + // remove the file at sock_path if it already exists fs::remove_file(&listener.sock_path)?; + let listener = TokioUnixListener::bind(&listener.sock_path) .mize_result_msg(format!("Could not bind to unix socket at '{}'", listener.sock_path.display()))?; @@ -43,126 +46,87 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu let (unix_read, unix_write) = unix_sock.into_split(); - let peer = Peer::new(recieve_rx, send_tx); - - instance.add_peer(peer); + let conn_id = instance.new_connection(recieve_rx, send_tx)?; let cloned_instance = instance.clone(); - instance.spawn("incomming", || unix_incomming(unix_read, recieve_tx, cloned_instance)); - instance.spawn_async("outgoing", unix_outgoing(unix_write, send_rx)); + instance.spawn("incomming", move || { + let result = unix_incomming(unix_read, recieve_tx, cloned_instance, conn_id); + // if unix incomming fails, close the connection + if let Err(err) = result { + warn!("Connection closing") + } + Ok(()) + }); + + let outgoing_cloned_instance = instance.clone(); + instance.spawn("outgoing", move || { + let result = unix_outgoing(unix_write, send_rx, outgoing_cloned_instance, conn_id); + // if writing fails, close this connection + if let Err(err) = result { + warn!("Connection closing") + } + Ok(()) + }); } } -async fn unix_outgoing(unix_write: OwnedWriteHalf, send_rx: Receiver) -> MizeResult<()> { - return Ok(()); - loop { - let ready = unix_write.ready(Interest::WRITABLE).await.mize_result_msg("hi")?; - if ready.is_writable() { - // Try to write data, this may still fail with `WouldBlock` - // if the readiness event is a false positive. - match unix_write.try_write(b"hello world") { - Ok(n) => { - println!("wrote {} bytes to unix sock", n); - } - Err(ref e) if e.kind() == tokio::io::ErrorKind::WouldBlock => { - continue; - } - Err(e) => { - return Err(e.into()); - } - } - } +fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver, mut instance: Instance, conn_id: u64) -> MizeResult<()> { + for msg in send_rx { + let adapter = MyCiboriumWriter { inner: &mut unix_write, instance: &mut instance }; + let value = msg.value(); + ciborium::into_writer(&value, adapter)? } Ok(()) } -struct MyAdapter<'a> { - inner: &'a mut OwnedReadHalf, - instance: &'a mut Instance, -} - -impl<'a> ciborium_io::Read for MyAdapter<'a> { - type Error = MizeError; - fn read_exact(&mut self, data: &mut [u8]) -> Result<(), Self::Error> { - let runtime = self.instance.runtime.lock()?; - let num_read = runtime.block_on(self.inner.read_exact(data))?; - println!("read {} bytes", num_read); - Ok(()) - } -} - -fn unix_incomming(mut unix_read: OwnedReadHalf, recieve_tx: Sender, mut instance: Instance) -> MizeResult<()> { +fn unix_incomming(mut unix_read: OwnedReadHalf, recieve_tx: Sender, mut instance: Instance, conn_id: u64) -> MizeResult<()> { loop { - let adapter = MyAdapter { inner: &mut unix_read, instance: &mut instance }; + let adapter = MyCiboriumReader { inner: &mut unix_read, instance: &mut instance }; let value: CborValue = ciborium::from_reader(adapter)?; if let CborValue::Integer(_) = value { break; } - println!("value: {:?}", value); + let msg = MizeMessage::new(value, conn_id); + recieve_tx.send(msg)?; } Ok(()) } -/* +struct MyCiboriumWriter<'a> { + inner: &'a mut OwnedWriteHalf, + instance: &'a mut Instance, +} - let mut buf: Vec = vec![2,2,2]; - - let mut count: usize = 0; - - loop { - let num_bytes = unix_read.read(&mut buf[..]).await?; - let string = String::from_utf8(buf.clone())?; - println!("read num: {}", num_bytes); - println!("incoming: {}", string); - if num_bytes == 0 { - println!("zerooooooooooooooooooo"); - break - } - count += 1; - if count > 20 { - break - } +impl<'a> ciborium_io::Write for MyCiboriumWriter<'a> { + type Error = MizeError; + fn write_all(&mut self, data: &[u8]) -> Result<(), Self::Error> { + let runtime = self.instance.runtime.lock()?; + let num_read = runtime.block_on(self.inner.write_all(data))?; + Ok(()) } - println!("end of loop"); - - return Ok(()); - let mut count: usize = 0; - loop { - let num_bytes = unix_read.read(&mut buf[..]).await?; - let string = String::from_utf8(buf.clone())?; - println!("read num: {}", num_bytes); - println!("incoming: {}", string); - //if num_bytes == 0 { - //break - //} - count += 1; - if count > 20 { - break - } - } - return Ok(()); - - loop { - let ready = unix_read.ready(Interest::READABLE).await.mize_result_msg("unix_read_half.ready() failed")?; - - if ready.is_readable() { - let mut data = vec![0; 1024]; - // Try to read data, this may still fail with `WouldBlock` - // if the readiness event is a false positive. - match unix_read.try_read(&mut data) { - Ok(n) => { - println!("read {} bytes from unix sock", n); - } - Err(ref e) if e.kind() == tokio::io::ErrorKind::WouldBlock => { - continue; - } - Err(e) => { - return Err(e.into()); - } - } - - } + fn flush(&mut self) -> Result<(), Self::Error> { + Ok(()) } } -// */ + +struct MyCiboriumReader<'a> { + inner: &'a mut OwnedReadHalf, + instance: &'a mut Instance, +} + +impl<'a> ciborium_io::Read for MyCiboriumReader<'a> { + type Error = MizeError; + fn read_exact(&mut self, data: &mut [u8]) -> Result<(), Self::Error> { + let runtime = self.instance.runtime.lock()?; + let num_read = runtime.block_on(self.inner.read_exact(data))?; + + // if we read 0 bytes, that means the reader stream closed and we should terminate the + // connection + if num_read == 0 { + return Err(MizeError::new().msg("ReceiverStream closed")); + } + Ok(()) + } +} +