Use stream to do concurrency instead of spawning threads

api/src/archive/task.rs

@@ -136,8 +136,10 @@ async fn start_create_task(
     // Downloading mod files.
     update_task_progress(&task_id, Some(ArchiveTaskStage::Downloading), 0.1);
 
+    let task_id_clone = task_id.clone();
+
     download_files(&downloads, max_simultaneous_downloads, |progress| {
-        update_task_progress(&task_id, None, 0.1 + progress * 0.75)
+        update_task_progress(&task_id_clone, None, 0.1 + progress * 0.75)
     })
     .await?;
 

service/Cargo.toml

@@ -25,3 +25,5 @@ zip = "0.6.6"
 serde_json = "1.0.108"
 regex = "1.10.2"
 semver = "1.0.20"
+tokio-stream = "0.1.14"
+futures = "0.3.30"

service/src/archive/task.rs

@@ -9,9 +9,13 @@ use std::{
 
 use entity::{
     entry::text_entry,
-    minecraft::{mod_loader::{ModLoader, ModLoaderVec}, minecraft_mod},
+    minecraft::{
+        minecraft_mod,
+        mod_loader::{ModLoader, ModLoaderVec},
+    },
     misc::StringVec,
 };
+use futures::StreamExt;
 use lazy_static::lazy_static;
 use sea_orm::{sea_query::OnConflict, DatabaseConnection, EntityTrait, Set};
 use serde::Serialize;
@@ -77,32 +81,26 @@ pub async fn download_files(
 ) -> anyhow::Result<()> {
     let total_size: usize = downloads.iter().map(|x| x.size).sum();
     let mut downloaded_size = 0;
-
     create_dir_all(get_archives_directory())?;
 
-    for chuck in downloads.chunks(max_simultaneous_downloads) {
-        let mut handles = Vec::with_capacity(chuck.len());
-
-        for (index, info) in chuck.iter().enumerate() {
-            let url = info.url.clone();
-            let path = info.path.clone();
-
-            let handle = tokio::spawn(async move {
-                let bytes = reqwest::get(url).await?.bytes().await?;
-                tokio::fs::write(path, bytes).await?;
+    let mut handles = Vec::new();
+    for info in downloads {
+        let url = &info.url;
+        let path = &info.path;
+        handles.push(async move {
+            let bytes = reqwest::get(url).await?.bytes().await?;
+            tokio::fs::write(path, bytes).await?;
 
-                Ok::<_, anyhow::Error>(index)
-            });
-            handles.push(handle);
-        }
+            Ok::<_, anyhow::Error>(info.size)
+        })
+    }
 
-        for handle in handles {
-            let index = handle.await??;
-            let info = chuck.get(index).unwrap();
+    let mut stream = tokio_stream::iter(handles).buffer_unordered(max_simultaneous_downloads);
 
-            downloaded_size += info.size;
-            progress_changed(downloaded_size as f32 / total_size as f32);
-        }
+    while let Some(x) = stream.next().await {
+        let download_size = x?;
+        downloaded_size += download_size;
+        progress_changed(downloaded_size as f32 / total_size as f32);
     }
 
     Ok(())