fs 0.0.6

#![deny(missing_debug_implementations, missing_docs, unsafe_code)]

//! Asynchronous filesystem operations backed by a dedicated blocking thread pool.
//!
//! # Examples
//!
//! ```no_run
//! use fs::FsPool;
//! use futures::executor::block_on;
//! use futures::StreamExt;
//! use std::io;
//!
//! fn main() -> io::Result<()> {
//!     let fs = FsPool::default();
//!
//!     block_on(async {
//!         let reader = fs.read("input.txt", Default::default());
//!         let writer = fs.write("output.txt", Default::default());
//!         reader.forward(writer).await
//!     })
//! }
//! ```

use std::fmt;
use std::fs;
use std::future::Future;
use std::io;
use std::path::Path;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use futures::channel::oneshot::{self, Receiver};
use futures::executor::ThreadPool;
use futures::task::{Spawn, SpawnExt};

pub use self::read::{FsReadStream, ReadOptions};
pub use self::write::{FsWriteSink, WriteOptions};

mod read;
mod write;

/// A pool of threads to handle file IO.
#[derive(Clone)]
pub struct FsPool {
    executor: Arc<dyn Spawn + Send + Sync>,
}

impl FsPool {
    /// Creates a new `FsPool`, with the supplied number of threads.
    ///
    /// Each thread is dedicated to blocking filesystem work. The thread count
    /// must be greater than zero.
    ///
    /// # Panics
    ///
    /// Panics if `threads` is zero or the worker threads cannot be created.
    pub fn new(threads: usize) -> Self {
        assert!(
            threads > 0,
            "filesystem worker count must be greater than zero"
        );

        let executor = ThreadPool::builder()
            .pool_size(threads)
            .name_prefix("futures-fs-")
            .create()
            .expect("failed to create filesystem worker pool");

        Self::with_executor(executor)
    }

    /// Creates an `FsPool` from an existing futures [`Spawn`] implementation.
    ///
    /// Filesystem operations block while running. The supplied executor should
    /// therefore be reserved for blocking work rather than shared with
    /// latency-sensitive asynchronous tasks.
    pub fn with_executor<E>(executor: E) -> Self
    where
        E: Spawn + Send + Sync + 'static,
    {
        Self {
            executor: Arc::new(executor),
        }
    }

    #[doc(hidden)]
    #[deprecated(since = "0.0.5", note = "use FsPool::with_executor")]
    pub fn from_executor<E>(executor: E) -> Self
    where
        E: Spawn + Send + Sync + 'static,
    {
        Self::with_executor(executor)
    }

    /// Returns a stream of byte chunks read from the file at `path`.
    pub fn read<P>(&self, path: P, opts: ReadOptions) -> FsReadStream
    where
        P: AsRef<Path>,
    {
        read::new(self, path.as_ref().to_owned(), opts)
    }

    /// Returns a stream of byte chunks read from an open file.
    pub fn read_file(&self, file: fs::File, opts: ReadOptions) -> FsReadStream {
        read::new_from_file(self, file, opts)
    }

    /// Returns a sink that writes byte chunks to the file at `path`.
    pub fn write<P>(&self, path: P, opts: WriteOptions) -> FsWriteSink
    where
        P: AsRef<Path>,
    {
        write::new(self, path.as_ref().to_owned(), opts)
    }

    /// Returns a sink that writes byte chunks to an open file.
    pub fn write_file(&self, file: fs::File) -> FsWriteSink {
        write::new_from_file(self, file)
    }

    /// Returns a future that resolves when the target file is deleted.
    pub fn delete<P>(&self, path: P) -> FsFuture<()>
    where
        P: AsRef<Path>,
    {
        let path = path.as_ref().to_owned();
        self.spawn(move || fs::remove_file(path))
    }

    pub(crate) fn spawn<T, F>(&self, operation: F) -> FsFuture<T>
    where
        T: Send + 'static,
        F: FnOnce() -> io::Result<T> + Send + 'static,
    {
        let (sender, receiver) = oneshot::channel();
        let task = async move {
            let _ = sender.send(operation());
        };
        let _ = self.executor.spawn(task);
        FsFuture { inner: receiver }
    }
}

impl Default for FsPool {
    fn default() -> Self {
        Self::new(4)
    }
}

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

/// A future representing work in the `FsPool`.
#[must_use = "futures do nothing unless polled or awaited"]
pub struct FsFuture<T> {
    inner: Receiver<io::Result<T>>,
}

impl<T> Future for FsFuture<T> {
    type Output = io::Result<T>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        match Pin::new(&mut this.inner).poll(cx) {
            Poll::Ready(Ok(result)) => Poll::Ready(result),
            Poll::Ready(Err(_)) => Poll::Ready(Err(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "filesystem worker pool stopped before completing the operation",
            ))),
            Poll::Pending => Poll::Pending,
        }
    }
}

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