Socket communication with Tokio

3 years ago · 37eb023104
3 changed files with 297 additions and 0 deletions
--- a/tokiounixsocket/Cargo.lock
+++ b/tokiounixsocket/Cargo.lock
@ -0,0 +1,191 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 3
+
+[[package]]
+name = "bincode"
+version = "1.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "cfg-if"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
+
+[[package]]
+name = "libc"
+version = "0.2.119"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1bf2e165bb3457c8e098ea76f3e3bc9db55f87aa90d52d0e6be741470916aaa4"
+
+[[package]]
+name = "log"
+version = "0.4.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51b9bbe6c47d51fc3e1a9b945965946b4c44142ab8792c50835a980d362c2710"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
+name = "mio"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba272f85fa0b41fc91872be579b3bbe0f56b792aa361a380eb669469f68dafb2"
+dependencies = [
+ "libc",
+ "log",
+ "miow",
+ "ntapi",
+ "winapi",
+]
+
+[[package]]
+name = "miow"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21"
+dependencies = [
+ "winapi",
+]
+
+[[package]]
+name = "ntapi"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c28774a7fd2fbb4f0babd8237ce554b73af68021b5f695a3cebd6c59bac0980f"
+dependencies = [
+ "winapi",
+]
+
+[[package]]
+name = "pin-project-lite"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e280fbe77cc62c91527259e9442153f4688736748d24660126286329742b4c6c"
+
+[[package]]
+name = "proc-macro2"
+version = "1.0.36"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7342d5883fbccae1cc37a2353b09c87c9b0f3afd73f5fb9bba687a1f733b029"
+dependencies = [
+ "unicode-xid",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b4af2ec4714533fcdf07e886f17025ace8b997b9ce51204ee69b6da831c3da57"
+dependencies = [
+ "proc-macro2",
+]
+
+[[package]]
+name = "serde"
+version = "1.0.136"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce31e24b01e1e524df96f1c2fdd054405f8d7376249a5110886fb4b658484789"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.136"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08597e7152fcd306f41838ed3e37be9eaeed2b61c42e2117266a554fab4662f9"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "socket2"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "66d72b759436ae32898a2af0a14218dbf55efde3feeb170eb623637db85ee1e0"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
+[[package]]
+name = "syn"
+version = "1.0.89"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ea297be220d52398dcc07ce15a209fce436d361735ac1db700cab3b6cdfb9f54"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-xid",
+]
+
+[[package]]
+name = "tokio"
+version = "1.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2af73ac49756f3f7c01172e34a23e5d0216f6c32333757c2c61feb2bbff5a5ee"
+dependencies = [
+ "libc",
+ "mio",
+ "pin-project-lite",
+ "socket2",
+ "tokio-macros",
+ "winapi",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b557f72f448c511a979e2564e55d74e6c4432fc96ff4f6241bc6bded342643b7"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "tokiounixsocket"
+version = "0.1.0"
+dependencies = [
+ "bincode",
+ "serde",
+ "tokio",
+]
+
+[[package]]
+name = "unicode-xid"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8ccb82d61f80a663efe1f787a51b16b5a51e3314d6ac365b08639f52387b33f3"
+
+[[package]]
+name = "winapi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
+dependencies = [
+ "winapi-i686-pc-windows-gnu",
+ "winapi-x86_64-pc-windows-gnu",
+]
+
+[[package]]
+name = "winapi-i686-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
+
+[[package]]
+name = "winapi-x86_64-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
--- a/tokiounixsocket/Cargo.toml
+++ b/tokiounixsocket/Cargo.toml
@ -0,0 +1,11 @@
+[package]
+name = "tokiounixsocket"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+bincode = "1.3.3"
+serde = { version = "1.0.136", features = ["derive"] }
+tokio = { version = "1.17.0", features = ["rt", "sync", "net", "macros"] }
--- a/tokiounixsocket/src/main.rs
+++ b/tokiounixsocket/src/main.rs
@ -0,0 +1,95 @@
+use std::io;
+use std::path::Path;
+
+use tokio::io::Interest;
+use tokio::net::UnixListener;
+use tokio::net::UnixStream;
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+enum Message {
+    First(i64, String),
+    Second(i64, String),
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    let path = Path::new("./comm.socket");
+
+    // producer
+    let path_prod = path.to_path_buf();
+    let producer = tokio::spawn(async move {
+        let listener = UnixListener::bind(&path_prod).unwrap();
+        loop {
+            match listener.accept().await {
+                Ok((stream, _addr)) => {
+                    println!("Client here");
+                    tokio::spawn(async move { send_to_consumer(&stream).await })
+                }
+                Err(e) => {
+                    println!("Erro! {:?}", e);
+                    break;
+                }
+            };
+        }
+    });
+
+    // consumer
+    let path_cons = path.to_path_buf();
+    let consumer = tokio::spawn(async move {
+        let socket = UnixStream::connect(&path_cons).await.unwrap();
+        loop {
+            let ready = socket.ready(Interest::READABLE).await.unwrap();
+            if ready.is_readable() {
+                let mut data = [0; 1024];
+                match socket.try_read(&mut data) {
+                    Ok(n) => {
+                        println!("Read {} bytes", n);
+                        let message: Result<Message, Box<bincode::ErrorKind>> =
+                            bincode::deserialize(&data[..n]);
+                        match message {
+                            Ok(the_message) => println!("Message: {:?}", the_message),
+                            Err(e) => println!("Deserialized message: {:?}", e),
+                        }
+                    }
+                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                        continue;
+                    }
+                    Err(e) => {
+                        println!("Error reading: {:?}", e);
+                    }
+                }
+            }
+        }
+    });
+
+    // and... WAIT!
+    let _ = tokio::join!(producer, consumer);
+}
+
+async fn send_to_consumer(stream: &UnixStream) {
+    let mut should_yield = false;
+    let mut seq = 0;
+    loop {
+        let message = if should_yield {
+            Message::First(seq, format!("{} first", seq))
+        } else {
+            Message::Second(seq, format!("{} second", seq))
+        };
+        seq += 1;
+
+        let encoded = bincode::serialize(&message).unwrap();
+
+        let ready = stream.ready(Interest::WRITABLE).await.unwrap();
+        if ready.is_writable() {
+            match stream.try_write(&encoded) {
+                Ok(n) => println!("Wrote {} bytes: {:?}", n, encoded),
+                Err(e) => println!("Error: {:?}", e),
+            }
+
+            should_yield = !should_yield;
+            if should_yield {
+                tokio::task::yield_now().await;
+            }
+        }
+    }
+}