fix(installation): extract handle creation to named functions

Author: FranciscoTGouveia

src/dist/manifestation.rs

@@ -232,89 +232,32 @@ impl Manifestation {
             // The `mpsc` channel was used as we need to send many messages from one producer (download's thread) to one consumer (install's thread)
             // This is recommended in the official docs: https://docs.rs/tokio/latest/tokio/sync/index.html#mpsc-channel
             let total_components = components.len();
-            let (download_tx, mut download_rx) =
+            let (download_tx, download_rx) =
                 mpsc::channel::<Result<(ComponentBinary, File)>>(total_components);
 
             let semaphore = Arc::new(Semaphore::new(concurrent_downloads));
-            let component_stream = tokio_stream::iter(components.into_iter()).map({
-                let download_tx = download_tx.clone();
-                {
-                    let new_manifest = Arc::clone(&new_manifest);
-                    let download_cfg = download_cfg.clone();
-                    move |bin| {
-                        let sem = semaphore.clone();
-                        let download_tx = download_tx.clone();
-                        let tmp_cx = Arc::clone(&tmp_cx);
-                        let new_manifest = Arc::clone(&new_manifest);
-                        let download_cfg = download_cfg.clone();
-                        async move {
-                            let _permit = sem.acquire().await.unwrap();
-                            let url = if altered {
-                                utils::parse_url(
-                                    &bin.binary
-                                        .url
-                                        .replace(DEFAULT_DIST_SERVER, tmp_cx.dist_server.as_str()),
-                                )?
-                            } else {
-                                utils::parse_url(&bin.binary.url)?
-                            };
-
-                            let installer_file = bin
-                                .download(&url, &download_cfg, max_retries, &new_manifest)
-                                .await?;
-                            let hash = bin.binary.hash.clone();
-                            let _ = download_tx.send(Ok((bin, installer_file))).await;
-                            Ok(hash)
-                        }
-                    }
-                }
-            });
 
-            let mut stream = component_stream.buffered(components_len);
-            let download_handle = async move {
-                let mut hashes = Vec::new();
-                while let Some(result) = stream.next().await {
-                    match result {
-                        Ok(hash) => {
-                            hashes.push(hash);
-                        }
-                        Err(e) => {
-                            let _ = download_tx.send(Err(e)).await;
-                        }
-                    }
-                }
-                hashes
-            };
-            let install_handle = {
-                let new_manifest = Arc::clone(&new_manifest);
-                let download_cfg = Arc::clone(&download_cfg);
-                async move {
-                    let mut current_tx = tx;
-                    let mut counter = 0;
-                    while counter < total_components
-                        && let Some(message) = download_rx.recv().await
-                    {
-                        let (component_bin, installer_file) = message?;
-                        current_tx = tokio::task::spawn_blocking({
-                            let this = Arc::clone(&self);
-                            let new_manifest = Arc::clone(&new_manifest);
-                            let download_cfg = Arc::clone(&download_cfg);
-                            move || {
-                                this.install_component(
-                                    component_bin,
-                                    installer_file,
-                                    download_cfg,
-                                    new_manifest,
-                                    current_tx,
-                                )
-                            }
-                        })
-                        .await??;
-                        counter += 1;
-                    }
-                    Ok::<_, Error>(current_tx)
-                }
-            };
+            let component_stream = Self::component_stream(
+                components,
+                download_tx.clone(),
+                semaphore,
+                Arc::clone(&new_manifest),
+                Arc::clone(&download_cfg),
+                tmp_cx.dist_server.as_str(),
+                max_retries,
+                altered,
+            );
+
+            let stream = component_stream.buffered(components_len);
+            let download_handle = Self::download_handle(stream, download_tx);
+            let install_handle = Self::install_handle(
+                Arc::clone(&self),
+                download_rx,
+                total_components,
+                Arc::clone(&new_manifest),
+                Arc::clone(&download_cfg),
+                tx,
+            );
 
             let (download_results, install_result) = tokio::join!(download_handle, install_handle);
             things_downloaded = download_results;
@@ -568,6 +511,94 @@ impl Manifestation {
         component_bin.status.installed();
         tx
     }
+
+    #[allow(clippy::too_many_arguments)]
+    fn component_stream(
+        components: Vec<ComponentBinary>,
+        download_tx: mpsc::Sender<Result<(ComponentBinary, File)>>,
+        semaphore: Arc<Semaphore>,
+        new_manifest: Arc<Manifest>,
+        download_cfg: Arc<DownloadCfg>,
+        dist_server: &str,
+        max_retries: usize,
+        altered: bool,
+    ) -> impl futures_util::Stream<Item = impl Future<Output = Result<String>>> {
+        tokio_stream::iter(components).map({
+            move |bin| {
+                let sem = semaphore.clone();
+                let download_tx = download_tx.clone();
+                let new_manifest = Arc::clone(&new_manifest);
+                let download_cfg = download_cfg.clone();
+                async move {
+                    let _permit = sem.acquire().await.unwrap();
+                    let url = if altered {
+                        utils::parse_url(&bin.binary.url.replace(DEFAULT_DIST_SERVER, dist_server))?
+                    } else {
+                        utils::parse_url(&bin.binary.url)?
+                    };
+
+                    let installer_file = bin
+                        .download(&url, &download_cfg, max_retries, &new_manifest)
+                        .await?;
+                    let hash = bin.binary.hash.clone();
+                    let _ = download_tx.send(Ok((bin, installer_file))).await;
+                    Ok(hash)
+                }
+            }
+        })
+    }
+
+    async fn download_handle(
+        mut stream: impl futures_util::Stream<Item = Result<String>> + Unpin,
+        download_tx: mpsc::Sender<Result<(ComponentBinary, File)>>,
+    ) -> Vec<String> {
+        let mut hashes = Vec::new();
+        while let Some(result) = stream.next().await {
+            match result {
+                Ok(hash) => {
+                    hashes.push(hash);
+                }
+                Err(e) => {
+                    let _ = download_tx.send(Err(e)).await;
+                }
+            }
+        }
+        hashes
+    }
+
+    async fn install_handle(
+        self: Arc<Manifestation>,
+        mut download_rx: mpsc::Receiver<Result<(ComponentBinary, File)>>,
+        total_components: usize,
+        new_manifest: Arc<Manifest>,
+        download_cfg: Arc<DownloadCfg>,
+        tx: Transaction,
+    ) -> Result<Transaction> {
+        let mut current_tx = tx;
+        let mut counter = 0;
+        while counter < total_components
+            && let Some(message) = download_rx.recv().await
+        {
+            let (component_bin, installer_file) = message?;
+            current_tx = tokio::task::spawn_blocking({
+                let this = Arc::clone(&self);
+                let new_manifest = Arc::clone(&new_manifest);
+                let download_cfg = Arc::clone(&download_cfg);
+                move || {
+                    this.install_component(
+                        component_bin,
+                        installer_file,
+                        download_cfg,
+                        new_manifest,
+                        current_tx,
+                    )
+                }
+            })
+            .await??;
+            counter += 1;
+        }
+        Ok(current_tx)
+    }
 }
 
 #[derive(Debug)]