use std::{error::Error, time::Instant};

use bytes::Buf;
use futures_util::StreamExt;
use opendal::Operator;

fn main() -> Result<(), Box<dyn Error>> {
    let root = std::env::current_dir()?;
    let root_str = root.to_str().ok_or("Invalid path")?;
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("Failed to create Tokio runtime");

    let builder = opendal::services::Fs::default().root(root_str);

    let op = Operator::new(builder)?.finish();

    let join_handle = runtime.spawn(async move {
        let mut stream = op
            .reader_with("large_file.dat")
            .concurrent(4)
            .chunk(4 * 1024 * 1024)
            .await
            .expect("Failed to create reader")
            .into_stream(..)
            .await
            .expect("Failed to create stream");

        let mut byte_count = 0;
        let mut read_count = 0;

        let start_time = Instant::now();
        while let Some(chunk) = stream.next().await {
            let b = chunk
                .inspect_err(|e| {
                    eprintln!("Error reading chunk: {}", e);
                })
                .unwrap();
            byte_count += b.remaining();
            read_count += 1;
        }
        let end_time = Instant::now();
        let duration = end_time.duration_since(start_time);
        let duration_secs = duration.as_secs_f64();
        let throughput = byte_count as f64 / duration_secs / (1024.0 * 1024.0); // MB/s

        println!("Read {} bytes in {} chunks", byte_count, read_count);
        println!("Total time: {:.2} seconds", duration_secs);
        println!("Throughput: {:.2} MB/s", throughput);
    });

    // Wait for the async task to complete
    runtime.block_on(join_handle)?;

    Ok(())
}