feat: update to new protocolhandler

Author: dignifiedquire

Cargo.lock

@@ -95,7 +95,7 @@ futures-util = "0.3.30"
 testdir = "0.9.1"
 
 [features]
-default = ["fs-store", "net_protocol"]
+default = ["fs-store", "net_protocol", "rpc"]
 downloader = ["dep:parking_lot", "tokio-util/time", "dep:hashlink"]
 net_protocol = ["downloader", "dep:futures-util"]
 fs-store = ["dep:reflink-copy", "redb", "dep:redb_v1", "dep:tempfile"]
@@ -184,6 +184,5 @@ panic = 'abort'
 incremental = false
 
 [patch.crates-io]
-# iroh-metrics = { git = "https://github.com/n0-computer/iroh", branch = "main" }
-# iroh-base = { git = "https://github.com/n0-computer/iroh", branch = "main" }
-# iroh = { git = "https://github.com/n0-computer/iroh", branch = "main" }
+iroh-base = { git = "https://github.com/n0-computer/iroh" }
+iroh = { git = "https://github.com/n0-computer/iroh" }

Cargo.toml

@@ -47,14 +47,14 @@ impl Default for GcState {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Blobs<S> {
     rt: LocalPoolHandle,
     pub(crate) store: S,
     events: EventSender,
     downloader: Downloader,
     #[cfg(feature = "rpc")]
-    batches: tokio::sync::Mutex<BlobBatches>,
+    batches: Arc<tokio::sync::Mutex<BlobBatches>>,
     endpoint: Endpoint,
     gc_state: Arc<std::sync::Mutex<GcState>>,
     #[cfg(feature = "rpc")]
@@ -131,15 +131,15 @@ impl<S: crate::store::Store> Builder<S> {
 
     /// Build the Blobs protocol handler.
     /// You need to provide a local pool handle and an endpoint.
-    pub fn build(self, rt: &LocalPoolHandle, endpoint: &Endpoint) -> Arc<Blobs<S>> {
+    pub fn build(self, rt: &LocalPoolHandle, endpoint: &Endpoint) -> Blobs<S> {
         let downloader = Downloader::new(self.store.clone(), endpoint.clone(), rt.clone());
-        Arc::new(Blobs::new(
+        Blobs::new(
             self.store,
             rt.clone(),
             self.events.unwrap_or_default(),
             downloader,
             endpoint.clone(),
-        ))
+        )
     }
 }
 
@@ -391,82 +391,26 @@ impl<S: crate::store::Store> Blobs<S> {
     }
 }
 
-// trait BlobsInner: Debug + Send + Sync + 'static {
-//     fn shutdown(self: Arc<Self>) -> BoxedFuture<()>;
-//     fn accept(self: Arc<Self>, conn: Connecting) -> BoxedFuture<Result<()>>;
-//     fn client(self: Arc<Self>) -> MemClient;
-//     fn local_pool_handle(&self) -> &LocalPoolHandle;
-//     fn downloader(&self) -> &Downloader;
-// }
-
-// #[derive(Debug)]
-// struct Blobs2 {
-//     inner: Arc<dyn BlobsInner>,
-// }
-
-// impl Blobs2 {
-//     fn client(&self) -> MemClient {
-//         self.inner.clone().client()
-//     }
-
-//     fn local_pool_handle(&self) -> &LocalPoolHandle {
-//         self.inner.local_pool_handle()
-//     }
-
-//     fn downloader(&self) -> &Downloader {
-//         self.inner.downloader()
-//     }
-// }
-
-// impl<S: crate::store::Store> BlobsInner for Blobs<S> {
-//     fn shutdown(self: Arc<Self>) -> BoxedFuture<()> {
-//         ProtocolHandler::shutdown(self)
-//     }
-
-//     fn accept(self: Arc<Self>, conn: Connecting) -> BoxedFuture<Result<()>> {
-//         ProtocolHandler::accept(self, conn)
-//     }
-
-//     fn client(self: Arc<Self>) -> MemClient {
-//         Blobs::client(self)
-//     }
-
-//     fn local_pool_handle(&self) -> &LocalPoolHandle {
-//         self.rt()
-//     }
-
-//     fn downloader(&self) -> &Downloader {
-//         self.downloader()
-//     }
-// }
-
-// impl ProtocolHandler for Blobs2 {
-//     fn accept(self: Arc<Self>, conn: Connecting) -> BoxedFuture<Result<()>> {
-//         self.inner.clone().accept(conn)
-//     }
-
-//     fn shutdown(self: Arc<Self>) -> BoxedFuture<()> {
-//         self.inner.clone().shutdown()
-//     }
-// }
-
 impl<S: crate::store::Store> ProtocolHandler for Blobs<S> {
-    fn accept(self: Arc<Self>, conn: Connecting) -> BoxedFuture<Result<()>> {
+    fn accept(&self, conn: Connecting) -> BoxedFuture<Result<()>> {
+        let this = self.clone();
+
         Box::pin(async move {
             crate::provider::handle_connection(
                 conn.await?,
-                self.store.clone(),
-                self.events.clone(),
-                self.rt.clone(),
+                this.store.clone(),
+                this.events.clone(),
+                this.rt.clone(),
             )
             .await;
             Ok(())
         })
     }
 
-    fn shutdown(self: Arc<Self>) -> BoxedFuture<()> {
+    fn shutdown(&self) -> BoxedFuture<()> {
+        let this = self.clone();
         Box::pin(async move {
-            self.store.shutdown().await;
+            this.store.shutdown().await;
         })
     }
 }

src/net_protocol.rs

@@ -63,7 +63,7 @@ const RPC_BLOB_GET_CHANNEL_CAP: usize = 2;
 
 impl<D: crate::store::Store> Blobs<D> {
     /// Get a client for the blobs protocol
-    pub fn client(self: Arc<Self>) -> blobs::MemClient {
+    pub fn client(&self) -> blobs::MemClient {
         let client = self
             .rpc_handler
             .get_or_init(|| RpcHandler::new(&self))
@@ -74,7 +74,7 @@ impl<D: crate::store::Store> Blobs<D> {
 
     /// Handle an RPC request
     pub async fn handle_rpc_request<C>(
-        self: Arc<Self>,
+        self,
         msg: Request,
         chan: RpcChannel<RpcService, C>,
     ) -> std::result::Result<(), RpcServerError<C>>
@@ -91,7 +91,7 @@ impl<D: crate::store::Store> Blobs<D> {
 }
 
 #[derive(Clone)]
-struct Handler<S>(Arc<Blobs<S>>);
+struct Handler<S>(Blobs<S>);
 
 impl<S> Deref for Handler<S> {
     type Target = Blobs<S>;
@@ -903,7 +903,7 @@ pub(crate) struct RpcHandler {
 }
 
 impl RpcHandler {
-    fn new<D: crate::store::Store>(blobs: &Arc<Blobs<D>>) -> Self {
+    fn new<D: crate::store::Store>(blobs: &Blobs<D>) -> Self {
         let blobs = blobs.clone();
         let (listener, connector) = quic_rpc::transport::flume::channel(1);
         let listener = RpcServer::new(listener);

src/rpc.rs

@@ -1003,7 +1003,7 @@ mod tests {
 
     mod node {
         //! An iroh node that just has the blobs transport
-        use std::{path::Path, sync::Arc};
+        use std::path::Path;
 
         use iroh::{protocol::Router, Endpoint, NodeAddr, NodeId};
         use tokio_util::task::AbortOnDropHandle;
@@ -1068,13 +1068,13 @@ mod tests {
                 // Setup blobs
                 let downloader =
                     Downloader::new(store.clone(), endpoint.clone(), local_pool.handle().clone());
-                let blobs = Arc::new(Blobs::new(
+                let blobs = Blobs::new(
                     store.clone(),
                     local_pool.handle().clone(),
                     events,
                     downloader,
                     endpoint.clone(),
-                ));
+                );
                 router = router.accept(crate::ALPN, blobs.clone());
 
                 // Build the router

src/rpc/client/blobs.rs

@@ -3,7 +3,6 @@ use std::{
     io,
     io::{Cursor, Write},
     path::PathBuf,
-    sync::Arc,
     time::Duration,
 };
 
@@ -41,7 +40,7 @@ use tokio::io::AsyncReadExt;
 #[derive(Debug)]
 pub struct Node<S> {
     pub router: iroh::protocol::Router,
-    pub blobs: Arc<Blobs<S>>,
+    pub blobs: Blobs<S>,
     pub store: S,
     pub _local_pool: LocalPool,
 }