wip

Cargo.toml

@@ -6,7 +6,7 @@ members = [
   "tokio-codec",
   "tokio-current-thread",
   "tokio-executor",
-  # "tokio-fs",
+  "tokio-fs",
   "tokio-futures",
   "tokio-io",
   "tokio-macros",

tokio-fs/Cargo.toml

@@ -24,9 +24,9 @@ categories = ["asynchronous", "network-programming", "filesystem"]
 publish = false
 
 [dependencies]
-futures = "0.1.21"
 tokio-threadpool = { version = "0.2.0", path = "../tokio-threadpool" }
 tokio-io = { version = "0.2.0", path = "../tokio-io" }
+tokio-futures = { version = "0.2.0", path = "../tokio-futures" }
 
 [dev-dependencies]
 rand = "0.6"

tokio-fs/examples/std-echo.rs

@@ -5,7 +5,7 @@ use tokio_codec::{FramedRead, FramedWrite, LinesCodec};
 use tokio_fs::{stderr, stdin, stdout};
 use tokio_threadpool::Builder;
 
-use futures::{Future, Sink, Stream};
+use std::future::Future;
 
 use std::io;
 

tokio-fs/src/create_dir.rs

@@ -1,7 +1,10 @@
-use futures::{Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::fs;
 use std::io;
 use std::path::Path;
+use std::pin::Pin;
 
 /// Creates a new, empty directory at the provided path
 ///
@@ -34,10 +37,9 @@ impl<P> Future for CreateDirFuture<P>
 where
     P: AsRef<Path>,
 {
-    type Item = ();
-    type Error = io::Error;
+    type Output = io::Result<()>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
         crate::blocking_io(|| fs::create_dir(&self.path))
     }
 }

tokio-fs/src/create_dir_all.rs

@@ -1,7 +1,10 @@
-use futures::{Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::fs;
 use std::io;
 use std::path::Path;
+use std::pin::Pin;
 
 /// Recursively create a directory and all of its parent components if they
 /// are missing.
@@ -35,10 +38,9 @@ impl<P> Future for CreateDirAllFuture<P>
 where
     P: AsRef<Path>,
 {
-    type Item = ();
-    type Error = io::Error;
+    type Output = io::Result<()>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
         crate::blocking_io(|| fs::create_dir_all(&self.path))
     }
 }

tokio-fs/src/file/clone.rs

@@ -1,6 +1,9 @@
 use super::File;
-use futures::{Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::io;
+use std::pin::Pin;
 
 /// Future returned by `File::try_clone`.
 ///
@@ -21,15 +24,15 @@ impl CloneFuture {
 }
 
 impl Future for CloneFuture {
-    type Item = (File, File);
-    type Error = (File, io::Error);
+    type Output = Result<(File, File), (File, io::Error)>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        self.file
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let inner_self = Pin::get_mut(self);
+        inner_self.file
             .as_mut()
             .expect("Cannot poll `CloneFuture` after it resolves")
             .poll_try_clone()
-            .map(|inner| inner.map(|cloned| (self.file.take().unwrap(), cloned)))
-            .map_err(|err| (self.file.take().unwrap(), err))
+            .map(|inner| inner.map(|cloned| (inner_self.file.take().unwrap(), cloned)))
+            .map_err(|err| (inner_self.file.take().unwrap(), err))
     }
 }

tokio-fs/src/file/create.rs

@@ -1,8 +1,11 @@
 use super::File;
-use futures::{try_ready, Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::fs::File as StdFile;
 use std::io;
 use std::path::Path;
+use std::pin::Pin;
 
 /// Future returned by `File::create` and resolves to a `File` instance.
 #[derive(Debug)]
@@ -19,17 +22,18 @@ where
     }
 }
 
+impl<P> Unpin for CreateFuture<P> {} 
+
 impl<P> Future for CreateFuture<P>
 where
     P: AsRef<Path> + Send + 'static,
 {
-    type Item = File;
-    type Error = io::Error;
+    type Output = io::Result<File>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let std = try_ready!(crate::blocking_io(|| StdFile::create(&self.path)));
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let std = ready!(crate::blocking_io(|| StdFile::create(&self.path)))?;
 
         let file = File::from_std(std);
-        Ok(file.into())
+        Poll::Ready(Ok(file.into()))
     }
 }

tokio-fs/src/file/metadata.rs

@@ -1,8 +1,11 @@
 use super::File;
-use futures::{try_ready, Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::fs::File as StdFile;
 use std::fs::Metadata;
 use std::io;
+use std::pin::Pin;
 
 const POLL_AFTER_RESOLVE: &str = "Cannot poll MetadataFuture after it resolves";
 
@@ -23,13 +26,13 @@ impl MetadataFuture {
 }
 
 impl Future for MetadataFuture {
-    type Item = (File, Metadata);
-    type Error = io::Error;
+    type Output = io::Result<(File, Metadata)>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let metadata = try_ready!(crate::blocking_io(|| StdFile::metadata(self.std())));
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let inner = Pin::get_mut(self);
+        let metadata = ready!(crate::blocking_io(|| StdFile::metadata(inner.std())))?;
 
-        let file = self.file.take().expect(POLL_AFTER_RESOLVE);
-        Ok((file, metadata).into())
+        let file = inner.file.take().expect(POLL_AFTER_RESOLVE);
+        Poll::Ready(Ok((file, metadata).into()))
     }
 }

tokio-fs/src/file/mod.rs

@@ -16,10 +16,12 @@ pub use self::open::OpenFuture;
 pub use self::open_options::OpenOptions;
 pub use self::seek::SeekFuture;
 
-use futures::Poll;
+use std::task::Poll;
+use std::task::Context;
 use std::fs::{File as StdFile, Metadata, Permissions};
 use std::io::{self, Read, Seek, Write};
 use std::path::Path;
+use std::pin::Pin;
 use tokio_io::{AsyncRead, AsyncWrite};
 
 /// A reference to an open file on the filesystem.
@@ -191,7 +193,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_seek(&mut self, pos: io::SeekFrom) -> Poll<u64, io::Error> {
+    pub fn poll_seek(&mut self, pos: io::SeekFrom) -> Poll<io::Result<u64>> {
         crate::blocking_io(|| self.std().seek(pos))
     }
 
@@ -243,7 +245,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_sync_all(&mut self) -> Poll<(), io::Error> {
+    pub fn poll_sync_all(&mut self) -> Poll<io::Result<()>> {
         crate::blocking_io(|| self.std().sync_all())
     }
 
@@ -273,7 +275,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_sync_data(&mut self) -> Poll<(), io::Error> {
+    pub fn poll_sync_data(&mut self) -> Poll<io::Result<()>> {
         crate::blocking_io(|| self.std().sync_data())
     }
 
@@ -305,7 +307,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_set_len(&mut self, size: u64) -> Poll<(), io::Error> {
+    pub fn poll_set_len(&mut self, size: u64) -> Poll<io::Result<()>> {
         crate::blocking_io(|| self.std().set_len(size))
     }
 
@@ -344,7 +346,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_metadata(&mut self) -> Poll<Metadata, io::Error> {
+    pub fn poll_metadata(&mut self) -> Poll<io::Result<Metadata>> {
         crate::blocking_io(|| self.std().metadata())
     }
 
@@ -366,7 +368,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_try_clone(&mut self) -> Poll<File, io::Error> {
+    pub fn poll_try_clone(&mut self) -> Poll<io::Result<File>> {
         crate::blocking_io(|| {
             let std = self.std().try_clone()?;
             Ok(File::from_std(std))
@@ -437,7 +439,7 @@ impl File {
     ///
     /// tokio::run(task);
     /// ```
-    pub fn poll_set_permissions(&mut self, perm: Permissions) -> Poll<(), io::Error> {
+    pub fn poll_set_permissions(&mut self, perm: Permissions) -> Poll<io::Result<()>> {
         crate::blocking_io(|| self.std().set_permissions(perm))
     }
 
@@ -479,8 +481,15 @@ impl Read for File {
 }
 
 impl AsyncRead for File {
-    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
-        false
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8]
+    ) -> Poll<io::Result<usize>> {
+        match Pin::get_mut(self).read(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
+            other => Poll::Ready(other),
+        }
     }
 }
 
@@ -495,11 +504,26 @@ impl Write for File {
 }
 
 impl AsyncWrite for File {
-    fn shutdown(&mut self) -> Poll<(), io::Error> {
-        crate::blocking_io(|| {
-            self.std = None;
-            Ok(())
-        })
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &[u8]
+    ) -> Poll<io::Result<usize>> {
+        match Pin::get_mut(self).write(buf) {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
+            other => Poll::Ready(other),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        match Pin::get_mut(self).flush() {
+            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Poll::Pending,
+            other => Poll::Ready(other),
+        }
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
     }
 }
 

tokio-fs/src/file/open.rs

@@ -1,8 +1,11 @@
 use super::File;
-use futures::{try_ready, Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::fs::OpenOptions as StdOpenOptions;
 use std::io;
 use std::path::Path;
+use std::pin::Pin;
 
 /// Future returned by `File::open` and resolves to a `File` instance.
 #[derive(Debug)]
@@ -20,17 +23,18 @@ where
     }
 }
 
+impl<P> Unpin for OpenFuture<P> {} 
+
 impl<P> Future for OpenFuture<P>
 where
     P: AsRef<Path> + Send + 'static,
 {
-    type Item = File;
-    type Error = io::Error;
+    type Output = io::Result<File>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let std = try_ready!(crate::blocking_io(|| self.options.open(&self.path)));
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let std = ready!(crate::blocking_io(|| self.options.open(&self.path)))?;
 
         let file = File::from_std(std);
-        Ok(file.into())
+        Poll::Ready(Ok(file.into()))
     }
 }

tokio-fs/src/file/seek.rs

@@ -1,6 +1,9 @@
 use super::File;
-use futures::{try_ready, Future, Poll};
+use std::future::Future;
+use std::task::Poll;
+use std::task::Context;
 use std::io;
+use std::pin::Pin;
 
 /// Future returned by `File::seek`.
 #[derive(Debug)]
@@ -19,16 +22,16 @@ impl SeekFuture {
 }
 
 impl Future for SeekFuture {
-    type Item = (File, u64);
-    type Error = io::Error;
+    type Output = io::Result<(File, u64)>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let pos = try_ready!(self
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let inner_self = Pin::get_mut(self);
+        let pos = ready!(inner_self
             .inner
             .as_mut()
             .expect("Cannot poll `SeekFuture` after it resolves")
-            .poll_seek(self.pos));
-        let inner = self.inner.take().unwrap();
-        Ok((inner, pos).into())
+            .poll_seek(inner_self.pos))?;
+        let inner = inner_self.inner.take().unwrap();
+        Poll::Ready(Ok((inner, pos).into()))
     }
 }