Allow getting the downloader config so we can test it

Author: rklaehn

src/downloader.rs

@@ -141,7 +141,7 @@ pub enum GetOutput<N> {
 }
 
 /// Concurrency limits for the [`Downloader`].
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct ConcurrencyLimits {
     /// Maximum number of requests the service performs concurrently.
     pub max_concurrent_requests: usize,
@@ -193,7 +193,7 @@ impl ConcurrencyLimits {
 }
 
 /// Configuration for retry behavior of the [`Downloader`].
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct RetryConfig {
     /// Maximum number of retry attempts for a node that failed to dial or failed with IO errors.
     pub max_retries_per_node: u32,
@@ -325,6 +325,15 @@ impl Future for DownloadHandle {
     }
 }
 
+/// All numerical config options for the downloader.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub struct Config {
+    /// Concurrency limits for the downloader.
+    pub concurrency: ConcurrencyLimits,
+    /// Retry configuration for the downloader.
+    pub retry: RetryConfig,
+}
+
 /// Handle for the download services.
 #[derive(Clone, Debug)]
 pub struct Downloader {
@@ -374,6 +383,15 @@ impl Downloader {
         }
     }
 
+    /// Get the current configuration.
+    pub async fn get_config(&self) -> anyhow::Result<Config> {
+        let (tx, rx) = oneshot::channel();
+        let msg = Message::GetConfig { tx };
+        self.msg_tx.send(msg).await?;
+        let config = rx.await?;
+        Ok(config)
+    }
+
     /// Queue a download.
     pub async fn queue(&self, request: DownloadRequest) -> DownloadHandle {
         let kind = request.kind;
@@ -441,6 +459,11 @@ enum Message {
     /// Cancel an intent. The associated request will be cancelled when the last intent is
     /// cancelled.
     CancelIntent { id: IntentId, kind: DownloadKind },
+    /// Get the config
+    GetConfig {
+        #[debug(skip)]
+        tx: oneshot::Sender<Config>,
+    },
 }
 
 #[derive(derive_more::Debug)]
@@ -668,6 +691,13 @@ impl<G: Getter<Connection = D::Connection>, D: DialerT> Service<G, D> {
                     self.queue.unpark_hash(hash);
                 }
             }
+            Message::GetConfig { tx } => {
+                let config = Config {
+                    concurrency: self.concurrency_limits,
+                    retry: self.retry_config,
+                };
+                tx.send(config).ok();
+            }
         }
     }
 

src/net_protocol.rs

@@ -142,18 +142,12 @@ impl BlobBatches {
     }
 }
 
-#[derive(Debug, Default)]
-pub struct DownloaderConfig {
-    pub concurrency: downloader::ConcurrencyLimits,
-    pub retry: downloader::RetryConfig,
-}
-
 /// Builder for the Blobs protocol handler
 #[derive(Debug)]
 pub struct Builder<S> {
     store: S,
     events: Option<EventSender>,
-    downloader: Option<DownloaderConfig>,
+    downloader: Option<crate::downloader::Config>,
     rt: Option<LocalPoolHandle>,
 }
 
@@ -171,7 +165,7 @@ impl<S: crate::store::Store> Builder<S> {
     }
 
     /// Set a custom downloader configuration.
-    pub fn downloader(mut self, config: DownloaderConfig) -> Self {
+    pub fn downloader(mut self, config: downloader::Config) -> Self {
         self.downloader = Some(config);
         self
     }
@@ -183,7 +177,7 @@ impl<S: crate::store::Store> Builder<S> {
             .rt
             .map(Rt::Handle)
             .unwrap_or_else(|| Rt::Owned(LocalPool::default()));
-        let DownloaderConfig { concurrency, retry } = self.downloader.unwrap_or_default();
+        let downloader::Config { concurrency, retry } = self.downloader.unwrap_or_default();
         let downloader = Downloader::with_config(
             self.store.clone(),
             endpoint.clone(),

tests/rpc.rs

@@ -1,7 +1,7 @@
 #![cfg(feature = "test")]
 use std::{net::SocketAddr, path::PathBuf, vec};
 
-use iroh_blobs::net_protocol::Blobs;
+use iroh_blobs::{downloader, net_protocol::Blobs};
 use quic_rpc::client::QuinnConnector;
 use tempfile::TempDir;
 use testresult::TestResult;
@@ -85,3 +85,26 @@ async fn quinn_rpc_large() -> TestResult<()> {
     assert_eq!(data, &data2[..]);
     Ok(())
 }
+
+#[tokio::test]
+async fn downloader_config() -> TestResult<()> {
+    let _ = tracing_subscriber::fmt::try_init();
+    let endpoint = iroh::Endpoint::builder().bind().await?;
+    let store = iroh_blobs::store::mem::Store::default();
+    let expected = downloader::Config {
+        concurrency: downloader::ConcurrencyLimits {
+            max_concurrent_requests: usize::MAX,
+            max_concurrent_requests_per_node: usize::MAX,
+            max_open_connections: usize::MAX,
+            max_concurrent_dials_per_hash: usize::MAX,
+        },
+        retry: downloader::RetryConfig {
+            max_retries_per_node: u32::MAX,
+            initial_retry_delay: std::time::Duration::from_secs(1),
+        },
+    };
+    let blobs = Blobs::builder(store).downloader(expected).build(&endpoint);
+    let actual = blobs.downloader().get_config().await?;
+    assert_eq!(expected, actual);
+    Ok(())
+}