fs 0.0.6

use std::fmt;
use std::fs::{File, Metadata};
use std::io::{self, Read};
use std::mem;
use std::path::PathBuf;
use std::pin::Pin;
use std::task::{Context, Poll};

use bytes::Bytes;
use futures::Stream;

const BUF_SIZE: usize = 8192;

/// Options for how to read the file.
///
/// The default is to automatically determine the buffer size.
#[derive(Clone, Debug, Default)]
pub struct ReadOptions {
    /// The buffer size to use.
    ///
    /// If set to `None`, this is automatically determined from the operating system.
    buffer_size: Option<usize>,
}

impl ReadOptions {
    /// The buffer size to use when reading.
    ///
    /// Default is automatically determined from the operating system.
    ///
    /// # Panics
    ///
    /// The passed argument must be larger than 0.
    pub fn buffer_size(mut self, buffer_size: usize) -> Self {
        assert!(buffer_size > 0, "buffer size must be larger than 0");
        self.buffer_size = Some(buffer_size);
        self
    }
}

pub(crate) fn new(pool: &crate::FsPool, path: PathBuf, opts: ReadOptions) -> FsReadStream {
    FsReadStream {
        path,
        pool: pool.clone(),
        state: State::Init(opts.buffer_size),
    }
}

pub(crate) fn new_from_file(pool: &crate::FsPool, file: File, opts: ReadOptions) -> FsReadStream {
    let final_buf_size = finalize_buf_size(opts.buffer_size, &file);
    FsReadStream {
        path: PathBuf::new(),
        pool: pool.clone(),
        state: State::Ready(file, final_buf_size),
    }
}

/// A stream of byte chunks read from a target file.
#[must_use = "streams do nothing unless polled"]
pub struct FsReadStream {
    path: PathBuf,
    pool: crate::FsPool,
    state: State,
}

enum State {
    Init(Option<usize>),
    Working(crate::FsFuture<(File, Bytes, usize)>),
    Ready(File, usize),
    Eof,
}

impl Stream for FsReadStream {
    type Item = io::Result<Bytes>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        loop {
            match mem::replace(&mut this.state, State::Eof) {
                State::Init(buf_size) => {
                    let path = this.path.clone();
                    this.state =
                        State::Working(this.pool.spawn(move || open_and_read(path, buf_size)));
                }
                State::Working(mut operation) => match Pin::new(&mut operation).poll(cx) {
                    Poll::Pending => {
                        this.state = State::Working(operation);
                        return Poll::Pending;
                    }
                    Poll::Ready(Ok((_, chunk, _))) if chunk.is_empty() => {
                        this.state = State::Eof;
                        return Poll::Ready(None);
                    }
                    Poll::Ready(Ok((file, chunk, buf_size))) => {
                        this.state = State::Ready(file, buf_size);
                        return Poll::Ready(Some(Ok(chunk)));
                    }
                    Poll::Ready(Err(error)) => {
                        this.state = State::Eof;
                        return Poll::Ready(Some(Err(error)));
                    }
                },
                State::Ready(file, buf_size) => {
                    this.state =
                        State::Working(this.pool.spawn(move || read_chunk(file, buf_size)));
                }
                State::Eof => return Poll::Ready(None),
            }
        }
    }
}

impl fmt::Debug for FsReadStream {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("FsReadStream")
            .field("path", &self.path)
            .finish()
    }
}

fn read_chunk(mut file: File, buf_size: usize) -> io::Result<(File, Bytes, usize)> {
    let mut buffer = vec![0; buf_size];
    let read = file.read(&mut buffer)?;
    buffer.truncate(read);
    Ok((file, Bytes::from(buffer), buf_size))
}

fn finalize_buf_size(buf_size: Option<usize>, file: &File) -> usize {
    match file.metadata() {
        Ok(metadata) => {
            let buf_size = buf_size.unwrap_or_else(|| get_block_size(&metadata)).max(1);
            let file_size = usize::try_from(metadata.len()).unwrap_or(usize::MAX);
            if file_size == 0 {
                buf_size
            } else {
                file_size.min(buf_size)
            }
        }
        Err(_) => buf_size.unwrap_or(BUF_SIZE),
    }
}

fn open_and_read(path: PathBuf, buf_size: Option<usize>) -> io::Result<(File, Bytes, usize)> {
    let file = File::open(path)?;
    let final_buf_size = finalize_buf_size(buf_size, &file);
    read_chunk(file, final_buf_size)
}

#[cfg(unix)]
fn get_block_size(metadata: &Metadata) -> usize {
    use std::os::unix::fs::MetadataExt;
    usize::try_from(metadata.blksize()).unwrap_or(BUF_SIZE)
}

#[cfg(not(unix))]
fn get_block_size(_metadata: &Metadata) -> usize {
    BUF_SIZE
}