iroh-net transport (#105)

Author: rklaehn

Cargo.lock

@@ -21,6 +21,7 @@ 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-net = { version = "0.28.1", optional = true }
 pin-project = "1"
 quinn = { package = "iroh-quinn", version = "0.12", optional = true }
 serde = { version = "1.0.183", features = ["derive"] }
@@ -56,6 +57,7 @@ nested_enum_utils = "0.1.0"
 hyper-transport = ["dep:flume", "dep:hyper", "dep:bincode", "dep:bytes", "dep:tokio-serde", "dep:tokio-util"]
 quinn-transport = ["dep:flume", "dep:quinn", "dep:bincode", "dep:tokio-serde", "dep:tokio-util"]
 flume-transport = ["dep:flume"]
+iroh-net-transport = ["dep:iroh-net", "dep:flume", "dep:quinn", "dep:bincode", "dep:tokio-serde", "dep:tokio-util"]
 macros = []
 default = ["flume-transport"]
 

Cargo.toml

@@ -398,6 +398,50 @@ impl<In: RpcMessage, Out: RpcMessage> BoxableListener<In, Out>
     }
 }
 
+#[cfg(feature = "iroh-net-transport")]
+impl<In: RpcMessage, Out: RpcMessage> BoxableConnector<In, Out>
+    for super::iroh_net::IrohNetConnector<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) = super::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)
+    }
+}
+
+#[cfg(feature = "iroh-net-transport")]
+impl<In: RpcMessage, Out: RpcMessage> BoxableListener<In, Out>
+    for super::iroh_net::IrohNetListener<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) = super::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) -> &[super::LocalAddr] {
+        super::Listener::local_addr(self)
+    }
+}
+
 #[cfg(feature = "flume-transport")]
 impl<In: RpcMessage, Out: RpcMessage> BoxableConnector<In, Out>
     for super::flume::FlumeConnector<In, Out>