From 37eb023104b0c9f4b0382adc970b3295c8496d76 Mon Sep 17 00:00:00 2001 From: Julio Biason Date: Mon, 21 Mar 2022 12:00:37 -0300 Subject: [PATCH] Socket communication with Tokio --- tokiounixsocket/Cargo.lock | 191 ++++++++++++++++++++++++++++++++++++ tokiounixsocket/Cargo.toml | 11 +++ tokiounixsocket/src/main.rs | 95 ++++++++++++++++++ 3 files changed, 297 insertions(+) create mode 100644 tokiounixsocket/Cargo.lock create mode 100644 tokiounixsocket/Cargo.toml create mode 100644 tokiounixsocket/src/main.rs diff --git a/tokiounixsocket/Cargo.lock b/tokiounixsocket/Cargo.lock new file mode 100644 index 0000000..4df5c93 --- /dev/null +++ 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" diff --git a/tokiounixsocket/Cargo.toml b/tokiounixsocket/Cargo.toml new file mode 100644 index 0000000..287a8e7 --- /dev/null +++ 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"] } diff --git a/tokiounixsocket/src/main.rs b/tokiounixsocket/src/main.rs new file mode 100644 index 0000000..0662d31 --- /dev/null +++ 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> = + 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; + } + } + } +}