back to the lazy client

Author: rklaehn

examples/custom-protocol.rs

@@ -91,7 +91,7 @@ async fn main() -> Result<()> {
     let local_pool = LocalPool::default();
     let blobs = Blobs::memory().build(local_pool.handle(), builder.endpoint());
     let builder = builder.accept(iroh_blobs::ALPN, blobs.clone());
-    let blobs_client = blobs.spawn_rpc();
+    let blobs_client = blobs.client();
 
     // Build our custom protocol handler. The `builder` exposes access to various subsystems in the
     // iroh node. In our case, we need a blobs client and the endpoint.
@@ -122,7 +122,7 @@ async fn main() -> Result<()> {
 
             // Print out our query results.
             for hash in hashes {
-                read_and_print(&blobs_client, hash).await?;
+                read_and_print(blobs_client, hash).await?;
             }
         }
     }

examples/hello-world-fetch.rs

@@ -42,7 +42,7 @@ async fn main() -> Result<()> {
     let blobs = Blobs::memory().build(local_pool.handle(), builder.endpoint());
     let builder = builder.accept(iroh_blobs::ALPN, blobs.clone());
     let node = builder.spawn().await?;
-    let blobs_client = blobs.spawn_rpc();
+    let blobs_client = blobs.client();
 
     println!("fetching hash:  {}", ticket.hash());
     println!("node id:        {}", node.endpoint().node_id());

examples/hello-world-provide.rs

@@ -28,7 +28,7 @@ async fn main() -> anyhow::Result<()> {
     let local_pool = LocalPool::default();
     let blobs = Blobs::memory().build(local_pool.handle(), builder.endpoint());
     let builder = builder.accept(iroh_blobs::ALPN, blobs.clone());
-    let blobs_client = blobs.spawn_rpc();
+    let blobs_client = blobs.client();
     let node = builder.spawn().await?;
 
     // add some data and remember the hash

examples/local-swarm-discovery.rs

@@ -78,7 +78,7 @@ async fn main() -> anyhow::Result<()> {
     let blobs = Blobs::memory().build(local_pool.handle(), builder.endpoint());
     let builder = builder.accept(iroh_blobs::ALPN, blobs.clone());
     let node = builder.spawn().await?;
-    let blobs_client = blobs.spawn_rpc();
+    let blobs_client = blobs.client();
 
     match &cli.command {
         Commands::Accept { path } => {

examples/transfer.rs

@@ -26,7 +26,7 @@ async fn main() -> Result<()> {
         .spawn()
         .await?;
 
-    let blobs = blobs.spawn_rpc();
+    let blobs = blobs.client();
 
     let args = std::env::args().collect::<Vec<_>>();
     match &args.iter().map(String::as_str).collect::<Vec<_>>()[..] {

src/net_protocol.rs

@@ -5,28 +5,22 @@
 
 use std::{collections::BTreeSet, fmt::Debug, ops::DerefMut, sync::Arc};
 
-use anyhow::{anyhow, bail, Result};
+use anyhow::{bail, Result};
 use futures_lite::future::Boxed as BoxedFuture;
 use futures_util::future::BoxFuture;
 use iroh::{endpoint::Connecting, protocol::ProtocolHandler, Endpoint, NodeAddr};
 use iroh_base::hash::{BlobFormat, Hash};
 use serde::{Deserialize, Serialize};
-use tracing::{debug, warn};
+use tracing::debug;
 
 use crate::{
-    downloader::{DownloadRequest, Downloader},
-    get::{
-        db::{DownloadProgress, GetState},
-        Stats,
-    },
+    downloader::Downloader,
     provider::EventSender,
     store::GcConfig,
     util::{
         local_pool::{self, LocalPoolHandle},
-        progress::{AsyncChannelProgressSender, ProgressSender},
         SetTagOption,
     },
-    HashAndFormat,
 };
 
 /// A callback that blobs can ask about a set of hashes that should not be garbage collected.
@@ -48,20 +42,22 @@ impl Default for GcState {
 }
 
 #[derive(Debug)]
-struct BlobsInner<S> {
-    rt: LocalPoolHandle,
+pub(crate) struct BlobsInner<S> {
+    pub(crate) rt: LocalPoolHandle,
     pub(crate) store: S,
     events: EventSender,
-    downloader: Downloader,
-    endpoint: Endpoint,
+    pub(crate) downloader: Downloader,
+    pub(crate) endpoint: Endpoint,
     gc_state: std::sync::Mutex<GcState>,
     #[cfg(feature = "rpc")]
-    batches: tokio::sync::Mutex<BlobBatches>,
+    pub(crate) batches: tokio::sync::Mutex<BlobBatches>,
 }
 
 #[derive(Debug, Clone)]
 pub struct Blobs<S> {
-    inner: Arc<BlobsInner<S>>,
+    pub(crate) inner: Arc<BlobsInner<S>>,
+    #[cfg(feature = "rpc")]
+    pub(crate) rpc_handler: Arc<std::sync::OnceLock<crate::rpc::RpcHandler>>,
 }
 
 /// Keeps track of all the currently active batch operations of the blobs api.
@@ -79,7 +75,7 @@ pub(crate) struct BlobBatches {
 #[derive(Debug, Default)]
 struct BlobBatch {
     /// The tags in this batch.
-    tags: std::collections::BTreeMap<HashAndFormat, Vec<crate::TempTag>>,
+    tags: std::collections::BTreeMap<iroh::hash::HashAndFormat, Vec<crate::TempTag>>,
 }
 
 #[cfg(feature = "rpc")]
@@ -98,7 +94,11 @@ impl BlobBatches {
     }
 
     /// Remove a tag from a batch.
-    pub fn remove_one(&mut self, batch: BatchId, content: &HashAndFormat) -> Result<()> {
+    pub fn remove_one(
+        &mut self,
+        batch: BatchId,
+        content: &iroh::hash::HashAndFormat,
+    ) -> Result<()> {
         if let Some(batch) = self.batches.get_mut(&batch) {
             if let Some(tags) = batch.tags.get_mut(content) {
                 tags.pop();
@@ -191,6 +191,8 @@ impl<S: crate::store::Store> Blobs<S> {
                 batches: Default::default(),
                 gc_state: Default::default(),
             }),
+            #[cfg(feature = "rpc")]
+            rpc_handler: Default::default(),
         }
     }
 
@@ -255,147 +257,6 @@ impl<S: crate::store::Store> Blobs<S> {
         *state = GcState::Started(Some(run));
         Ok(())
     }
-
-    #[cfg(feature = "rpc")]
-    pub(crate) async fn batches(&self) -> tokio::sync::MutexGuard<'_, BlobBatches> {
-        self.inner.batches.lock().await
-    }
-
-    pub(crate) async fn download(
-        &self,
-        endpoint: Endpoint,
-        req: BlobDownloadRequest,
-        progress: AsyncChannelProgressSender<DownloadProgress>,
-    ) -> Result<()> {
-        let BlobDownloadRequest {
-            hash,
-            format,
-            nodes,
-            tag,
-            mode,
-        } = req;
-        let hash_and_format = HashAndFormat { hash, format };
-        let temp_tag = self.store().temp_tag(hash_and_format);
-        let stats = match mode {
-            DownloadMode::Queued => {
-                self.download_queued(endpoint, hash_and_format, nodes, progress.clone())
-                    .await?
-            }
-            DownloadMode::Direct => {
-                self.download_direct_from_nodes(endpoint, hash_and_format, nodes, progress.clone())
-                    .await?
-            }
-        };
-
-        progress.send(DownloadProgress::AllDone(stats)).await.ok();
-        match tag {
-            SetTagOption::Named(tag) => {
-                self.store().set_tag(tag, Some(hash_and_format)).await?;
-            }
-            SetTagOption::Auto => {
-                self.store().create_tag(hash_and_format).await?;
-            }
-        }
-        drop(temp_tag);
-
-        Ok(())
-    }
-
-    async fn download_queued(
-        &self,
-        endpoint: Endpoint,
-        hash_and_format: HashAndFormat,
-        nodes: Vec<NodeAddr>,
-        progress: AsyncChannelProgressSender<DownloadProgress>,
-    ) -> Result<Stats> {
-        /// Name used for logging when new node addresses are added from gossip.
-        const BLOB_DOWNLOAD_SOURCE_NAME: &str = "blob_download";
-
-        let mut node_ids = Vec::with_capacity(nodes.len());
-        let mut any_added = false;
-        for node in nodes {
-            node_ids.push(node.node_id);
-            if !node.info.is_empty() {
-                endpoint.add_node_addr_with_source(node, BLOB_DOWNLOAD_SOURCE_NAME)?;
-                any_added = true;
-            }
-        }
-        let can_download = !node_ids.is_empty() && (any_added || endpoint.discovery().is_some());
-        anyhow::ensure!(can_download, "no way to reach a node for download");
-        let req = DownloadRequest::new(hash_and_format, node_ids).progress_sender(progress);
-        let handle = self.downloader().queue(req).await;
-        let stats = handle.await?;
-        Ok(stats)
-    }
-
-    #[tracing::instrument("download_direct", skip_all, fields(hash=%hash_and_format.hash.fmt_short()))]
-    async fn download_direct_from_nodes(
-        &self,
-        endpoint: Endpoint,
-        hash_and_format: HashAndFormat,
-        nodes: Vec<NodeAddr>,
-        progress: AsyncChannelProgressSender<DownloadProgress>,
-    ) -> Result<Stats> {
-        let mut last_err = None;
-        let mut remaining_nodes = nodes.len();
-        let mut nodes_iter = nodes.into_iter();
-        'outer: loop {
-            match crate::get::db::get_to_db_in_steps(
-                self.store().clone(),
-                hash_and_format,
-                progress.clone(),
-            )
-            .await?
-            {
-                GetState::Complete(stats) => return Ok(stats),
-                GetState::NeedsConn(needs_conn) => {
-                    let (conn, node_id) = 'inner: loop {
-                        match nodes_iter.next() {
-                            None => break 'outer,
-                            Some(node) => {
-                                remaining_nodes -= 1;
-                                let node_id = node.node_id;
-                                if node_id == endpoint.node_id() {
-                                    debug!(
-                                        ?remaining_nodes,
-                                        "skip node {} (it is the node id of ourselves)",
-                                        node_id.fmt_short()
-                                    );
-                                    continue 'inner;
-                                }
-                                match endpoint.connect(node, crate::protocol::ALPN).await {
-                                    Ok(conn) => break 'inner (conn, node_id),
-                                    Err(err) => {
-                                        debug!(
-                                            ?remaining_nodes,
-                                            "failed to connect to {}: {err}",
-                                            node_id.fmt_short()
-                                        );
-                                        continue 'inner;
-                                    }
-                                }
-                            }
-                        }
-                    };
-                    match needs_conn.proceed(conn).await {
-                        Ok(stats) => return Ok(stats),
-                        Err(err) => {
-                            warn!(
-                                ?remaining_nodes,
-                                "failed to download from {}: {err}",
-                                node_id.fmt_short()
-                            );
-                            last_err = Some(err);
-                        }
-                    }
-                }
-            }
-        }
-        match last_err {
-            Some(err) => Err(err.into()),
-            None => Err(anyhow!("No nodes to download from provided")),
-        }
-    }
 }
 
 impl<S: crate::store::Store> ProtocolHandler for Blobs<S> {