weired implementation of stream parsing cbor items from a unix sock, but it works

Cargo.toml

@@ -18,7 +18,7 @@ required-features = [ "os-target" ]
 
 [features]
 os-target = [ "clap", "home", "daemonize", "nix", "toml", "serde_json", "sysinfo", "ciborium/default", "tracing-subscriber", "async", "tracing-subscriber/env-filter"]
-async = ["tokio/net", "tokio", "tokio/rt-multi-thread"]
+async = ["tokio/net", "tokio", "tokio/rt-multi-thread", "tokio/io-util"]
 wasm-target = [ "wasm-bindgen", "console_error_panic_hook", "wasm-bindgen-cli" ]
 
 
@@ -55,6 +55,9 @@ axum-extra = { version = "0.4.2", features = ["spa"], optional = true }
 axum-macros = { version = "0.3.0", optional = true }
 crossbeam = "0.8.4"
 dyn-clone = "1.0.17"
+tokio-core = "0.1.18"
+tokio-io = "0.1.13"
+ciborium-io = "0.2.2"
 
 #tokio-stream = { version = "0.1.9", optional = true }
 #derive_more = "0.99.17"

src/core/instance/mod.rs

@@ -1,6 +1,7 @@
 use core::fmt;
 use std::collections::HashMap;
 use std::fs::create_dir;
+use std::thread::JoinHandle;
 use std::{thread, vec};
 use crossbeam::channel;
 use std::sync::{Arc, Mutex};
@@ -51,7 +52,7 @@ pub struct Instance {
     context: Vec<MizeId>,
     threads: Vec<String>,
     #[cfg(feature = "async")]
-    runtime: Arc<Mutex<Runtime>>,
+    pub runtime: Arc<Mutex<Runtime>>,
 }
 
 pub struct InstanceRef {
@@ -218,6 +219,12 @@ impl Instance {
         let runtime_inner = self.runtime.lock().unwrap();
         runtime_inner.spawn(func);
     }
+
+    pub fn wait(&self) {
+        loop {
+            thread::sleep_ms(10000000)
+        }
+    }
 }
 
 impl fmt::Debug for Instance {

src/platform/os/cli/run.rs

@@ -8,17 +8,16 @@ use nix::sys::signal::{self, Signal};
 
 use mize::error::{MizeResultTrait, IntoMizeResult, MizeError, MizeResult};
 use mize::instance::Instance;
+use mize::platform::os::config_from_cli_args;
 
 
 pub fn run(sub_matches: &ArgMatches) -> MizeResult<()> {
 
-    // TODO NEXT!!!
-    // - --config arg
-    // - --folder
-    // - to set a location, where there will be a /store and a /sock
-    // - listen for connections on that sock
-    // - msg handeling
-    let instance = Instance::with_config(todo!());
+    let instance = Instance::with_config(config_from_cli_args(sub_matches)?)?;
+
+    instance.wait();
+
+
 
     /*
     let instance_path = match sub_matches.get_one::<String>("store") {

src/platform/os/main.rs

@@ -55,7 +55,7 @@ fn main() {
     // match command
     let result = match cli_matches.subcommand() {
         // mi daemon
-        Some(("daemon", sub_matches)) => cli::run(sub_matches),
+        Some(("run", sub_matches)) => cli::run(sub_matches),
 
         // mi serve
         Some(("serve", sub_matches)) => cli::stop(sub_matches),

src/platform/os/unix_socket.rs

@@ -1,8 +1,10 @@
+use std::fs;
 use std::path::{Path, PathBuf};
 use tokio::net::UnixListener as TokioUnixListener;
 use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
-use tokio::io::Interest;
+use tokio::io::{Interest, AsyncReadExt};
 use crossbeam::channel::{Receiver, Sender, unbounded};
+use ciborium::Value as CborValue;
 
 use crate::instance::peer::{PeerListener, Peer};
 use crate::instance::{self, Instance};
@@ -28,27 +30,31 @@ impl PeerListener for UnixListener {
 
 
 async fn unix_listen(listener: UnixListener, mut instance: Instance) -> MizeResult<()> {
+    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()))?;
 
     loop {
-        let (unix_sock, addr) = listener.accept().await
+        let (mut unix_sock, addr) = listener.accept().await
             .mize_result_msg("Error while accepting Unix sock 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 peer = Peer::new(recieve_rx, send_tx);
 
         instance.add_peer(peer);
-        instance.spawn_async("incomming", unix_incomming(unix_read, recieve_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));
     }
 }
 
 
 async fn unix_outgoing(unix_write: OwnedWriteHalf, send_rx: Receiver<MizeMessage>) -> MizeResult<()> {
+    return Ok(());
     loop {
         let ready = unix_write.ready(Interest::WRITABLE).await.mize_result_msg("hi")?;
         if ready.is_writable() {
@@ -67,9 +73,76 @@ async fn unix_outgoing(unix_write: OwnedWriteHalf, send_rx: Receiver<MizeMessage
             }
         }
     }
+    Ok(())
 }
 
-async fn unix_incomming(unix_read: OwnedReadHalf, recieve_tx: Sender<MizeMessage>) -> MizeResult<()> {
+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<MizeMessage>, mut instance: Instance) -> MizeResult<()> {
+    loop {
+        let adapter = MyAdapter { inner: &mut unix_read, instance: &mut instance };
+        let value: CborValue = ciborium::from_reader(adapter)?;
+        if let CborValue::Integer(_) = value {
+            break;
+        }
+        println!("value: {:?}", value);
+    }
+    Ok(())
+}
+
+/*
+
+
+    let mut buf: Vec<u8> = 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
+        }
+    }
+    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")?;
 
@@ -92,3 +165,4 @@ async fn unix_incomming(unix_read: OwnedReadHalf, recieve_tx: Sender<MizeMessage
         }
     }
 }
+// */