Asyncify and finish client server impl (for now)

Cargo.lock

@@ -780,6 +780,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "clap",
+ "once_cell",
  "protobuf",
  "smol",
  "uds",

Cargo.toml

@@ -12,3 +12,4 @@ zbus = "3.12.0"
 clap = { version = "4.2.4", features = ["derive"] }
 uds = "0.2.6"
 anyhow = "1.0.70"
+once_cell = "1.17.1"

src/args.rs

@@ -1,11 +1,15 @@
-use clap::Parser;
+use clap::{Parser, Subcommand};
 
-#[derive(Parser, Debug)]
+#[derive(Parser)]
 #[command(author, version, about, long_about = None)]
-pub struct Args {
-    #[arg(short, long)]
-    pub server: bool,
+#[command(propagate_version = true)]
+pub struct Cli {
+    #[command(subcommand)]
+    pub command: Commands,
+}
 
-    #[arg(short, long)]
-    pub client: bool
-}
+#[derive(Subcommand)]
+pub enum Commands {
+    Client { address: String },
+    Server,
+}

src/main.rs

@@ -1,55 +1,74 @@
 mod args;
-use anyhow::bail;
-use args::Args;
+
+use std::os::unix::net::{UnixListener as StdUnixListener, UnixStream as StdUnixStream};
+
+use args::{Cli, Commands};
 use clap::Parser;
-use uds::{UnixSeqpacketConn, UnixSeqpacketListener};
+use once_cell::sync::Lazy;
+use smol::channel::unbounded;
+use smol::io::{AsyncReadExt, AsyncWriteExt};
+use smol::net::unix::{UnixListener, UnixStream};
+use smol::Executor;
+use uds::{UnixListenerExt, UnixSocketAddr, UnixStreamExt};
 
-fn main() {
-    let args = Args::parse();
+static EXEC: Lazy<Executor> = Lazy::new(Executor::new);
 
-    let result = validate_arguments(&args);
-    match result {
-        Ok(_) => {}
-        Err(err) => println!("{err}"),
+fn main() -> anyhow::Result<()> {
+    let cli = Cli::parse();
+    let (_signal, shutdown) = unbounded::<()>();
+    for _ in 0..std::thread::available_parallelism().unwrap().into() {
+        std::thread::spawn({
+            let shutdown = shutdown.clone();
+            move || smol::future::block_on(EXEC.run(shutdown.recv()))
+        });
     }
+    smol::block_on(async {
+        match cli.command {
+            Commands::Server => server().await?,
+            Commands::Client { address } => {
+                let address = UnixSocketAddr::new(&address)?;
+                client(address).await?;
+            }
+        }
+        Ok(())
+    })
 }
 
-fn validate_arguments(args: &Args) -> anyhow::Result<()> {
-    if !args.server && !args.client {
-        bail!("Please provide either --client or --server");
+async fn server() -> anyhow::Result<()> {
+    let (listener, addr) = create_rings_socket()?;
+    println!("{}", addr);
+    loop {
+        match listener.accept().await {
+            Ok((mut conn, _)) => {
+                EXEC.spawn(async move {
+                    let mut buf = String::new();
+                    match conn.read_to_string(&mut buf).await {
+                        Ok(_) => println!("Got: {buf}"),
+                        Err(e) => eprintln!("{e}"),
+                    }
+                })
+                .detach();
+            }
+            Err(e) => {
+                eprintln!("{e}");
+            }
+        }
     }
+    // Ok(())
+}
 
-    if args.server && args.client {
-        bail!("You can't have both flags active at the same time");
-    }
-
-    if args.server {
-        server()?;
-    } else if args.client {
-        client(todo!())?;
-    }
+async fn client(addr: uds::UnixSocketAddr) -> anyhow::Result<()> {
+    let mut stream: UnixStream = smol::unblock(move || StdUnixStream::connect_to_unix_addr(&addr))
+        .await?
+        .try_into()?;
+    let msg = "Saturn has cool rings uwu";
+    stream.write_all(msg.as_bytes()).await?;
     Ok(())
 }
 
-fn server() -> anyhow::Result<()> {
-    create_rings_socket()?;
-    Ok(())
-}
-
-fn client(addr: &uds::UnixSocketAddr) -> anyhow::Result<UnixSeqpacketConn> {
-    let client = uds::UnixSeqpacketConn::connect_unix_addr(addr);
-    match client {
-        Ok(_client) => {
-            todo!()
-        }
-        Err(_) => {
-            todo!()
-        }
-    }
-}
-
-fn create_rings_socket() -> anyhow::Result<UnixSeqpacketListener> {
-    let addr = uds::UnixSocketAddr::new("@abstract")?;
-    let listener = UnixSeqpacketListener::bind_unix_addr(&addr).expect("create seqpacket listener");
-    Ok(listener)
+fn create_rings_socket() -> anyhow::Result<(UnixListener, UnixSocketAddr)> {
+    let addr = uds::UnixSocketAddr::new_unspecified();
+    let listener = StdUnixListener::bind_unix_addr(&addr)?;
+    let addr = listener.local_unix_addr()?;
+    Ok((listener.try_into()?, addr))
 }