we connect and get the namespace of the peer

src/core/instance/connection.rs

@@ -12,7 +12,6 @@ use crate::item::{get_raw_from_cbor, Item};
 // only created with Instance::new_connection()
 #[derive(Clone)]
 pub struct Connection {
-    pub rx: Receiver<MizeMessage>,
     pub tx: Sender<MizeMessage>,
     pub id: u64,
     pub ns: Option<Namespace>,
@@ -46,13 +45,15 @@ pub fn value_raw_from_peer(item: &mut Item) -> MizeResult<Vec<u8>> {
     let skipped = new_id_str.skip(4);
     let new_id = item.instance.new_id(skipped.collect::<Vec<&String>>())?;
 
+    println!("value_raw_from_peer hereeeeeeeeeee");
     let connection = item.instance.get_connection(conn_id)?;
+    println!("value_raw_from_peer afteeeeerrrrrrrrrrrrrrrr");
 
-    let msg = MizeMessage::new_get(new_id, connection.id);
+    let msg = MizeMessage::new_get(new_id.clone(), connection.id);
 
     connection.tx.send(msg)?;
 
-    let data = item.instance.give_msg_wait(item.id())?;
+    let data = item.instance.give_msg_wait(new_id)?;
 
     let raw_data = get_raw_from_cbor(data.cbor(), vec![])?;
     return Ok(raw_data.to_owned());

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::{self, Receiver, Sender};
+use crossbeam::channel::{self, bounded, Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use tracing::{trace, debug, info, warn, error};
 use uuid::Uuid;
@@ -23,6 +23,7 @@ use crate::memstore::MemStore;
 use crate::instance::updater::Operation;
 use crate::mize_err;
 use crate::proto::MizeMessage;
+use tokio::runtime::Handle;
 
 use self::connection::{ConnListener, Connection};
 
@@ -56,7 +57,7 @@ pub struct Instance {
     threads: Vec<String>,
     #[cfg(feature = "async")]
     pub runtime: Arc<Mutex<Runtime>>,
-    give_msg_wait: Arc<Mutex<HashMap<MizeId, (Sender<ItemData>, Receiver<ItemData>)>>>,
+    give_msg_wait: Arc<Mutex<HashMap<MizeId, Vec<Sender<ItemData>>>>>,
 }
 
 pub struct InstanceRef {
@@ -213,14 +214,15 @@ impl Instance {
         Ok(())
     }
 
-    pub fn new_connection(&mut self, rx: Receiver<MizeMessage>, tx: Sender<MizeMessage>) -> MizeResult<u64> {
+    pub fn new_connection(&mut 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;
 
-        let connection = Connection { id: next_con_id.to_owned(), rx, tx, ns: None};
+        let connection = Connection { id: next_con_id.to_owned(), tx, ns: None};
         conn_inner.push(connection);
         *next_con_id += 1;
-        Ok(next_con_id.to_owned())
+        Ok(old_next_con_id)
     }
 
     pub fn connection_set_namespace(&mut self, conn_id: u64, namespace: Namespace) -> MizeResult<()> {
@@ -230,6 +232,10 @@ impl Instance {
         Ok(())
     }
 
+    pub(crate) fn got_msg(&mut self, msg: MizeMessage) -> MizeResult<()> {
+        Ok(self.op_tx.send(Operation::Msg(msg))?)
+    }
+
     fn set_connection(&mut self, conn_id: u64, new_connection: Connection) -> MizeResult<()> {
         let mut conn_inner = self.connections.lock()?;
 
@@ -248,6 +254,7 @@ impl Instance {
         let mut conn_inner = self.connections.lock()?;
 
         for connection in conn_inner.iter() {
+            println!("get_connection iterrrrrrrrrrrrrr");
             if connection.id == conn_id {
                 return Ok(connection.clone());
             }
@@ -266,18 +273,20 @@ impl Instance {
 
         let mut give_msg_wait_inner = self.give_msg_wait.lock()?;
 
-        let rx = match give_msg_wait_inner.get(&id) {
+        let (tx, rx) = bounded::<ItemData>(1);
-            // if this id is already in the process of being requested, just use the rx already there
-            Some((tx, rx)) => rx.to_owned(),
 
+        let vec = match give_msg_wait_inner.get_mut(&id) {
+            Some(vec) => vec,
             None => {
-                let (tx, rx) = crossbeam::channel::bounded::<ItemData>(1);
+                give_msg_wait_inner.insert(id.clone(), Vec::new());
-                give_msg_wait_inner.insert(id, (tx, rx.clone()));
+                give_msg_wait_inner.get_mut(&id).unwrap()
-                rx
+            }
-            },
         };
 
+        vec.push(tx);
+
         // so that another thread can also give_msg_wait(), while we wait in the recv() of rx
+        drop(vec);
         drop(give_msg_wait_inner);
 
         let data = rx.recv()?;
@@ -291,15 +300,27 @@ impl Instance {
         let runtime_inner = self.runtime.lock().unwrap();
         runtime_inner.spawn(func);
     }
+    #[cfg(feature = "async")]
+    pub fn async_get_handle(&mut self) -> Handle {
+        let runtime_inner = self.runtime.lock().unwrap();
+        let handle = runtime_inner.handle().to_owned();
+        handle
+    }
 
     #[cfg(feature = "async")]
-    pub fn spawn_async_blocking<F: Future<Output = impl Send + Sync + 'static> + Send + Sync + 'static>(&mut self, name: &str, func: F) {
+    pub fn spawn_async_blocking<F: Future<Output = impl Send + Sync + 'static> + Send + Sync + 'static>(&mut self, name: &str, func: F) -> F::Output {
+        use std::process::Output;
+
         self.threads.push(name.to_owned());
         let runtime_inner = self.runtime.lock().unwrap();
-        runtime_inner.block_on(func);
+        let handle = runtime_inner.handle().to_owned();
+        drop(runtime_inner);
+        let result = handle.block_on(func);
+        return result;
     }
 
     pub fn wait(&self) {
+        info!("Instance main thread waiting");
         loop {
             thread::sleep_ms(10000000)
         }

src/core/instance/updater.rs

@@ -1,5 +1,5 @@
 use crossbeam::channel::Receiver;
-use tracing::{trace, error};
+use tracing::{error, trace, warn};
 
 use crate::mize_err;
 use crate::{instance::Instance, item::ItemData};
@@ -23,6 +23,7 @@ pub fn updater_thread(operation_rx : Receiver<Operation>, mut instance: Instance
 
         if let Err(err) = result {
             error!("OPERATION FAILED: {:?}", operation);
+            err.log();
         }
     }
     Ok(())
@@ -57,6 +58,18 @@ fn handle_msg(msg: &mut MizeMessage, instance: &mut Instance) -> MizeResult<()>
             let id = msg.id(instance)?;
             instance.set(id, data);
         },
+        MessageCmd::Give => {
+            let id = msg.id(instance)?;
+            let data = msg.data()?;
+            let give_msg_wait_inner = instance.give_msg_wait.lock()?;
+            if let Some(vec) = give_msg_wait_inner.get(&id) {
+                for tx in vec {
+                    tx.send(data.clone());
+                }
+            } else {
+                warn!("got give msg for id '{}', that has no one waiting for it", id);
+            }
+        },
         _ => {
             return Err(mize_err!("got a message, that is not handeled"));
         }

src/core/proto/mod.rs

@@ -81,6 +81,7 @@ impl MizeMessage {
         let cmd = match val_of_c {
             1 => MessageCmd::Get,
             2 => MessageCmd::Set,
+            3 => MessageCmd::Give,
             _ => {
                 return Err(MizeError::new().msg("error cmd of msg was not a valid command"));
             },
@@ -104,7 +105,7 @@ impl MizeMessage {
             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();
+                    let two: Integer = 2.into();
                     if key_int == &two {
                         data = val;
                     }
@@ -170,6 +171,7 @@ impl MizeMessage {
 pub enum MessageCmd {
     Get,
     Set,
+    Give,
 }
 
 

src/platform/os/fsstore.rs

@@ -41,12 +41,13 @@ impl FileStore {
         }
 
         // init the store
-        fs::write(path.join("next_id"), "0");
+        fs::write(path.join("next_id"), "1");
 
         Ok(FileStore { path: Path::new(&path).to_owned() })
     }
 
     pub fn store_is_opened(store_path: String) -> MizeResult<bool> {
+        println!("valid_pid_file: {}", valid_pid_file(Path::new(&store_path))?.is_some());
         Ok(valid_pid_file(Path::new(&store_path))?.is_some())
     }
 }

src/platform/os/unix_socket.rs

@@ -25,22 +25,8 @@ impl UnixListener {
 
 
 pub fn connect(instance: &mut Instance, store_path: PathBuf) -> MizeResult<()> {
-    instance.spawn_async_blocking("unix_connection", connect_async(instance.clone(), store_path) );
+    let conn_id = instance.spawn_async_blocking("unix_connection", connect_async(instance.clone(), store_path) )?;
 
-    Ok(())
-}
-
-
-async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResult<()> {
-    println!("we are never hereeeeeeeeeeeeee");
-    let mut stream = UnixStream::connect(store_path.join("sock")).await?;
-    let (read_half, write_half) = stream.split();
-    println!("we are never hoooooooooooooooooooooooooooooo");
-
-    let (recieve_tx, recieve_rx) = unbounded::<MizeMessage>();
-    let (send_tx, send_rx) = unbounded::<MizeMessage>();
-    
-    let conn_id = instance.new_connection(recieve_rx, send_tx)?;
     let ns_of_peer = instance.get(format!("inst/con_by_id/{}/peer/0/config/namespace", conn_id))?.value_string()?;
 
     info!("connect_async ... ns_of_peer: {}", ns_of_peer);
@@ -51,6 +37,39 @@ async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResul
 }
 
 
+async fn connect_async(mut instance: Instance, store_path: PathBuf) -> MizeResult<u64> {
+    let mut stream = UnixStream::connect(store_path.join("sock")).await?;
+    let (unix_read, unix_write) = stream.into_split();
+
+    let (send_tx, send_rx) = unbounded::<MizeMessage>();
+    
+    let conn_id = instance.new_connection(send_tx)?;
+
+
+    let cloned_instance = instance.clone();
+    instance.spawn("incomming", move || {
+        let result = unix_incomming(unix_read, cloned_instance, conn_id);
+        // if unix incomming fails, close the connection
+        if let Err(err) = result {
+            warn!("Connection closing because of EOF");
+        }
+        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 because of EOF");
+        }
+        Ok(())
+    });
+
+    return Ok(conn_id);
+}
+
+
 impl ConnListener for UnixListener {
     fn listen(self, mut instance: Instance) -> MizeResult<()> {
         instance.spawn_async("unix listen async", unix_listen(self, instance.clone()));
@@ -71,18 +90,17 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu
             .mize_result_msg("Error while accepting Unix sock connection")?;
         info!("new connection");
 
-        let (recieve_tx, recieve_rx) = unbounded::<MizeMessage>();
         let (send_tx, send_rx) = unbounded::<MizeMessage>();
         let (unix_read, unix_write) = unix_sock.into_split();
 
 
-        let conn_id = instance.new_connection(recieve_rx, send_tx)?;
+        let conn_id = instance.new_connection(send_tx)?;
         let cloned_instance = instance.clone();
         instance.spawn("incomming", move || {
-            let result = unix_incomming(unix_read, recieve_tx, cloned_instance, conn_id);
+            let result = unix_incomming(unix_read, cloned_instance, conn_id);
             // if unix incomming fails, close the connection
             if let Err(err) = result {
-                warn!("Connection closing")
+                warn!("Connection closing, {:?}", err);
             }
             Ok(())
         });
@@ -92,7 +110,7 @@ async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResu
             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")
+                warn!("Connection closing, {:?}", err);
             }
             Ok(())
         });
@@ -102,6 +120,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);
         let adapter = MyCiboriumWriter { inner: &mut unix_write, instance: &mut instance };
         let value = msg.value();
         ciborium::into_writer(&value, adapter)?
@@ -109,15 +128,17 @@ fn unix_outgoing(mut unix_write: OwnedWriteHalf, send_rx: Receiver<MizeMessage>,
     Ok(())
 }
 
-fn unix_incomming(mut unix_read: OwnedReadHalf, recieve_tx: Sender<MizeMessage>, mut instance: Instance, conn_id: u64) -> MizeResult<()> {
+fn unix_incomming(mut unix_read: OwnedReadHalf, mut instance: Instance, conn_id: u64) -> MizeResult<()> {
     loop {
+        println!("unix incoming loop...");
         let adapter = MyCiboriumReader { inner: &mut unix_read, instance: &mut instance };
         let value: CborValue = ciborium::from_reader(adapter)?;
         if let CborValue::Integer(_) = value {
             break;
         }
         let msg = MizeMessage::new(value, conn_id);
-        recieve_tx.send(msg)?;
+        println!("unix incoming got msg: {:?}", msg);
+        instance.got_msg(msg);
     }
     Ok(())
 }
@@ -131,11 +152,17 @@ struct MyCiboriumWriter<'a> {
 impl<'a> ciborium_io::Write for MyCiboriumWriter<'a> {
     type Error = MizeError;
     fn write_all(&mut self, data: &[u8]) -> Result<(), Self::Error> {
+        println!("CALL: write_all");
         let runtime = self.instance.runtime.lock()?;
+        println!("after runtime");
         let num_read = runtime.block_on(self.inner.write_all(data))?;
         Ok(())
     }
     fn flush(&mut self) -> Result<(), Self::Error> {
+        println!("CALL flush");
+        let handle = self.instance.async_get_handle();
+        let result = handle.block_on(self.inner.flush());
+        println!("flush result: {:?}", result);
         Ok(())
     }
 }
@@ -145,17 +172,37 @@ struct MyCiboriumReader<'a> {
     instance: &'a mut Instance,
 }
 
+async fn test(test: &mut OwnedReadHalf, buf: &mut [u8]) {
+    println!("hello from async...");
+    println!("len: {}", buf.len());
+    let num_read = test.read_u8().await;
+    println!("num_read: {:?}", num_read);
+    //test.read_exact(buf).await;
+    println!("hello from async... twoooooooooooooo");
+}
+
 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()?;
+        //println!("data before: {:?}", data);
-        let num_read = runtime.block_on(self.inner.read_exact(data))?;
+        //let runtime = self.instance.runtime.lock()?;
+        //println!("are we here after runtime lock()");
+        //runtime.block_on(test(self.inner, data));
+        //println!("are we hereeeeeeeeeeeeeeeeeeee");
+
+        //return Ok(());
+
+        println!("incoming before block_on");
+        let handle = self.instance.async_get_handle();
+        let num_read = handle.block_on(self.inner.read_exact(data))?;
+        println!("incoming num_read: {:?}", num_read);
+        //println!("data after: {:?}", data);
 
         // if we read 0 bytes, that means the reader stream closed and we should terminate the
         // connection
-        if num_read == 0 {
+        //if num_read == 0 {
-            return Err(MizeError::new().msg("ReceiverStream closed"));
+            //return Err(MizeError::new().msg("ReceiverStream closed"));
-        }
+        //}
         Ok(())
     }
 }