feat: extract iroh transport

Cargo.toml

@@ -19,7 +19,6 @@ futures-lite = "2.3.0"
 futures-sink = "0.3.30"
 futures-util = { version = "0.3.30", features = ["sink"] }
 hyper = { version = "0.14.16", features = ["full"], optional = true }
-iroh = { version = "0.31", optional = true }
 pin-project = "1"
 quinn = { package = "iroh-quinn", version = "0.12", optional = true }
 serde = { version = "1", features = ["derive"] }
@@ -46,7 +45,6 @@ anyhow = "1"
 async-stream = "0.3.3"
 derive_more = { version = "1", features = ["from", "try_into", "display"] }
 rand = "0.8"
-
 serde = { version = "1", features = ["derive"] }
 tokio = { version = "1", features = ["full"] }
 quinn = { package = "iroh-quinn", version = "0.12", features = ["ring"] }
@@ -67,8 +65,6 @@ hyper-transport = ["dep:flume", "dep:hyper", "dep:postcard", "dep:bytes", "dep:t
 quinn-transport = ["dep:flume", "dep:quinn", "dep:postcard", "dep:bytes", "dep:tokio-serde", "tokio-util/codec"]
 ## In memory transport using the `flume` crate
 flume-transport = ["dep:flume"]
-## p2p QUIC transport using the `iroh` crate
-iroh-transport = ["dep:iroh", "dep:flume", "dep:postcard", "dep:tokio-serde", "tokio-util/codec"]
 ## Macros for creating request handlers
 macros = []
 ## Utilities for testing
@@ -100,4 +96,10 @@ name = "modularize"
 required-features = ["flume-transport"]
 
 [workspace]
-members = ["examples/split/types", "examples/split/server", "examples/split/client", "quic-rpc-derive"]
+members = [
+  "examples/split/types",
+  "examples/split/server",
+  "examples/split/client",
+  "quic-rpc-derive",
+  "quic-rpc-transport-iroh"
+]

quic-rpc-transport-iroh/Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "quic-rpc-transport-iroh"
+authors = ["Rüdiger Klaehn <[email protected]>", "n0 team"]
+version = "0.31.0"
+edition = "2021"
+keywords = ["api", "protocol", "network", "rpc"]
+categories = ["network-programming"]
+license = "Apache-2.0/MIT"
+repository = "https://github.com/n0-computer/quic-rpc"
+description = "An iroh transport for quic-rpc"
+
+
+[dependencies]
+iroh = "0.31"
+quic-rpc = { version = "0.18.0", path = "../", default-features = false, features = ["quinn-transport"] }
+flume = "0.11"
+postcard = { version = "1", features = ["use-std"] }
+tokio-serde = { version = "0.9", features = [] }
+tokio-util = { version = "0.7", features = ["codec"] }
+quinn = { package = "iroh-quinn", version = "0.12" }
+futures-lite = "2.3.0"
+futures-sink = "0.3.30"
+futures-util = { version = "0.3.30", features = ["sink"] }
+tokio = { version = "1", default-features = false, features = ["macros", "sync"] }
+serde = { version = "1", features = ["derive"] }
+tracing = "0.1"
+anyhow = "1"
+pin-project = "1"
+
+[dev-dependencies]
+testresult = "0.4.1"
+tokio = { version = "1", features = ["full"] }
+rand = "0.8"
+async-stream = "0.3.3"
+derive_more = { version = "1", features = ["from", "try_into", "display"] }
+futures-buffered = "0.2.4"
+thousands = "0.2.0"
+tracing-subscriber = "0.3.16"

quic-rpc-transport-iroh/README.md

@@ -0,0 +1,6 @@
+# Quic-Rpc Transport [iroh]
+
+> A transport allowing for quic rpc to be run over any [iroh] connection
+
+
+[iroh]: https://github.com/n0-computer/iroh

quic-rpc-transport-iroh/src/lib.rs

@@ -0,0 +1,60 @@
+//! `iroh` transport implementation based on [iroh](https://crates.io/crates/iroh)
+
+pub mod transport;
+
+use futures_util::sink::SinkExt;
+use futures_util::TryStreamExt;
+use quic_rpc::transport::boxed::{
+    AcceptFuture, BoxableConnector, BoxableListener, OpenFuture, RecvStream, SendSink,
+};
+use quic_rpc::transport::{Connector, Listener, LocalAddr};
+use quic_rpc::{RpcMessage, Service};
+
+/// An iroh listener for the given [`Service`]
+pub type IrohListener<S> = crate::transport::IrohListener<<S as Service>::Req, <S as Service>::Res>;
+
+/// An iroh connector for the given [`Service`]
+pub type IrohConnector<S> =
+    crate::transport::IrohConnector<<S as Service>::Res, <S as Service>::Req>;
+
+impl<In: RpcMessage, Out: RpcMessage> BoxableConnector<In, Out>
+    for crate::transport::IrohConnector<In, Out>
+{
+    fn clone_box(&self) -> Box<dyn BoxableConnector<In, Out>> {
+        Box::new(self.clone())
+    }
+
+    fn open_boxed(&self) -> OpenFuture<In, Out> {
+        let f = Box::pin(async move {
+            let (send, recv) = Connector::open(self).await?;
+            // map the error types to anyhow
+            let send = send.sink_map_err(anyhow::Error::from);
+            let recv = recv.map_err(anyhow::Error::from);
+            // return the boxed streams
+            anyhow::Ok((SendSink::boxed(send), RecvStream::boxed(recv)))
+        });
+        OpenFuture::boxed(f)
+    }
+}
+
+impl<In: RpcMessage, Out: RpcMessage> BoxableListener<In, Out>
+    for crate::transport::IrohListener<In, Out>
+{
+    fn clone_box(&self) -> Box<dyn BoxableListener<In, Out>> {
+        Box::new(self.clone())
+    }
+
+    fn accept_bi_boxed(&self) -> AcceptFuture<In, Out> {
+        let f = async move {
+            let (send, recv) = Listener::accept(self).await?;
+            let send = send.sink_map_err(anyhow::Error::from);
+            let recv = recv.map_err(anyhow::Error::from);
+            anyhow::Ok((SendSink::boxed(send), RecvStream::boxed(recv)))
+        };
+        AcceptFuture::boxed(f)
+    }
+
+    fn local_addr(&self) -> &[LocalAddr] {
+        Listener::local_addr(self)
+    }
+}

src/transport/iroh.rs → quic-rpc-transport-iroh/src/transport.rs

@@ -1,5 +1,3 @@
-//! iroh transport implementation based on [iroh](https://crates.io/crates/iroh)
-
 use std::{
     collections::BTreeSet,
     fmt,
@@ -23,12 +21,11 @@ use serde::{de::DeserializeOwned, Serialize};
 use tokio::{sync::oneshot, task::yield_now};
 use tracing::{debug_span, Instrument};
 
-use super::{
-    util::{FramedPostcardRead, FramedPostcardWrite},
-    StreamTypes,
-};
-use crate::{
-    transport::{ConnectionErrors, Connector, Listener, LocalAddr},
+use quic_rpc::{
+    transport::{
+        ConnectionErrors, Connector, FramedPostcardRead, FramedPostcardWrite, Listener, LocalAddr,
+        StreamTypes,
+    },
     RpcMessage,
 };
 

tests/iroh.rs → quic-rpc-transport-iroh/tests/iroh.rs

@@ -1,15 +1,12 @@
-#![cfg(feature = "iroh-transport")]
-
 use iroh::{NodeAddr, SecretKey};
-use quic_rpc::{transport, RpcClient, RpcServer};
+use quic_rpc::{RpcClient, RpcServer};
 use testresult::TestResult;
 
-use crate::transport::iroh::{IrohConnector, IrohListener};
+use quic_rpc_transport_iroh::transport::{IrohConnector, IrohListener};
 
 mod math;
 use math::*;
 use tokio_util::task::AbortOnDropHandle;
-mod util;
 
 const ALPN: &[u8] = b"quic-rpc/iroh/test";
 
@@ -99,23 +96,22 @@ async fn server_away_and_back() -> TestResult<()> {
     let server_node_id = server_secret_key.public();
 
     // create the RPC client
-    let client_connection = transport::iroh::IrohConnector::<ComputeResponse, ComputeRequest>::new(
+    let client_connection = IrohConnector::<ComputeResponse, ComputeRequest>::new(
         client_endpoint.clone(),
         server_node_id,
         ALPN.into(),
     );
-    let client = RpcClient::<
-        ComputeService,
-        transport::iroh::IrohConnector<ComputeResponse, ComputeRequest>,
-    >::new(client_connection);
+    let client = RpcClient::<ComputeService, IrohConnector<ComputeResponse, ComputeRequest>>::new(
+        client_connection,
+    );
 
     // send a request. No server available so it should fail
     client.rpc(Sqr(4)).await.unwrap_err();
 
     let server_endpoint = make_endpoint(server_secret_key.clone(), ALPN).await?;
 
     // create the RPC Server
-    let connection = transport::iroh::IrohListener::new(server_endpoint.clone())?;
+    let connection = IrohListener::new(server_endpoint.clone())?;
     let server = RpcServer::new(connection);
     let server_handle = tokio::spawn(ComputeService::server_bounded(server, 1));
 
@@ -138,7 +134,7 @@ async fn server_away_and_back() -> TestResult<()> {
     let server_endpoint = make_endpoint(server_secret_key.clone(), ALPN).await?;
 
     // make the server run again
-    let connection = transport::iroh::IrohListener::new(server_endpoint.clone())?;
+    let connection = IrohListener::new(server_endpoint.clone())?;
     let server = RpcServer::new(connection);
     let server_handle = tokio::spawn(ComputeService::server_bounded(server, 5));