Lift the error type out of Poll into a Result

That is:

    type Poll<T, E> = Result<Async<T>, E>

    enum Async<T> {
        Ready(T),
        NotReady,
    }

This commit tweaks the return type of the `Future::poll` method to be more
`try!`-friendly and also more amenable to `if let` for matching against
readiness.

Additionally, `Async` is used instead of `Option` to be more clear about what's
actually happening, and it's also theoretically more extensible in the future if
need be.

Finally, the `try_poll!` macro has been removed while a new `try_ready!` macro
has been added. This new macro has *different semantics* as it propagates both
`NotReady` *and* errors to the caller. It only returns the `Async::Ready`
variant.

Closes #108

Author: alexcrichton

futures-cpupool/src/lib.rs

@@ -48,7 +48,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::thread;
 
 use crossbeam::sync::MsQueue;
-use futures::{IntoFuture, Future, oneshot, Oneshot, Complete, Poll};
+use futures::{IntoFuture, Future, oneshot, Oneshot, Complete, Poll, Async};
 use futures::task::{self, Run, Executor};
 
 /// A thread pool intended to run CPU intensive work.
@@ -223,11 +223,11 @@ impl<T: Send + 'static, E: Send + 'static> Future for CpuFuture<T, E> {
     type Error = E;
 
     fn poll(&mut self) -> Poll<T, E> {
-        match self.inner.poll() {
-            Poll::Ok(Ok(res)) => res.into(),
-            Poll::Ok(Err(e)) => panic::resume_unwind(e),
-            Poll::Err(_) => panic!("shouldn't be canceled"),
-            Poll::NotReady => Poll::NotReady,
+        match self.inner.poll().expect("shouldn't be canceled") {
+            Async::Ready(Ok(Ok(e))) => Ok(e.into()),
+            Async::Ready(Ok(Err(e))) => Err(e),
+            Async::Ready(Err(e)) => panic::resume_unwind(e),
+            Async::NotReady => Ok(Async::NotReady),
         }
     }
 }
@@ -237,13 +237,17 @@ impl<F: Future> Future for Sender<F, Result<F::Item, F::Error>> {
     type Error = ();
 
     fn poll(&mut self) -> Poll<(), ()> {
-        if let Poll::Ok(_) = self.tx.as_mut().unwrap().poll_cancel() {
+        if let Ok(Async::Ready(_)) = self.tx.as_mut().unwrap().poll_cancel() {
             // Cancelled, bail out
-            return Poll::Ok(());
+            return Ok(().into())
         }
 
-        let res = try_poll!(self.fut.poll());
+        let res = match self.fut.poll() {
+            Ok(Async::Ready(e)) => Ok(e),
+            Ok(Async::NotReady) => return Ok(Async::NotReady),
+            Err(e) => Err(e),
+        };
         self.tx.take().unwrap().complete(res);
-        Poll::Ok(())
+        Ok(Async::Ready(()))
     }
 }

src/catch_unwind.rs

@@ -1,9 +1,9 @@
-use {Future, Poll};
-
 use std::prelude::v1::*;
 use std::any::Any;
 use std::panic::{catch_unwind, UnwindSafe, AssertUnwindSafe};
 
+use {Future, Poll, Async};
+
 /// Future for the `catch_unwind` combinator.
 ///
 /// This is created by this `Future::catch_unwind` method.
@@ -27,14 +27,14 @@ impl<F> Future for CatchUnwind<F>
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         let mut future = self.future.take().expect("cannot poll twice");
-        match catch_unwind(|| (future.poll(), future)) {
-            Ok((Poll::NotReady, f)) => {
-                self.future = Some(f);
-                Poll::NotReady
+        let (res, future) = try!(catch_unwind(|| (future.poll(), future)));
+        match res {
+            Ok(Async::NotReady) => {
+                self.future = Some(future);
+                Ok(Async::NotReady)
             }
-            Ok((Poll::Ok(v), _)) => Poll::Ok(Ok(v)),
-            Ok((Poll::Err(e), _)) => Poll::Ok(Err(e)),
-            Err(e) => Poll::Err(e),
+            Ok(Async::Ready(t)) => Ok(Async::Ready(Ok(t))),
+            Err(e) => Ok(Async::Ready(Err(e))),
         }
     }
 }

src/chain.rs

@@ -1,6 +1,6 @@
 use core::mem;
 
-use {Future, Poll};
+use {Future, Poll, Async};
 
 pub enum Chain<A, B, C> where A: Future {
     First(A, C),
@@ -21,22 +21,27 @@ impl<A, B, C> Chain<A, B, C>
                         -> Result<Result<B::Item, B>, B::Error>,
     {
         let a_result = match *self {
-            Chain::First(ref mut a, _) => try_poll!(a.poll()),
+            Chain::First(ref mut a, _) => {
+                match a.poll() {
+                    Ok(Async::NotReady) => return Ok(Async::NotReady),
+                    Ok(Async::Ready(t)) => Ok(t),
+                    Err(e) => Err(e),
+                }
+            }
             Chain::Second(ref mut b) => return b.poll(),
             Chain::Done => panic!("cannot poll a chained future twice"),
         };
         let data = match mem::replace(self, Chain::Done) {
             Chain::First(_, c) => c,
             _ => panic!(),
         };
-        match f(a_result, data) {
-            Ok(Ok(e)) => Poll::Ok(e),
-            Ok(Err(mut b)) => {
+        match try!(f(a_result, data)) {
+            Ok(e) => Ok(Async::Ready(e)),
+            Err(mut b) => {
                 let ret = b.poll();
                 *self = Chain::Second(b);
                 ret
             }
-            Err(e) => Poll::Err(e),
         }
     }
 }

src/collect.rs

@@ -2,7 +2,7 @@ use std::prelude::v1::*;
 
 use std::mem;
 
-use {Future, IntoFuture, Poll};
+use {Future, IntoFuture, Poll, Async};
 
 /// A future which takes a list of futures and resolves with a vector of the
 /// completed values.
@@ -77,8 +77,9 @@ impl<I> Future for Collect<I>
         loop {
             match self.cur {
                 Some(ref mut cur) => {
-                    match try_poll!(cur.poll()) {
-                        Ok(e) => self.result.push(e),
+                    match cur.poll() {
+                        Ok(Async::Ready(e)) => self.result.push(e),
+                        Ok(Async::NotReady) => return Ok(Async::NotReady),
 
                         // If we hit an error, drop all our associated resources
                         // ASAP.
@@ -89,12 +90,12 @@ impl<I> Future for Collect<I>
                             for f in self.result.drain(..) {
                                 drop(f);
                             }
-                            return Poll::Err(e)
+                            return Err(e)
                         }
                     }
                 }
                 None => {
-                    return Poll::Ok(mem::replace(&mut self.result, Vec::new()))
+                    return Ok(Async::Ready(mem::replace(&mut self.result, Vec::new())))
                 }
             }
 

src/done.rs

@@ -1,4 +1,4 @@
-use {Future, Poll};
+use {Future, Poll, Async};
 
 /// A future representing a value that is immediately ready.
 ///
@@ -30,6 +30,6 @@ impl<T, E> Future for Done<T, E> {
     type Error = E;
 
     fn poll(&mut self) -> Poll<T, E> {
-        self.inner.take().expect("cannot poll Done twice").into()
+        self.inner.take().expect("cannot poll Done twice").map(Async::Ready)
     }
 }

src/empty.rs

@@ -1,6 +1,6 @@
 use core::marker;
 
-use {Future, Poll};
+use {Future, Poll, Async};
 
 /// A future which is never resolved.
 ///
@@ -12,7 +12,7 @@ pub struct Empty<T, E> {
 /// Creates a future which never resolves, representing a computation that never
 /// finishes.
 ///
-/// The returned future will forever return `Poll::NotReady`.
+/// The returned future will forever return `Async::NotReady`.
 pub fn empty<T, E>() -> Empty<T, E> {
     Empty { _data: marker::PhantomData }
 }
@@ -22,6 +22,6 @@ impl<T, E> Future for Empty<T, E> {
     type Error = E;
 
     fn poll(&mut self) -> Poll<T, E> {
-        Poll::NotReady
+        Ok(Async::NotReady)
     }
 }

src/failed.rs

@@ -31,6 +31,6 @@ impl<T, E> Future for Failed<T, E> {
     type Error = E;
 
     fn poll(&mut self) -> Poll<T, E> {
-        Poll::Err(self.e.take().expect("cannot poll Failed twice"))
+        Err(self.e.take().expect("cannot poll Failed twice"))
     }
 }

src/finished.rs

@@ -1,6 +1,6 @@
 use core::marker;
 
-use {Future, Poll};
+use {Future, Poll, Async};
 
 /// A future representing a finished successful computation.
 ///
@@ -33,6 +33,6 @@ impl<T, E> Future for Finished<T, E> {
 
 
     fn poll(&mut self) -> Poll<T, E> {
-        Poll::Ok(self.t.take().expect("cannot poll Finished twice"))
+        Ok(Async::Ready(self.t.take().expect("cannot poll Finished twice")))
     }
 }

src/fuse.rs

@@ -1,9 +1,9 @@
-use {Future, Poll};
+use {Future, Poll, Async};
 
 /// A future which "fuse"s a future once it's been resolved.
 ///
 /// Normally futures can behave unpredictable once they're used after a future
-/// has been resolved, but `Fuse` is always defined to return `Poll::NotReady`
+/// has been resolved, but `Fuse` is always defined to return `Async::NotReady`
 /// from `poll` after it has succeeded, and after it has succeeded all future
 /// calls to `schedule` will be ignored.
 pub struct Fuse<A: Future> {
@@ -21,10 +21,14 @@ impl<A: Future> Future for Fuse<A> {
     type Error = A::Error;
 
     fn poll(&mut self) -> Poll<A::Item, A::Error> {
-        let ret = self.future.as_mut().map(|f| f.poll());
-        if ret.as_ref().map(|r| r.is_ready()) == Some(true) {
-            self.future = None;
+        let res = self.future.as_mut().map(|f| f.poll());
+        match res.unwrap_or(Ok(Async::NotReady)) {
+            res @ Ok(Async::Ready(_)) |
+            res @ Err(_) => {
+                self.future = None;
+                return res
+            }
+            Ok(Async::NotReady) => Ok(Async::NotReady)
         }
-        ret.unwrap_or(Poll::NotReady)
     }
 }

src/join.rs

@@ -2,7 +2,7 @@
 
 use core::mem;
 
-use {Future, Poll, IntoFuture};
+use {Future, Poll, IntoFuture, Async};
 
 macro_rules! generate {
     ($(
@@ -50,23 +50,23 @@ macro_rules! generate {
                     Ok(done) => done,
                     Err(e) => {
                         self.erase();
-                        return Poll::Err(e)
+                        return Err(e)
                     }
                 };
                 $(
                     all_done = match self.$B.poll() {
                         Ok(done) => all_done && done,
                         Err(e) => {
                             self.erase();
-                            return Poll::Err(e)
+                            return Err(e)
                         }
                     };
                 )*
 
                 if all_done {
-                    Poll::Ok((self.a.take(), $(self.$B.take()),*))
+                    Ok(Async::Ready((self.a.take(), $(self.$B.take()),*)))
                 } else {
-                    Poll::NotReady
+                    Ok(Async::NotReady)
                 }
             }
         }
@@ -131,17 +131,16 @@ enum MaybeDone<A: Future> {
 impl<A: Future> MaybeDone<A> {
     fn poll(&mut self) -> Result<bool, A::Error> {
         let res = match *self {
-            MaybeDone::NotYet(ref mut a) => a.poll(),
+            MaybeDone::NotYet(ref mut a) => try!(a.poll()),
             MaybeDone::Done(_) => return Ok(true),
             MaybeDone::Gone => panic!("cannot poll Join twice"),
         };
         match res {
-            Poll::Ok(res) => {
+            Async::Ready(res) => {
                 *self = MaybeDone::Done(res);
                 Ok(true)
             }
-            Poll::Err(res) => Err(res),
-            Poll::NotReady => Ok(false),
+            Async::NotReady => Ok(false),
         }
     }