diff --git a/Cargo.toml b/Cargo.toml index 8790d71bb..a26518c7b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,6 +24,7 @@ keywords = ["io", "async", "non-blocking", "futures"] members = [ "./", "tokio-executor", + "tokio-fs", "tokio-io", "tokio-reactor", "tokio-threadpool", @@ -45,6 +46,7 @@ tokio-threadpool = { version = "0.1.2", path = "tokio-threadpool" } tokio-tcp = { version = "0.1.0", path = "tokio-tcp" } tokio-udp = { version = "0.1.0", path = "tokio-udp" } tokio-timer = { version = "0.2.1", path = "tokio-timer" } +tokio-fs = { version = "0.1.0", path = "tokio-fs" } futures = "0.1.20" diff --git a/src/fs.rs b/src/fs.rs new file mode 100644 index 000000000..f3d5eb243 --- /dev/null +++ b/src/fs.rs @@ -0,0 +1,13 @@ +//! Asynchronous filesystem manipulation operations. +//! +//! This module contains basic methods and types for manipulating the contents +//! of the local filesystem from within the context of the Tokio runtime. +//! +//! Unlike *most* other Tokio APIs, the filesystem APIs **must** be used from +//! the context of the Tokio runtime as they require Tokio specific features to +//! function. + +pub use tokio_fs::{ + file, + File, +}; diff --git a/src/lib.rs b/src/lib.rs index 024fa45f0..96449bae8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ //! * A [reactor][reactor] backed by the operating system's event queue (epoll, kqueue, //! IOCP, etc...). //! * Asynchronous [TCP and UDP][net] sockets. +//! * Asynchronous [filesystem][fs] operations. //! * [Timer][timer] API for scheduling work in the future. //! //! Tokio is built using [futures] as the abstraction for managing the @@ -71,6 +72,7 @@ extern crate futures; extern crate mio; extern crate tokio_io; extern crate tokio_executor; +extern crate tokio_fs; extern crate tokio_reactor; extern crate tokio_threadpool; extern crate tokio_timer; @@ -81,6 +83,7 @@ extern crate tokio_udp; extern crate futures2; pub mod executor; +pub mod fs; pub mod net; pub mod reactor; pub mod runtime; @@ -100,15 +103,35 @@ pub mod io { //! defines two traits, [`AsyncRead`] and [`AsyncWrite`], which extend the //! `Read` and `Write` traits of the standard library. //! + //! # AsyncRead and AsyncWrite + //! //! [`AsyncRead`] and [`AsyncWrite`] must only be implemented for //! non-blocking I/O types that integrate with the futures type system. In //! other words, these types must never block the thread, and instead the //! current task is notified when the I/O resource is ready. //! + //! # Standard input and output + //! + //! Tokio provides asynchronous APIs to standard [input], [output], and [error]. + //! These APIs are very similar to the ones provided by `std`, but they also + //! implement [`AsyncRead`] and [`AsyncWrite`]. + //! + //! Unlike *most* other Tokio APIs, the standard input / output APIs + //! **must** be used from the context of the Tokio runtime as they require + //! Tokio specific features to function. + //! + //! [input]: fn.stdin.html + //! [output]: fn.stdout.html + //! [error]: fn.stderr.html + //! + //! # Utility functions + //! //! Utilities functions are provided for working with [`AsyncRead`] / //! [`AsyncWrite`] types. For example, [`copy`] asynchronously copies all //! data from a source to a destination. //! + //! # `std` re-exports + //! //! Additionally, [`Read`], [`Write`], [`Error`], [`ErrorKind`], and //! [`Result`] are re-exported from `std::io` for ease of use. //! @@ -126,6 +149,16 @@ pub mod io { AsyncWrite, }; + // standard input, output, and error + pub use tokio_fs::{ + stdin, + Stdin, + stdout, + Stdout, + stderr, + Stderr, + }; + // Utils pub use tokio_io::io::{ copy, diff --git a/tokio-fs/CHANGELOG.md b/tokio-fs/CHANGELOG.md new file mode 100644 index 000000000..fb22aa7c8 --- /dev/null +++ b/tokio-fs/CHANGELOG.md @@ -0,0 +1,3 @@ +# 0.1.0 (unreleased) + +* Initial release diff --git a/tokio-fs/Cargo.toml b/tokio-fs/Cargo.toml new file mode 100644 index 000000000..7bc4db67d --- /dev/null +++ b/tokio-fs/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "tokio-fs" + +# When releasing to crates.io: +# - Update html_root_url. +# - Update CHANGELOG.md. +# - Create "v0.1.x" git tag. +version = "0.1.0" +authors = ["Carl Lerche "] +license = "MIT" +readme = "README.md" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-fs/0.1" +description = """ +Filesystem API for Tokio. +""" +keywords = ["tokio", "futures", "fs", "file", "async"] +categories = ["asynchronous", "network-programming", "filesystem"] + +[dependencies] +futures = "0.1.21" +# TODO: Set real version +tokio-threadpool = { version = "0.1.1", path = "../tokio-threadpool" } +tokio-io = { version = "0.1.6", path = "../tokio-io" } + +[dev-dependencies] +rand = "0.4.2" +tempdir = "0.3.7" +tokio-io = { version = "0.1.6", path = "../tokio-io" } diff --git a/tokio-fs/LICENSE b/tokio-fs/LICENSE new file mode 100644 index 000000000..38c1e27b8 --- /dev/null +++ b/tokio-fs/LICENSE @@ -0,0 +1,25 @@ +Copyright (c) 2018 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/tokio-fs/README.md b/tokio-fs/README.md new file mode 100644 index 000000000..f359d7676 --- /dev/null +++ b/tokio-fs/README.md @@ -0,0 +1,19 @@ +# Tokio FS + +Asynchronous filesystem manipulation operations (and stdin, stdout, stderr). + +[Documentation](https://tokio-rs.github.io/tokio/tokio_fs/) + +## Overview + +This crate provides filesystem manipulation facilities for usage with Tokio. + +## License + +This project is licensed under the [MIT license](LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/tokio-fs/examples/std-echo.rs b/tokio-fs/examples/std-echo.rs new file mode 100644 index 000000000..cdfdf0080 --- /dev/null +++ b/tokio-fs/examples/std-echo.rs @@ -0,0 +1,47 @@ +//! Echo everything received on STDIN to STDOUT. + +extern crate futures; +extern crate tokio_fs; +extern crate tokio_io; +extern crate tokio_threadpool; + +use tokio_fs::{stdin, stdout, stderr}; +use tokio_io::codec::{FramedRead, FramedWrite, LinesCodec}; +use tokio_threadpool::Builder; + +use futures::{Future, Stream, Sink}; + +use std::io; + +pub fn main() { + let pool = Builder::new() + .pool_size(1) + .build(); + + pool.spawn({ + let input = FramedRead::new(stdin(), LinesCodec::new()); + + let output = FramedWrite::new(stdout(), LinesCodec::new()) + .with(|line: String| { + let mut out = "OUT: ".to_string(); + out.push_str(&line); + Ok::<_, io::Error>(out) + }); + + let error = FramedWrite::new(stderr(), LinesCodec::new()) + .with(|line: String| { + let mut out = "ERR: ".to_string(); + out.push_str(&line); + Ok::<_, io::Error>(out) + }); + + let dst = output.fanout(error); + + input + .forward(dst) + .map(|_| ()) + .map_err(|e| panic!("io error = {:?}", e)) + }); + + pool.shutdown_on_idle().wait().unwrap(); +} diff --git a/tokio-fs/src/file/create.rs b/tokio-fs/src/file/create.rs new file mode 100644 index 000000000..402741856 --- /dev/null +++ b/tokio-fs/src/file/create.rs @@ -0,0 +1,37 @@ +use super::File; + +use futures::{Future, Poll}; + +use std::fs::File as StdFile; +use std::io; +use std::path::Path; + +/// Future returned by `File::create` and resolves to a `File` instance. +#[derive(Debug)] +pub struct CreateFuture

{ + path: P, +} + +impl

CreateFuture

+where P: AsRef + Send + 'static, +{ + pub(crate) fn new(path: P) -> Self { + CreateFuture { path } + } +} + +impl

Future for CreateFuture

+where P: AsRef + Send + 'static, +{ + type Item = File; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + let std = try_ready!(::blocking_io(|| { + StdFile::create(&self.path) + })); + + let file = File::from_std(std); + Ok(file.into()) + } +} diff --git a/tokio-fs/src/file/mod.rs b/tokio-fs/src/file/mod.rs new file mode 100644 index 000000000..3b3a08b64 --- /dev/null +++ b/tokio-fs/src/file/mod.rs @@ -0,0 +1,201 @@ +//! Types for working with [`File`]. +//! +//! [`File`]: struct.File.html + +mod create; +mod open; + +pub use self::create::CreateFuture; +pub use self::open::OpenFuture; + +use tokio_io::{AsyncRead, AsyncWrite}; + +use futures::Poll; + +use std::fs::{File as StdFile, Metadata, Permissions}; +use std::io::{self, Read, Write, Seek}; +use std::path::Path; + +/// A reference to an open file on the filesystem. +/// +/// This is a specialized version of [`std::fs::File`][std] for usage from the +/// Tokio runtime. +/// +/// An instance of a `File` can be read and/or written depending on what options +/// it was opened with. Files also implement Seek to alter the logical cursor +/// that the file contains internally. +/// +/// Files are automatically closed when they go out of scope. +/// +/// [std]: https://doc.rust-lang.org/std/fs/struct.File.html +#[derive(Debug)] +pub struct File { + std: Option, +} + +impl File { + /// Attempts to open a file in read-only mode. + /// + /// # Errors + /// + /// `OpenFuture` results in an error if called from outside of the Tokio + /// runtime or if the underlying [`open`] call results in an error. + /// + /// [`open`]: https://doc.rust-lang.org/std/fs/struct.OpenOptions.html#method.open + pub fn open

(path: P) -> OpenFuture

+ where P: AsRef + Send + 'static, + { + OpenFuture::new(path) + } + + /// Opens a file in write-only mode. + /// + /// This function will create a file if it does not exist, and will truncate + /// it if it does. + /// + /// `CreateFuture` results in an error if called from outside of the Tokio + /// runtime or if the underlying [`create`] call results in an error. + /// + /// [`open`]: https://doc.rust-lang.org/std/fs/struct.File.html#method.create + pub fn create

(path: P) -> CreateFuture

+ where P: AsRef + Send + 'static, + { + CreateFuture::new(path) + } + + /// Convert a [`std::fs::File`][std] to a `tokio_fs::File`. + /// + /// [std]: https://doc.rust-lang.org/std/fs/struct.File.html + pub(crate) fn from_std(std: StdFile) -> File { + File { std: Some(std) } + } + + /// Seek to an offset, in bytes, in a stream. + /// + /// A seek beyond the end of a stream is allowed, but implementation + /// defined. + /// + /// If the seek operation completed successfully, this method returns the + /// new position from the start of the stream. That position can be used + /// later with `SeekFrom::Start`. + /// + /// # Errors + /// + /// Seeking to a negative offset is considered an error. + pub fn poll_seek(&mut self, pos: io::SeekFrom) -> Poll { + ::blocking_io(|| self.std().seek(pos)) + } + + /// Attempts to sync all OS-internal metadata to disk. + /// + /// This function will attempt to ensure that all in-core data reaches the + /// filesystem before returning. + pub fn poll_sync_all(&mut self) -> Poll<(), io::Error> { + ::blocking_io(|| self.std().sync_all()) + } + + /// This function is similar to `poll_sync_all`, except that it may not + /// synchronize file metadata to the filesystem. + /// + /// This is intended for use cases that must synchronize content, but don't + /// need the metadata on disk. The goal of this method is to reduce disk + /// operations. + /// + /// Note that some platforms may simply implement this in terms of `poll_sync_all`. + pub fn poll_sync_data(&mut self) -> Poll<(), io::Error> { + ::blocking_io(|| self.std().sync_data()) + } + + /// Truncates or extends the underlying file, updating the size of this file to become size. + /// + /// If the size is less than the current file's size, then the file will be + /// shrunk. If it is greater than the current file's size, then the file + /// will be extended to size and have all of the intermediate data filled in + /// with 0s. + /// + /// # Errors + /// + /// This function will return an error if the file is not opened for + /// writing. + pub fn poll_set_len(&mut self, size: u64) -> Poll<(), io::Error> { + ::blocking_io(|| self.std().set_len(size)) + } + + /// Queries metadata about the underlying file. + pub fn poll_metadata(&mut self) -> Poll { + ::blocking_io(|| self.std().metadata()) + } + + /// Create a new `File` instance that shares the same underlying file handle + /// as the existing `File` instance. Reads, writes, and seeks will affect both + /// File instances simultaneously. + pub fn poll_try_clone(&mut self) -> Poll { + ::blocking_io(|| { + let std = self.std().try_clone()?; + Ok(File::from_std(std)) + }) + } + + /// Changes the permissions on the underlying file. + /// + /// # Platform-specific behavior + /// + /// This function currently corresponds to the `fchmod` function on Unix and + /// the `SetFileInformationByHandle` function on Windows. Note that, this + /// [may change in the future][changes]. + /// + /// [changes]: https://doc.rust-lang.org/std/io/index.html#platform-specific-behavior + /// + /// # Errors + /// + /// This function will return an error if the user lacks permission change + /// attributes on the underlying file. It may also return an error in other + /// os-specific unspecified cases. + pub fn poll_set_permissions(&mut self, perm: Permissions) -> Poll<(), io::Error> { + ::blocking_io(|| self.std().set_permissions(perm)) + } + + fn std(&mut self) -> &mut StdFile { + self.std.as_mut().expect("`File` instance already shutdown") + } +} + +impl Read for File { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + ::would_block(|| self.std().read(buf)) + } +} + +impl AsyncRead for File { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } +} + +impl Write for File { + fn write(&mut self, buf: &[u8]) -> io::Result { + ::would_block(|| self.std().write(buf)) + } + + fn flush(&mut self) -> io::Result<()> { + ::would_block(|| self.std().flush()) + } +} + +impl AsyncWrite for File { + fn shutdown(&mut self) -> Poll<(), io::Error> { + ::blocking_io(|| { + self.std = None; + Ok(()) + }) + } +} + +impl Drop for File { + fn drop(&mut self) { + if let Some(_std) = self.std.take() { + // This is probably fine as closing a file *shouldn't* be a blocking + // operation. That said, ideally `shutdown` is called first. + } + } +} diff --git a/tokio-fs/src/file/open.rs b/tokio-fs/src/file/open.rs new file mode 100644 index 000000000..dc842bd80 --- /dev/null +++ b/tokio-fs/src/file/open.rs @@ -0,0 +1,37 @@ +use super::File; + +use futures::{Future, Poll}; + +use std::fs::File as StdFile; +use std::io; +use std::path::Path; + +/// Future returned by `File::open` and resolves to a `File` instance. +#[derive(Debug)] +pub struct OpenFuture

{ + path: P, +} + +impl

OpenFuture

+where P: AsRef + Send + 'static, +{ + pub(crate) fn new(path: P) -> Self { + OpenFuture { path } + } +} + +impl

Future for OpenFuture

+where P: AsRef + Send + 'static, +{ + type Item = File; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + let std = try_ready!(::blocking_io(|| { + StdFile::open(&self.path) + })); + + let file = File::from_std(std); + Ok(file.into()) + } +} diff --git a/tokio-fs/src/lib.rs b/tokio-fs/src/lib.rs new file mode 100644 index 000000000..424886cc6 --- /dev/null +++ b/tokio-fs/src/lib.rs @@ -0,0 +1,64 @@ +//! Asynchronous filesystem manipulation operations (and stdin, stdout, stderr). +//! +//! This module contains basic methods and types for manipulating the contents +//! of the local filesystem from within the context of the Tokio runtime. +//! +//! Tasks running on the Tokio runtime are expected to be asynchronous, i.e., +//! they will not block the thread of execution. Filesystem operations do not +//! satisfy this requirement. In order to perform filesystem operations +//! asynchronously, this library uses the [`blocking`][blocking] annotation +//! to signal to the runtime that a blocking operation is being performed. This +//! allows the runtime to compensate. +//! +//! [blocking]: https://docs.rs/tokio-threadpool/0.1/tokio_threadpool/fn.blocking.html + +#[macro_use] +extern crate futures; +extern crate tokio_io; +extern crate tokio_threadpool; + +pub mod file; +mod stdin; +mod stdout; +mod stderr; + +pub use file::File; +pub use stdin::{stdin, Stdin}; +pub use stdout::{stdout, Stdout}; +pub use stderr::{stderr, Stderr}; + +use futures::Poll; +use futures::Async::*; + +use std::io; +use std::io::ErrorKind::{Other, WouldBlock}; + +fn blocking_io(f: F) -> Poll +where F: FnOnce() -> io::Result, +{ + match tokio_threadpool::blocking(f) { + Ok(Ready(Ok(v))) => Ok(v.into()), + Ok(Ready(Err(err))) => Err(err), + Ok(NotReady) => Ok(NotReady), + Err(_) => Err(blocking_err()), + } +} + +fn would_block(f: F) -> io::Result +where F: FnOnce() -> io::Result, +{ + match tokio_threadpool::blocking(f) { + Ok(Ready(Ok(v))) => Ok(v), + Ok(Ready(Err(err))) => { + debug_assert_ne!(err.kind(), WouldBlock); + Err(err) + } + Ok(NotReady) => Err(WouldBlock.into()), + Err(_) => Err(blocking_err()), + } +} + +fn blocking_err() -> io::Error { + io::Error::new(Other, "tokio-fs::File::open must be called \ + from the context of the Tokio runtime.") +} diff --git a/tokio-fs/src/stderr.rs b/tokio-fs/src/stderr.rs new file mode 100644 index 000000000..cf6439075 --- /dev/null +++ b/tokio-fs/src/stderr.rs @@ -0,0 +1,45 @@ +use tokio_io::{AsyncWrite}; + +use futures::Poll; + +use std::io::{self, Write, Stderr as StdStderr}; + +/// A handle to the standard error stream of a process. +/// +/// The handle implements the [`AsyncWrite`] trait, but beware that concurrent +/// writes to `Stderr` must be executed with care. +/// +/// Created by the [`stderr`] function. +/// +/// [`stderr`]: fn.stderr.html +/// [`AsyncWrite`]: trait.AsyncWrite.html +#[derive(Debug)] +pub struct Stderr { + std: StdStderr, +} + +/// Constructs a new handle to the standard error of the current process. +/// +/// The returned handle allows writing to standard error from the within the +/// Tokio runtime. +pub fn stderr() -> Stderr { + let std = io::stderr(); + Stderr { std } +} + +impl Write for Stderr { + fn write(&mut self, buf: &[u8]) -> io::Result { + ::would_block(|| self.std.write(buf)) + } + + fn flush(&mut self) -> io::Result<()> { + ::would_block(|| self.std.flush()) + } +} + +impl AsyncWrite for Stderr { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } +} + diff --git a/tokio-fs/src/stdin.rs b/tokio-fs/src/stdin.rs new file mode 100644 index 000000000..f41b8dc84 --- /dev/null +++ b/tokio-fs/src/stdin.rs @@ -0,0 +1,38 @@ +use tokio_io::{AsyncRead}; + +use std::io::{self, Read, Stdin as StdStdin}; + +/// A handle to the standard input stream of a process. +/// +/// The handle implements the [`AsyncRead`] trait, but beware that concurrent +/// reads of `Stdin` must be executed with care. +/// +/// Created by the [`stdin`] function. +/// +/// [`stdin`]: fn.stdin.html +/// [`AsyncRead`]: trait.AsyncRead.html +#[derive(Debug)] +pub struct Stdin { + std: StdStdin, +} + +/// Constructs a new handle to the standard input of the current process. +/// +/// The returned handle allows reading from standard input from the within the +/// Tokio runtime. +pub fn stdin() -> Stdin { + let std = io::stdin(); + Stdin { std } +} + +impl Read for Stdin { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + ::would_block(|| self.std.read(buf)) + } +} + +impl AsyncRead for Stdin { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } +} diff --git a/tokio-fs/src/stdout.rs b/tokio-fs/src/stdout.rs new file mode 100644 index 000000000..1c4cd5ad0 --- /dev/null +++ b/tokio-fs/src/stdout.rs @@ -0,0 +1,44 @@ +use tokio_io::{AsyncWrite}; + +use futures::Poll; + +use std::io::{self, Write, Stdout as StdStdout}; + +/// A handle to the standard output stream of a process. +/// +/// The handle implements the [`AsyncWrite`] trait, but beware that concurrent +/// writes to `Stdout` must be executed with care. +/// +/// Created by the [`stdout`] function. +/// +/// [`stdout`]: fn.stdout.html +/// [`AsyncWrite`]: trait.AsyncWrite.html +#[derive(Debug)] +pub struct Stdout { + std: StdStdout, +} + +/// Constructs a new handle to the standard output of the current process. +/// +/// The returned handle allows writing to standard out from the within the Tokio +/// runtime. +pub fn stdout() -> Stdout { + let std = io::stdout(); + Stdout { std } +} + +impl Write for Stdout { + fn write(&mut self, buf: &[u8]) -> io::Result { + ::would_block(|| self.std.write(buf)) + } + + fn flush(&mut self) -> io::Result<()> { + ::would_block(|| self.std.flush()) + } +} + +impl AsyncWrite for Stdout { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } +} diff --git a/tokio-fs/tests/file.rs b/tokio-fs/tests/file.rs new file mode 100644 index 000000000..5f97deb9a --- /dev/null +++ b/tokio-fs/tests/file.rs @@ -0,0 +1,73 @@ +extern crate futures; +extern crate rand; +extern crate tempdir; +extern crate tokio_fs; +extern crate tokio_io; +extern crate tokio_threadpool; + +use tokio_fs::*; +use tokio_io::io; +use tokio_threadpool::*; + +use futures::Future; +use futures::future::poll_fn; +use futures::sync::oneshot; +use rand::{thread_rng, Rng}; +use tempdir::TempDir; + +use std::fs::File as StdFile; +use std::io::Read; + +#[test] +fn read_write() { + const NUM_CHARS: usize = 16 * 1_024; + + let dir = TempDir::new("tokio-fs-tests").unwrap(); + let file_path = dir.path().join("read_write.txt"); + + let contents: Vec = thread_rng().gen_ascii_chars() + .take(NUM_CHARS) + .collect::() + .into(); + + let pool = Builder::new() + .pool_size(1) + .build(); + + let (tx, rx) = oneshot::channel(); + + pool.spawn({ + let file_path = file_path.clone(); + let contents = contents.clone(); + + File::create(file_path) + .and_then(move |file| io::write_all(file, contents)) + .and_then(|(mut file, _)| { + poll_fn(move || file.poll_sync_all()) + }) + .then(|res| { + let _ = res.unwrap(); + tx.send(()).unwrap(); + Ok(()) + }) + }); + + rx.wait().unwrap(); + + let mut file = StdFile::open(&file_path).unwrap(); + + let mut dst = vec![]; + file.read_to_end(&mut dst).unwrap(); + + assert_eq!(dst, contents); + + pool.spawn({ + File::open(file_path) + .and_then(|file| io::read_to_end(file, vec![])) + .then(move |res| { + let (_, buf) = res.unwrap(); + assert_eq!(buf, contents); + Ok(()) + }) + }); +} diff --git a/tokio-threadpool/src/blocking.rs b/tokio-threadpool/src/blocking.rs index 212dbe01c..c78d25d98 100644 --- a/tokio-threadpool/src/blocking.rs +++ b/tokio-threadpool/src/blocking.rs @@ -2,6 +2,9 @@ use worker::Worker; use futures::Poll; +use std::error::Error; +use std::fmt; + /// Error raised by `blocking`. #[derive(Debug)] pub struct BlockingError { @@ -124,7 +127,7 @@ where F: FnOnce() -> T, }; // Transition the worker state to blocking. This will exit the fn early - // with `NotRead` if the pool does not have enough capacity to enter + // with `NotReady` if the pool does not have enough capacity to enter // blocking mode. worker.transition_to_blocking() }); @@ -146,3 +149,15 @@ where F: FnOnce() -> T, // Return the result Ok(ret.into()) } + +impl fmt::Display for BlockingError { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + write!(fmt, "{}", self.description()) + } +} + +impl Error for BlockingError { + fn description(&self) -> &str { + "`blocking` annotation used from outside the context of a thread pool" + } +}