Add code to create a client connected to the network

.gitea/workflows/build.yaml

@@ -7,7 +7,11 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v3
-      - name: Run clippy
+      - name: Run clippy (tcp feature)
-        run: nix develop . -c cargo clippy
+        run: nix develop . -c cargo clippy --no-default-features --features tcp
-      - name: Build & test
+      - name: Run clippy (unix feature)
-        run: nix build
+        run: nix develop . -c cargo clippy --no-default-features --features unix
+      - name: Build & test (tcp feature)
+        run: nix build .#tcp
+      - name: Build & test (unix feature)
+        run: nix build .#unix

Cargo.toml

@@ -4,6 +4,12 @@ version = "0.2.0"
 edition = "2021"
 license = "AGPL-3.0"
 publish = ["gitea"]
+resolver = "2"
+
+[features]
+default = ["tcp"]
+tcp =  ["tokio/net"]
+unix = ["tokio/net"]
 
 [dependencies]
 proc-macro2 = "1.0.85"
@@ -12,7 +18,9 @@ rand = "0.8.5"
 ron = "0.8.1"
 serde = { version = "1.0.203", features = ["serde_derive"] }
 syn = "2.0.66"
-# TODO: rt and macros should be removed unless we do tests
+tokio = { version = "1.38.0", features = ["sync", "io-util"] }
+
+[dev-dependencies]
 tokio = { version = "1.38.0", features = ["sync", "rt-multi-thread", "macros", "time", "io-util", "net"] }
 
 [lib]

flake.nix

@@ -14,9 +14,21 @@
         naersk-lib = pkgs.callPackage naersk {};
       in
       {
-        defaultPackage = naersk-lib.buildPackage {
+        packages = {
-          src = ./.;
+          default = naersk-lib.buildPackage {
-          doCheck = true;
+            src = ./.;
+            doCheck = true;
+          };
+          unix = naersk-lib.buildPackage {
+            src = ./.;
+            doCheck = true;
+            cargoOptions = x: x ++ ["--no-default-features" "--features" "unix"];
+          };
+          tcp = naersk-lib.buildPackage {
+            src = ./.;
+            doCheck = true;
+            cargoOptions = x: x ++ ["--no-default-features" "--features" "tcp"];
+          };
         };        
         devShell = with pkgs; mkShell {
           nativeBuildInputs = with pkgs; [rustc cargo rustfmt pre-commit clippy cargo-expand];

src/lib.rs

@@ -19,6 +19,13 @@ use proc_macro::TokenStream;
 use quote::{format_ident, quote};
 use syn::{parse2, spanned::Spanned, DeriveInput, Field, Ident};
 
+#[cfg(all(feature = "tcp", feature = "unix"))]
+compile_error!("You can only enable one of the 'tcp' or 'unix' features");
+#[cfg(all(not(feature = "tcp"), not(feature = "unix")))]
+compile_error!("You must enable either the 'tcp' or 'unix' feature");
+#[cfg(all(feature = "unix", not(unix)))]
+compile_error!("The 'unix' feature requires compiling for a unix target");
+
 #[proc_macro_derive(Protocol)]
 pub fn derive_protocol_derive(input: TokenStream) -> TokenStream {
     let expanded = derive_protocol(input.into());
@@ -212,7 +219,15 @@ fn derive_protocol(input: proc_macro2::TokenStream) -> proc_macro2::TokenStream
         }
     };
     // Create a struct to handle the connection from the client to the server
-    let stream_type = quote! { tokio::net::TcpStream }; // TODO: In the future we could support other stream types
+    #[cfg(feature = "tcp")]
+    let stream_type = quote! { tokio::net::TcpStream };
+    #[cfg(feature = "tcp")]
+    let stream_addr_trait = quote! { tokio::net::ToSocketAddrs };
+    #[cfg(feature = "unix")]
+    let stream_type = quote! { tokio::net::UnixStream };
+    #[cfg(feature = "unix")]
+    let stream_addr_trait = quote! { std::convert::AsRef<std::path::Path> };
+
     let cc_struct = quote! {
         struct #client_connection_struct_name {
             to_send: tokio::sync::mpsc::Receiver<(u64, #question_enum_name)>,
@@ -286,13 +301,30 @@ fn derive_protocol(input: proc_macro2::TokenStream) -> proc_macro2::TokenStream
             queries: #queries_struct_name,
             send_queue: tokio::sync::mpsc::Sender<(u64, #question_enum_name)>,
             recv_queue: #client_recv_queue_wrapper,
+            connection_task: Option<::std::sync::Arc<tokio::task::JoinHandle<()>>>,
         }
         impl #client_struct_name {
-            pub fn new(send_queue: tokio::sync::mpsc::Sender<(u64, #question_enum_name)>, recv_queue: tokio::sync::mpsc::Receiver<(u64, #answer_enum_name)>) -> Self {
+            pub fn new(send_queue: tokio::sync::mpsc::Sender<(u64, #question_enum_name)>,
+                        recv_queue: tokio::sync::mpsc::Receiver<(u64, #answer_enum_name)>,
+                        connection_task: Option<::std::sync::Arc<tokio::task::JoinHandle<()>>>) -> Self {
                 Self {
                     queries: #queries_struct_name::new(),
                     recv_queue: #client_recv_queue_wrapper::new(recv_queue),
                     send_queue,
+                    connection_task,
+                }
+            }
+            pub async fn connect<A: #stream_addr_trait>(addr: A) -> Result<Self, std::io::Error> {
+                let stream = #stream_type::connect(addr).await?;
+                let (send_queue, to_send) = tokio::sync::mpsc::channel(16);
+                let (to_recv, recv_queue) = tokio::sync::mpsc::channel(16);
+                let connection = #client_connection_struct_name::new(to_send, to_recv, stream);
+                let connection_task = tokio::spawn(connection.run());
+                Ok(Self::new(send_queue, recv_queue, Some(::std::sync::Arc::new(connection_task))))
+            }
+            pub fn close(self) {
+                if let Some(task) = self.connection_task {
+                    task.abort();
                 }
             }
             async fn send(&self, query: #question_enum_name) -> Result<u64, #error_enum_name> {

tests/mod.rs

@@ -30,7 +30,7 @@ enum TestProtocol {
 async fn main() {
     let (qtx, qrx) = mpsc::channel(16);
     let (atx, arx) = mpsc::channel(16);
-    let client = TestProtocolClient::new(qtx, arx);
+    let client = TestProtocolClient::new(qtx, arx, None);
     let server = tokio::spawn(server_loop(qrx, atx));
     let result = client.addition(2, 5).await.unwrap();
     assert_eq!(result, 7);
@@ -91,7 +91,7 @@ async fn server_loop(
 async fn heavy_async() {
     let (qtx, qrx) = mpsc::channel(16);
     let (atx, arx) = mpsc::channel(16);
-    let client = TestProtocolClient::new(qtx, arx);
+    let client = TestProtocolClient::new(qtx, arx, None);
     let server = tokio::spawn(server_loop(qrx, atx));
     let mut tasks = Vec::new();
     for i in 0..100 {