we can recieve and send messages on a unix sock

src/core/instance/connection.rs

@@ -6,18 +6,20 @@ use crate::error::{MizeError, MizeResult, IntoMizeResult};
 
 use super::Instance;
 
-pub struct Peer {
+#[derive(Clone)]
-    rx: Receiver<MizeMessage>,
+pub struct Connection {
-    tx: Sender<MizeMessage>,
+    pub rx: Receiver<MizeMessage>,
+    pub tx: Sender<MizeMessage>,
+    pub id: u64,
 }
 
-pub trait PeerListener : Send + Sync {
+pub trait ConnListener : Send + Sync {
     fn listen(self, instance: Instance) -> MizeResult<()>;
 }
 
-impl Peer {
+impl Connection {
-    pub fn new(rx: Receiver<MizeMessage>, tx: Sender<MizeMessage>) -> Peer {
+    pub fn send(&mut self, msg: MizeMessage) -> MizeResult<()> {
-        Peer { rx , tx }
+        Ok(self.tx.send(msg)?)
     }
 }
 

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<dyn Store>,
-    peers: Arc<Mutex<Vec<Peer>>>,
+    connections: Arc<Mutex<Vec<Connection>>>,
+    next_con_id: Arc<Mutex<u64>>,
     subs: Arc<Mutex<HashMap<MizeId, Subscription>>>,
     pub id_pool: Arc<Mutex<VecStringPool>>,
     pub namespace_pool: Arc<Mutex<StringPool>>,
@@ -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<T: PeerListener +'static>(&mut self, listener: T) -> MizeResult<()> {
+    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));
         Ok(())
     }
 
-    pub fn add_peer(&mut self, peer: Peer) -> MizeResult<()> {
+    pub fn new_connection(&mut self, rx: Receiver<MizeMessage>, tx: Sender<MizeMessage>) -> MizeResult<u64> {
-        let mut peer_inner = self.peers.lock()?;
+        let mut conn_inner = self.connections.lock()?;
-        peer_inner.push(peer);
+        let mut next_con_id = self.next_con_id.lock()?;
-        Ok(())
+
+        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<Connection> {
+        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<()> {

src/core/instance/msg_thread.rs

@@ -0,0 +1,2 @@
+
+

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<Operation>, mut instance: Instance) -> MizeResult<()> {
@@ -20,7 +25,7 @@ pub fn updater_thread(operation_rx : Receiver<Operation>, 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(())
 }
+
+
+

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<MessageCmd> {
+        // 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<MizeId> {
+        // 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<String> = 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<ItemData> {
+        // 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,
 }
 
 

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);
+        let conn_id = instance.new_connection(recieve_rx, send_tx)?;
-
-        instance.add_peer(peer);
         let cloned_instance = instance.clone();
-        instance.spawn("incomming", || unix_incomming(unix_read, recieve_tx, cloned_instance));
+        instance.spawn("incomming", move || {
-        instance.spawn_async("outgoing", unix_outgoing(unix_write, send_rx));
+            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<MizeMessage>) -> MizeResult<()> {
+fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver<MizeMessage>, mut instance: Instance, conn_id: u64) -> MizeResult<()> {
-    return Ok(());
+    for msg in send_rx {
-    loop {
+        let adapter = MyCiboriumWriter { inner: &mut unix_write, instance: &mut instance };
-        let ready = unix_write.ready(Interest::WRITABLE).await.mize_result_msg("hi")?;
+        let value = msg.value();
-        if ready.is_writable() {
+        ciborium::into_writer(&value, adapter)?
-            // 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());
-                }
-            }
-        }
     }
     Ok(())
 }
 
-struct MyAdapter<'a> {
+fn unix_incomming(mut unix_read: OwnedReadHalf, recieve_tx: Sender<MizeMessage>, mut instance: Instance, conn_id: u64) -> MizeResult<()> {
-    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<MizeMessage>, mut instance: Instance) -> 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<u8> = vec![2,2,2];
+impl<'a> ciborium_io::Write for MyCiboriumWriter<'a> {
-
+    type Error = MizeError;
-    let mut count: usize = 0;
+    fn write_all(&mut self, data: &[u8]) -> Result<(), Self::Error> {
-
+        let runtime = self.instance.runtime.lock()?;
-    loop {
+        let num_read = runtime.block_on(self.inner.write_all(data))?;
-        let num_bytes = unix_read.read(&mut buf[..]).await?;
+        Ok(())
-        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
-        }
     }
-    println!("end of loop");
+    fn flush(&mut self) -> Result<(), Self::Error> {
-
+        Ok(())
-    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());
-                }
-            }
-
-        }
     }
 }
-// */
+
+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(())
+    }
+}
+