[RFC datacenter-manager 2/6] parallel fetcher: allow to use custom client factory

[Lukas Wagner <l.wagner@proxmox.com>]

This will become useful later when the actual client factory is a custom
mocked one that we use for tests.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
 .../tasks/remote_tasks.rs                     |  2 ++
 server/src/parallel_fetcher.rs                | 24 +++++++++++++++++--
 2 files changed, 24 insertions(+), 2 deletions(-)

diff --git a/server/src/bin/proxmox-datacenter-api/tasks/remote_tasks.rs b/server/src/bin/proxmox-datacenter-api/tasks/remote_tasks.rs
index c71a0894..7ce37631 100644
--- a/server/src/bin/proxmox-datacenter-api/tasks/remote_tasks.rs
+++ b/server/src/bin/proxmox-datacenter-api/tasks/remote_tasks.rs
@@ -264,10 +264,12 @@ async fn fetch_remotes(
     remotes: Vec<Remote>,
     cache_state: Arc<State>,
 ) -> (Vec<TaskCacheItem>, NodeFetchSuccessMap) {
+    // FIXME: Use constructor or builder pattern here...
     let fetcher = ParallelFetcher {
         max_connections: MAX_CONNECTIONS,
         max_connections_per_remote: CONNECTIONS_PER_PVE_REMOTE,
         context: cache_state,
+        client_factory: connection::client_factory(),
     };
 
     let fetch_results = fetcher
diff --git a/server/src/parallel_fetcher.rs b/server/src/parallel_fetcher.rs
index 58ca1f55..f123460d 100644
--- a/server/src/parallel_fetcher.rs
+++ b/server/src/parallel_fetcher.rs
@@ -14,15 +14,18 @@ use tokio::{
     task::JoinSet,
 };
 
-use crate::connection;
+use crate::connection::{self, ClientFactory};
 
 pub const DEFAULT_MAX_CONNECTIONS: usize = 20;
 pub const DEFAULT_MAX_CONNECTIONS_PER_REMOTE: usize = 5;
 
+/// FIXME: Builder pattern, make fields private
 pub struct ParallelFetcher<C> {
     pub max_connections: usize,
     pub max_connections_per_remote: usize,
     pub context: C,
+
+    pub client_factory: Arc<dyn ClientFactory + Send + Sync>,
 }
 
 pub struct FetchResults<T> {
@@ -66,6 +69,20 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
             max_connections: DEFAULT_MAX_CONNECTIONS,
             max_connections_per_remote: DEFAULT_MAX_CONNECTIONS_PER_REMOTE,
             context,
+            client_factory: connection::client_factory(),
+        }
+    }
+
+    // TODO: maybe add actual builder pattern.
+    pub fn new_with_client_factory(
+        context: C,
+        client_factory: Arc<dyn ClientFactory + Send + Sync>,
+    ) -> Self {
+        Self {
+            max_connections: DEFAULT_MAX_CONNECTIONS,
+            max_connections_per_remote: DEFAULT_MAX_CONNECTIONS_PER_REMOTE,
+            context,
+            client_factory,
         }
     }
 
@@ -82,6 +99,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
 
         for remote in remotes {
             let semaphore = Arc::clone(&total_connections_semaphore);
+            let client_factory = Arc::clone(&self.client_factory);
 
             let f = func.clone();
 
@@ -91,6 +109,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
                 semaphore,
                 f,
                 self.max_connections_per_remote,
+                client_factory,
             ));
         }
 
@@ -116,6 +135,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
         semaphore: Arc<Semaphore>,
         func: F,
         max_connections_per_remote: usize,
+        client_factory: Arc<dyn ClientFactory + Send + Sync>,
     ) -> (String, Result<RemoteResult<T>, Error>)
     where
         F: Fn(C, Remote, String) -> Ft + Clone + Send + 'static,
@@ -132,7 +152,7 @@ impl<C: Clone + Send + 'static> ParallelFetcher<C> {
                 let remote_clone = remote.clone();
 
                 let nodes = match async move {
-                    let client = connection::make_pve_client(&remote_clone)?;
+                    let client = client_factory.make_pve_client(&remote_clone)?;
                     let nodes = client.list_nodes().await?;
 
                     Ok::<Vec<ClusterNodeIndexResponse>, Error>(nodes)
-- 
2.47.3