Transition futures-channel to 0.3.0-alpha

Author: aturon

.travis.yml

@@ -47,8 +47,9 @@ matrix:
     #     - cargo test --manifest-path futures/testcrate/Cargo.toml
 
 script:
-  - cargo test --all
-  - cargo test --all --release
+  - cargo build --all # for now build only; tests require the full suite of crates
+#  - cargo test --all
+#  - cargo test --all --release
 
 env:
   global:

Cargo.toml

@@ -3,7 +3,7 @@ members = [
 #  "futures",
 #  "futures-async-runtime",
   "futures-core",
-#  "futures-channel",
+  "futures-channel",
 #  "futures-executor",
 #  "futures-io",
 #  "futures-macro-async",

futures-channel/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "futures-channel"
-version = "0.2.0"
+version = "0.3.0-alpha"
 authors = ["Alex Crichton <[email protected]>"]
 license = "MIT/Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
@@ -15,8 +15,8 @@ std = ["futures-core/std"]
 default = ["std"]
 
 [dependencies]
-futures-core = { path = "../futures-core", version = "0.2.0", default-features = false }
+futures-core = { path = "../futures-core", version = "0.3.0-alpha", default-features = false }
 
 [dev-dependencies]
-futures = { path = "../futures", version = "0.2.0", default-features = true }
-futures-executor = { path = "../futures-executor", version = "0.2.0", default-features = true }
+# futures = { path = "../futures", version = "0.2.0", default-features = true }
+# futures-executor = { path = "../futures-executor", version = "0.2.0", default-features = true }

futures-channel/src/lib.rs

@@ -3,6 +3,8 @@
 //! This crate provides channels that can be used to communicate between
 //! asynchronous tasks.
 
+#![feature(pin, arbitrary_self_types)]
+
 #![deny(missing_docs, missing_debug_implementations)]
 #![doc(html_root_url = "https://docs.rs/futures-channel/0.2.0")]
 #![no_std]

futures-channel/src/mpsc/mod.rs

@@ -74,6 +74,8 @@
 // happens-before semantics required for the acquire / release semantics used
 // by the queue structure.
 
+use std::mem::Pin;
+use std::marker::Unpin;
 use std::fmt;
 use std::error::Error;
 use std::any::Any;
@@ -84,8 +86,7 @@ use std::thread;
 use std::usize;
 
 use futures_core::task::{self, Waker};
-use futures_core::{Async, Poll, Stream};
-use futures_core::never::Never;
+use futures_core::{Poll, Stream};
 
 use mpsc::queue::{Queue, PopResult};
 
@@ -109,6 +110,9 @@ pub struct Sender<T> {
     maybe_parked: bool,
 }
 
+// Safe because we treat the `T` opaquely
+unsafe impl<T> Unpin for Sender<T> {}
+
 /// The transmission end of an unbounded mpsc channel.
 ///
 /// This value is created by the [`unbounded`](unbounded) function.
@@ -126,12 +130,18 @@ pub struct Receiver<T> {
     inner: Arc<Inner<T>>,
 }
 
+// Safe because we treat the `T` opaquely
+unsafe impl<T> Unpin for Receiver<T> {}
+
 /// The receiving end of an unbounded mpsc channel.
 ///
 /// This value is created by the [`unbounded`](unbounded) function.
 #[derive(Debug)]
 pub struct UnboundedReceiver<T>(Receiver<T>);
 
+// Safe because we treat the `T` opaquely
+unsafe impl<T> Unpin for UnboundedReceiver<T> {}
+
 /// The error type for [`Sender`s](Sender) used as `Sink`s.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct SendError {
@@ -511,14 +521,14 @@ impl<T> Sender<T> {
         Ok(())
     }
 
-    fn poll_ready_nb(&self) -> Poll<(), SendError> {
+    fn poll_ready_nb(&self) -> Poll<Result<(), SendError>> {
         let state = decode_state(self.inner.state.load(SeqCst));
         if state.is_open {
-            Ok(Async::Ready(()))
+            Poll::Ready(Ok(()))
         } else {
-            Err(SendError {
+            Poll::Ready(Err(SendError {
                 kind: SendErrorKind::Full,
-            })
+            }))
         }
     }
 
@@ -637,15 +647,15 @@ impl<T> Sender<T> {
     /// - `Ok(Async::Pending)` if the channel may not have
     /// capacity, in which case the current task is queued to be notified once capacity is available;
     /// - `Err(SendError)` if the receiver has been dropped.
-    pub fn poll_ready(&mut self, cx: &mut task::Context) -> Poll<(), SendError> {
+    pub fn poll_ready(&mut self, cx: &mut task::Context) -> Poll<Result<(), SendError>> {
         let state = decode_state(self.inner.state.load(SeqCst));
         if !state.is_open {
-            return Err(SendError {
+            return Poll::Ready(Err(SendError {
                 kind: SendErrorKind::Disconnected,
-            });
+            }));
         }
 
-        Ok(self.poll_unparked(Some(cx)))
+        self.poll_unparked(Some(cx)).map(Ok)
     }
 
     /// Returns whether this channel is closed without needing a context.
@@ -662,7 +672,7 @@ impl<T> Sender<T> {
         let _ = self.do_send_nb(None);
     }
 
-    fn poll_unparked(&mut self, cx: Option<&mut task::Context>) -> Async<()> {
+    fn poll_unparked(&mut self, cx: Option<&mut task::Context>) -> Poll<()> {
         // First check the `maybe_parked` variable. This avoids acquiring the
         // lock in most cases
         if self.maybe_parked {
@@ -671,7 +681,7 @@ impl<T> Sender<T> {
 
             if !task.is_parked {
                 self.maybe_parked = false;
-                return Async::Ready(())
+                return Poll::Ready(())
             }
 
             // At this point, an unpark request is pending, so there will be an
@@ -682,16 +692,16 @@ impl<T> Sender<T> {
             // task
             task.task = cx.map(|cx| cx.waker().clone());
 
-            Async::Pending
+            Poll::Pending
         } else {
-            Async::Ready(())
+            Poll::Ready(())
         }
     }
 }
 
 impl<T> UnboundedSender<T> {
     /// Check if the channel is ready to receive a message.
-    pub fn poll_ready(&self, _: &mut task::Context) -> Poll<(), SendError> {
+    pub fn poll_ready(&self, _: &mut task::Context) -> Poll<Result<(), SendError>> {
         self.0.poll_ready_nb()
     }
 
@@ -832,14 +842,14 @@ impl<T> Receiver<T> {
     /// no longer empty.
     pub fn try_next(&mut self) -> Result<Option<T>, TryRecvError> {
         match self.next_message() {
-            Async::Ready(msg) => {
+            Poll::Ready(msg) => {
                 Ok(msg)
             },
-            Async::Pending => Err(TryRecvError { _inner: () }),
+            Poll::Pending => Err(TryRecvError { _inner: () }),
         }
     }
 
-    fn next_message(&mut self) -> Async<Option<T>> {
+    fn next_message(&mut self) -> Poll<Option<T>> {
         // Pop off a message
         loop {
             match unsafe { self.inner.message_queue.pop() } {
@@ -851,11 +861,11 @@ impl<T> Receiver<T> {
                     // Decrement number of messages
                     self.dec_num_messages();
 
-                    return Async::Ready(msg);
+                    return Poll::Ready(msg);
                 }
                 PopResult::Empty => {
                     // The queue is empty, return Pending
-                    return Async::Pending;
+                    return Poll::Pending;
                 }
                 PopResult::Inconsistent => {
                     // Inconsistent means that there will be a message to pop
@@ -937,27 +947,26 @@ impl<T> Receiver<T> {
 
 impl<T> Stream for Receiver<T> {
     type Item = T;
-    type Error = Never;
 
-    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
+    fn poll_next(mut self: Pin<Self>, cx: &mut task::Context) -> Poll<Option<T>> {
         loop {
             // Try to read a message off of the message queue.
             let msg = match self.next_message() {
-                Async::Ready(msg) => msg,
-                Async::Pending => {
+                Poll::Ready(msg) => msg,
+                Poll::Pending => {
                     // There are no messages to read, in this case, attempt to
                     // park. The act of parking will verify that the channel is
                     // still empty after the park operation has completed.
                     match self.try_park(cx) {
                         TryPark::Parked => {
                             // The task was parked, and the channel is still
                             // empty, return Pending.
-                            return Ok(Async::Pending);
+                            return Poll::Pending;
                         }
                         TryPark::Closed => {
                             // The channel is closed, there will be no further
                             // messages.
-                            return Ok(Async::Ready(None));
+                            return Poll::Ready(None);
                         }
                         TryPark::NotEmpty => {
                             // A message has been sent while attempting to
@@ -969,7 +978,7 @@ impl<T> Stream for Receiver<T> {
                 }
             };
             // Return the message
-            return Ok(Async::Ready(msg));
+            return Poll::Ready(msg);
         }
     }
 }
@@ -1005,10 +1014,9 @@ impl<T> UnboundedReceiver<T> {
 
 impl<T> Stream for UnboundedReceiver<T> {
     type Item = T;
-    type Error = Never;
 
-    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
-        self.0.poll_next(cx)
+    fn poll_next(mut self: Pin<Self>, cx: &mut task::Context) -> Poll<Option<T>> {
+        Pin::new(&mut self.0).poll_next(cx)
     }
 }
 

futures-channel/src/oneshot.rs

@@ -1,14 +1,14 @@
 //! A channel for sending a single message between asynchronous tasks.
 
+use std::mem::Pin;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering::SeqCst;
 use std::error::Error;
 use std::fmt;
 
-use futures_core::{Future, Poll, Async};
+use futures_core::{Future, Poll};
 use futures_core::task::{self, Waker};
-use futures_core::never::Never;
 
 use lock::Lock;
 
@@ -157,13 +157,13 @@ impl<T> Inner<T> {
         }
     }
 
-    fn poll_cancel(&self, cx: &mut task::Context) -> Poll<(), Never> {
+    fn poll_cancel(&self, cx: &mut task::Context) -> Poll<()> {
         // Fast path up first, just read the flag and see if our other half is
         // gone. This flag is set both in our destructor and the oneshot
         // destructor, but our destructor hasn't run yet so if it's set then the
         // oneshot is gone.
         if self.complete.load(SeqCst) {
-            return Ok(Async::Ready(()))
+            return Poll::Ready(())
         }
 
         // If our other half is not gone then we need to park our current task
@@ -182,12 +182,12 @@ impl<T> Inner<T> {
         let handle = cx.waker().clone();
         match self.tx_task.try_lock() {
             Some(mut p) => *p = Some(handle),
-            None => return Ok(Async::Ready(())),
+            None => return Poll::Ready(()),
         }
         if self.complete.load(SeqCst) {
-            Ok(Async::Ready(()))
+            Poll::Ready(())
         } else {
-            Ok(Async::Pending)
+            Poll::Pending
         }
     }
 
@@ -252,7 +252,7 @@ impl<T> Inner<T> {
         }
     }
 
-    fn recv(&self, cx: &mut task::Context) -> Poll<T, Canceled> {
+    fn recv(&self, cx: &mut task::Context) -> Poll<Result<T, Canceled>> {
         let mut done = false;
 
         // Check to see if some data has arrived. If it hasn't then we need to
@@ -286,12 +286,12 @@ impl<T> Inner<T> {
             // treat the send as a failure.
             if let Some(mut slot) = self.data.try_lock() {
                 if let Some(data) = slot.take() {
-                    return Ok(data.into());
+                    return Poll::Ready(Ok(data));
                 }
             }
-            Err(Canceled)
+            Poll::Ready(Err(Canceled))
         } else {
-            Ok(Async::Pending)
+            Poll::Pending
         }
     }
 
@@ -353,7 +353,7 @@ impl<T> Sender<T> {
     /// alive and may be able to receive a message if sent. The current task,
     /// however, is scheduled to receive a notification if the corresponding
     /// `Receiver` goes away.
-    pub fn poll_cancel(&mut self, cx: &mut task::Context) -> Poll<(), Never> {
+    pub fn poll_cancel(&mut self, cx: &mut task::Context) -> Poll<()> {
         self.inner.poll_cancel(cx)
     }
 
@@ -420,10 +420,9 @@ impl<T> Receiver<T> {
 }
 
 impl<T> Future for Receiver<T> {
-    type Item = T;
-    type Error = Canceled;
+    type Output = Result<T, Canceled>;
 
-    fn poll(&mut self, cx: &mut task::Context) -> Poll<T, Canceled> {
+    fn poll(self: Pin<Self>, cx: &mut task::Context) -> Poll<Result<T, Canceled>> {
         self.inner.recv(cx)
     }
 }