Concurrent File Downloader

Example Code

use tokio::fs::{File as AsyncFile, OpenOptions};
use tokio::io::{AsyncWriteExt, AsyncSeekExt};
use tokio::task;
use reqwest::Client;
use std::path::PathBuf;
use futures::future::join_all;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let url = "https://speed.hetzner.de/100MB.bin"; // Example URL for a 100MB test file
    let output_filename = "downloaded_file.bin";
    let num_concurrent_chunks = 8; // Desired number of parallel download chunks

    println!("Starting concurrent download of {} with {} chunks...", url, num_concurrent_chunks);

    let client = Client::new();
    
    // 1. Get total file size using a HEAD request
    let res = client.head(url).send().await?;
    let total_size: u64 = res
        .headers()
        .get("content-length")
        .and_then(|h| h.to_str().ok())
        .and_then(|s| s.parse().ok())
        .ok_or("Failed to get content-length from HEAD request")?;

    println!("Total file size: {} bytes", total_size);

    if total_size == 0 {
        println!("File is empty, nothing to download.");
        return Ok(());
    }

    // 2. Create or truncate the output file and set its length
    let file_path = PathBuf::from(output_filename);
    let mut file = AsyncFile::create(&file_path).await?;
    file.set_len(total_size).await?; // Pre-allocate file size to avoid fragmented writes
    drop(file); // Close the file handle as we'll open it per-chunk write via spawn_blocking

    // 3. Determine actual number of chunks and their sizes
    let actual_num_chunks = num_concurrent_chunks.min(total_size as usize); // Cap chunks to total_size if file is very small
    let base_chunk_size = total_size / actual_num_chunks as u64;
    let remainder = total_size % actual_num_chunks as u64;

    let mut tasks = Vec::new();
    let mut current_offset = 0;

    for i in 0..actual_num_chunks {
        let start = current_offset;
        let mut chunk_len = base_chunk_size;
        if (i as u64) < remainder { // Distribute remainder bytes to initial chunks
            chunk_len += 1;
        }
        let end = start + chunk_len - 1;

        // Clone necessary variables for the async task
        let client = client.clone();
        let file_path = file_path.clone();
        let url = url.to_string();

        let task = task::spawn(async move {
            println!("Downloading chunk {} (bytes {}-{})", i, start, end);
            let range = format!("bytes={}-{}", start, end);
            let res = client.get(&url).header("Range", range).send().await;

            match res {
                Ok(response) => {
                    let status = response.status();
                    if !status.is_success() && status != reqwest::StatusCode::PARTIAL_CONTENT {
                        eprintln!("Error: Received non-success status for chunk {}: {}", i, status);
                        return Err(format!("Bad status code for chunk {}: {}", i, status));
                    }
                    let bytes = response.bytes().await.map_err(|e| format!("Failed to read bytes for chunk {}: {}", i, e))?;
                    
                    // Use spawn_blocking for synchronous file I/O to avoid blocking the async runtime
                    task::spawn_blocking(move || {
                        let mut file = OpenOptions::new()
                            .write(true)
                            .open(&file_path)
                            .map_err(|e| format!("Failed to open file for writing chunk {}: {}", i, e))?;
                        
                        file.seek(std::io::SeekFrom::Start(start))
                            .map_err(|e| format!("Failed to seek file for chunk {}: {}", i, e))?;
                        
                        file.write_all(&bytes)
                            .map_err(|e| format!("Failed to write bytes for chunk {}: {}", i, e))?;
                        
                        println!("Finished writing chunk {} (bytes {}-{})", i, start, end);
                        Ok(())
                    }).await.map_err(|e| format!("JoinError on spawn_blocking for chunk {}: {}", i, e))??; // Two ?? to unpack nested results
                    Ok(())
                },
                Err(e) => {
                    eprintln!("Error downloading chunk {} (bytes {}-{}): {}", i, start, end, e);
                    Err(format!("Download error for chunk {}: {}", i, e))
                }
            }
        });
        tasks.push(task);
        current_offset += chunk_len;
    }

    let results = join_all(tasks).await;

    let mut successful_downloads = 0;
    let mut failed_downloads = 0;

    for result in results {
        match result {
            Ok(Ok(_)) => successful_downloads += 1,
            Ok(Err(e)) => {
                eprintln!("Chunk download failed: {}", e);
                failed_downloads += 1;
            },
            Err(e) => {
                eprintln!("Task join error: {}", e);
                failed_downloads += 1;
            }
        }
    }

    if failed_downloads == 0 {
        println!("\nConcurrent download completed successfully!");
        println!("File saved as: {}", output_filename);
    } else {
        eprintln!("\nConcurrent download completed with {} successful and {} failed chunks.", successful_downloads, failed_downloads);
        eprintln!("The downloaded file might be incomplete or corrupted: {}", output_filename);
    }

    Ok(())
}