Feat: Refactor core download logic with concurrency and async features

Implement basic unit testing

Implement automatic java executable switching based on game version

Split loader module into smaller modules

Implement basic documentation

1 files changed, 301 insertions, 122 deletions

diff --git a/src/minecraft/downloads.rs b/src/minecraft/downloads.rs
index 7017d3f..8df994e 100644
--- a/src/minecraft/downloads.rs
+++ b/src/minecraft/downloads.rs
@@ -1,181 +1,360 @@
-use log::{debug, info};
-use reqwest::get;
+use std::{
+    collections::HashMap,
+    io::{Write, stdout},
+    path::{Path, PathBuf},
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+};
+
+use McError::Config;
+use Ordering::SeqCst;
+use futures::stream::{FuturesUnordered, StreamExt};
+use reqwest::Response;
 use serde::Deserialize;
 use tokio::{
-    fs::{self, File, create_dir_all},
+    fs::{File, create_dir_all, read_to_string, write},
     io::AsyncWriteExt,
+    spawn,
+    sync::Semaphore,
+    task::{JoinError, JoinHandle},
 };
 
 use crate::{
-    config::Config,
+    config::RuntimeConfig,
     errors::McError,
-    minecraft::manifests::{Library, Version},
-    platform::paths,
+    minecraft::manifests::{AssetIndex, Library, LibraryArtifact, Version},
+    platform::paths::{assets_directory, client_jar, library_file},
 };
 
+const MAX_CONCURRENT_DOWNLOADS: usize = 100;
+const PROGRESS_INITIAL_VALUE: usize = 0;
+const ASSET_URL_BASE: &str = "https://resources.download.minecraft.net/";
+const MAX_PROGRESS_PERCENT: f64 = 100.0;
+const PROGRESS_PRINT_WIDTH: usize = 3;
+const ATOMIC_ORDERING: Ordering = SeqCst;
+const DOWNLOAD_COMPLETE_MESSAGE: &str = "All downloads completed successfully!";
+const ASSET_INDEX_MISSING_ERROR: &str = "Missing asset_index for the version";
+
+impl From<JoinError> for McError {
+    fn from(error: JoinError) -> Self {
+        Config(format!("Task panicked: {}", error))
+    }
+}
+
+/// Represents a single file to download, including the URL and destination
+/// path.
+#[derive(Debug, Clone)]
+struct DownloadJob {
+    url: String,
+    destination_path: PathBuf,
+}
+
+impl DownloadJob {
+    fn already_exists(&self) -> bool { self.destination_path.exists() }
+}
+
+/// Represents a single asset entry in the Minecraft asset index.
 #[derive(Debug, Deserialize)]
 struct AssetObject {
     hash: String,
-    // size: u64,
 }
 
+/// The Minecraft asset index manifest.
 #[derive(Debug, Deserialize)]
 struct AssetIndexManifest {
-    objects: std::collections::HashMap<String, AssetObject>,
+    objects: HashMap<String, AssetObject>,
 }
 
-pub async fn download_all(
-    config: &Config,
-    version: &Version,
+/// Download all files required to run a specific Minecraft version.
+pub async fn download_all_files(
+    http_client: &reqwest::Client,
+    runtime_config: &RuntimeConfig,
+    version_info: &Version,
 ) -> Result<(), McError> {
-    download_client(config, version).await?;
-    download_libraries(config, &version.libraries).await?;
-    download_assets(config, version).await?;
+    let assets_directory_path: PathBuf =
+        ensure_assets_directories_exist(runtime_config).await?;
+    let asset_manifest: AssetIndexManifest = load_asset_index_manifest(
+        http_client,
+        &assets_directory_path,
+        version_info,
+    )
+    .await?;
+
+    let download_jobs: Vec<DownloadJob> = build_download_jobs(
+        runtime_config,
+        version_info,
+        &assets_directory_path,
+        &asset_manifest,
+    );
+
+    execute_download_jobs(http_client, download_jobs).await?;
+    println!("\n{}", DOWNLOAD_COMPLETE_MESSAGE);
+
     Ok(())
 }
 
-async fn download_client(
-    config: &Config,
-    version: &Version,
-) -> Result<(), McError> {
-    let jar_path = paths::client_jar(config, &version.id)?;
-
-    if jar_path.exists() {
-        debug!("Client jar already exists: {}", jar_path.display());
-        return Ok(());
-    }
-
-    info!("Downloading client {}", version.id);
-    download_file(&version.downloads.client.url, &jar_path).await
+/// Ensure the essential assets directories exist.
+async fn ensure_assets_directories_exist(
+    config: &RuntimeConfig,
+) -> Result<PathBuf, McError> {
+    let assets_dir = assets_directory(config);
+    create_dir_all(assets_dir.join("objects")).await?;
+    create_dir_all(assets_dir.join("indexes")).await?;
+    Ok(assets_dir)
 }
 
-async fn download_libraries(
-    config: &Config,
-    libraries: &[Library],
-) -> Result<(), McError> {
-    for library in libraries {
-        if let Some(artifact) = &library.downloads.artifact {
-            let library_path = paths::library_file(config, &artifact.path)?;
-
-            if !library_path.exists() {
-                info!("Downloading library {}", artifact.path);
-                download_file(&artifact.url, &library_path).await?;
-            }
-        }
-
-        if let Some(classifiers) = &library.downloads.classifiers {
-            for (_, native) in classifiers {
-                let native_path = paths::library_file(config, &native.path)?;
+/// Load the asset index manifest for the given Minecraft version.
+async fn load_asset_index_manifest(
+    http_client: &reqwest::Client,
+    assets_dir: &Path,
+    version_info: &Version,
+) -> Result<AssetIndexManifest, McError> {
+    let asset_index: &AssetIndex = version_info
+        .asset_index
+        .as_ref()
+        .ok_or_else(|| Config(ASSET_INDEX_MISSING_ERROR.into()))?;
 
-                if native_path.exists() {
-                    continue;
-                }
+    let index_file_path: PathBuf = assets_dir
+        .join("indexes")
+        .join(format!("{}.json", asset_index.id));
 
-                info!("Downloading native library {}", native.path);
-                download_file(&native.url, &native_path).await?;
-            }
-        }
+    if !index_file_path.exists() {
+        download_text_file(http_client, &asset_index.url, &index_file_path)
+            .await?;
     }
 
-    Ok(())
+    let json_string: String = read_to_string(index_file_path).await?;
+    Ok(serde_json::from_str(&json_string)?)
 }
 
-async fn download_asset_index(
-    config: &Config,
-    version: &Version,
-) -> Result<AssetIndexManifest, McError> {
-    let assets_dir = paths::assets_dir(config);
-    create_dir_all(assets_dir.join("indexes")).await?;
+fn build_download_jobs(
+    config: &RuntimeConfig,
+    version_info: &Version,
+    assets_dir: &Path,
+    asset_manifest: &AssetIndexManifest,
+) -> Vec<DownloadJob> {
+    let mut jobs: Vec<DownloadJob> = Vec::new();
 
-    let asset_index = version.asset_index.as_ref().ok_or_else(|| {
-        McError::Config("Missing asset_index in version.json".into())
-    })?;
+    add_client_download_job(&mut jobs, config, version_info);
+    add_library_download_jobs(&mut jobs, config, version_info);
+    add_asset_download_jobs(&mut jobs, assets_dir, asset_manifest);
 
-    if asset_index.id == "legacy" {
-        return Err(McError::Config(
-            "Legacy assetIndex detected – pobierz właściwy version.json".into(),
-        ));
-    }
+    jobs
+}
 
-    let index_path = assets_dir
-        .join("indexes")
-        .join(format!("{}.json", asset_index.id));
+fn add_client_download_job(
+    jobs: &mut Vec<DownloadJob>,
+    config: &RuntimeConfig,
+    version_info: &Version,
+) {
+    jobs.push(DownloadJob {
+        url: version_info.downloads.client.url.clone(),
+        destination_path: client_jar(config, &version_info.id),
+    });
+}
 
-    if index_path.exists() {
-        let index_data = fs::read_to_string(&index_path).await?;
-        let manifest: AssetIndexManifest = serde_json::from_str(&index_data)?;
-        return Ok(manifest);
+fn add_library_download_jobs(
+    jobs: &mut Vec<DownloadJob>,
+    config: &RuntimeConfig,
+    version_info: &Version,
+) {
+    for library in &version_info.libraries {
+        add_library_artifact_job(jobs, config, library);
+        add_library_classifier_jobs(jobs, config, library);
     }
+}
+fn add_library_artifact_job(
+    jobs: &mut Vec<DownloadJob>,
+    config: &RuntimeConfig,
+    library: &Library,
+) {
+    let artifact: &LibraryArtifact = match &library.downloads.artifact {
+        | Some(a) => a,
+        | None => return,
+    };
+
+    jobs.push(DownloadJob {
+        url: artifact.url.clone(),
+        destination_path: library_file(config, &artifact.path),
+    });
+}
 
-    info!("Downloading asset index {}", asset_index.id);
-    let response = get(&asset_index.url).await?;
-    let manifest_text = response.text().await?;
-
-    fs::write(&index_path, &manifest_text).await?;
+fn add_library_classifier_jobs(
+    jobs: &mut Vec<DownloadJob>,
+    config: &RuntimeConfig,
+    library: &Library,
+) {
+    let classifiers: &HashMap<String, LibraryArtifact> =
+        match &library.downloads.classifiers {
+            | Some(values) => values,
+            | None => return,
+        };
+
+    for classifier_entry in classifiers.values() {
+        jobs.push(DownloadJob {
+            url: classifier_entry.url.clone(),
+            destination_path: library_file(config, &classifier_entry.path),
+        });
+    }
+}
 
-    let manifest: AssetIndexManifest = serde_json::from_str(&manifest_text)?;
-    Ok(manifest)
+fn add_asset_download_jobs(
+    jobs: &mut Vec<DownloadJob>,
+    assets_dir: &Path,
+    asset_manifest: &AssetIndexManifest,
+) {
+    for asset_object in asset_manifest.objects.values() {
+        let prefix: &str = &asset_object.hash[0..2];
+        jobs.push(DownloadJob {
+            url: format!("{}{}/{}", ASSET_URL_BASE, prefix, asset_object.hash),
+            destination_path: assets_dir
+                .join("objects")
+                .join(prefix)
+                .join(&asset_object.hash),
+        });
+    }
 }
 
-async fn download_assets(
-    config: &Config,
-    version: &Version,
+async fn execute_download_jobs(
+    http_client: &reqwest::Client,
+    download_jobs: Vec<DownloadJob>,
 ) -> Result<(), McError> {
-    let assets_dir = paths::assets_dir(config);
-
-    create_dir_all(assets_dir.join("objects")).await?;
-    create_dir_all(assets_dir.join("indexes")).await?;
-
-    let manifest = download_asset_index(config, version).await?;
+    let total_jobs_count: usize = download_jobs.len();
+    let completed_jobs_count: Arc<AtomicUsize> =
+        Arc::new(AtomicUsize::new(PROGRESS_INITIAL_VALUE));
+    let concurrent_download_semaphore: Arc<Semaphore> =
+        Arc::new(Semaphore::new(MAX_CONCURRENT_DOWNLOADS));
+
+    let mut tasks: FuturesUnordered<JoinHandle<Result<(), McError>>> =
+        spawn_missing_download_jobs(
+            http_client,
+            download_jobs,
+            &completed_jobs_count,
+            total_jobs_count,
+            concurrent_download_semaphore,
+        );
 
-    for (logical_path, asset) in &manifest.objects {
-        let subdir = &asset.hash[0..2];
-        let file_path = assets_dir
-            .join("objects")
-            .join(subdir)
-            .join(&asset.hash);
+    await_download_tasks(&mut tasks).await
+}
 
-        if file_path.exists() {
+fn spawn_missing_download_jobs(
+    http_client: &reqwest::Client,
+    download_jobs: Vec<DownloadJob>,
+    completed_jobs_counter: &Arc<AtomicUsize>,
+    total_jobs_count: usize,
+    concurrent_download_semaphore: Arc<Semaphore>,
+) -> FuturesUnordered<JoinHandle<Result<(), McError>>> {
+    let download_tasks: FuturesUnordered<JoinHandle<Result<(), McError>>> =
+        FuturesUnordered::new();
+
+    for job in download_jobs {
+        if job.already_exists() {
+            completed_jobs_counter.fetch_add(1, SeqCst);
+            print_download_progress(completed_jobs_counter, total_jobs_count);
             continue;
         }
 
-        let url = format!(
-            "https://resources.download.minecraft.net/{}/{}",
-            subdir, asset.hash
-        );
-        info!("Downloading asset {} -> {}", logical_path, file_path.display());
-        download_file(&url, &file_path).await?;
+        download_tasks.push(spawn_download_job(
+            http_client.clone(),
+            job,
+            concurrent_download_semaphore.clone(),
+            completed_jobs_counter.clone(),
+            total_jobs_count,
+        ));
     }
 
-    if let Some(asset) = manifest.objects.get("sounds.json") {
-        let file_path = assets_dir.join("indexes").join("sounds.json");
-        if !file_path.exists() {
-            let subdir = &asset.hash[0..2];
-            let url = format!(
-                "https://resources.download.minecraft.net/{}/{}",
-                subdir, asset.hash
-            );
-            info!("Downloading sounds.json");
-            download_file(&url, &file_path).await?;
-        }
+    download_tasks
+}
+
+async fn await_download_tasks(
+    tasks: &mut FuturesUnordered<JoinHandle<Result<(), McError>>>,
+) -> Result<(), McError> {
+    while let Some(task_result) = tasks.next().await {
+        task_result.map_err(McError::from)??;
     }
+    Ok(())
+}
 
+fn spawn_download_job(
+    http_client: reqwest::Client,
+    download_job: DownloadJob,
+    concurrent_download_semaphore: Arc<Semaphore>,
+    completed_jobs_count: Arc<AtomicUsize>,
+    total_jobs_count: usize,
+) -> JoinHandle<Result<(), McError>> {
+    spawn(async move {
+        let _permit = concurrent_download_semaphore
+            .acquire_owned()
+            .await
+            .unwrap();
+        download_file(&http_client, &download_job).await?;
+        completed_jobs_count.fetch_add(1, SeqCst);
+        print_download_progress(&completed_jobs_count, total_jobs_count);
+        Ok(())
+    })
+}
+
+/// Print the current progress of downloads to stdout.
+///
+/// # Parameters
+/// - `completed_jobs_count`: Atomic counter of completed download jobs.
+/// - `total_jobs_count`: Total number of jobs being processed.
+fn print_download_progress(
+    completed_jobs_count: &AtomicUsize,
+    total_jobs_count: usize,
+) {
+    let completed_jobs: usize = completed_jobs_count.load(ATOMIC_ORDERING);
+    let progress_percentage: f64 = ((completed_jobs as f64
+        / total_jobs_count as f64)
+        * MAX_PROGRESS_PERCENT)
+        .min(MAX_PROGRESS_PERCENT);
+
+    print!(
+        "\rDownloading game files: {:>width$.0}%",
+        progress_percentage,
+        width = PROGRESS_PRINT_WIDTH
+    );
+    stdout().flush().unwrap();
+}
+
+async fn download_text_file(
+    http_client: &reqwest::Client,
+    file_url: &str,
+    destination_path: &PathBuf,
+) -> Result<(), McError> {
+    let text_content = http_client
+        .get(file_url)
+        .send()
+        .await?
+        .error_for_status()?
+        .text()
+        .await?;
+    write(destination_path, text_content).await?;
     Ok(())
 }
 
 async fn download_file(
-    url: &str,
-    path: &std::path::Path,
+    http_client: &reqwest::Client,
+    download_job: &DownloadJob,
 ) -> Result<(), McError> {
-    if let Some(parent) = path.parent() {
-        create_dir_all(parent).await?;
+    if let Some(parent_dir) = download_job.destination_path.parent() {
+        create_dir_all(parent_dir).await?;
     }
 
-    let response = get(url).await?;
-    let bytes = response.bytes().await?;
-
-    let mut file = File::create(path).await?;
-    file.write_all(&bytes).await?;
+    let response: Response = http_client
+        .get(&download_job.url)
+        .send()
+        .await?
+        .error_for_status()?;
+    let mut byte_stream = response.bytes_stream();
+    let mut file_handle: File =
+        File::create(&download_job.destination_path).await?;
+
+    while let Some(chunk) = byte_stream.next().await {
+        file_handle.write_all(&chunk?).await?;
+    }
 
     Ok(())
 }