Merge branch 'master' into consistent-async-args

Author: boats

futures-core/src/stream/mod.rs

@@ -71,6 +71,9 @@ impl<'a, S: ?Sized + Stream> Stream for &'a mut S {
 }
 
 if_std! {
+    use Async;
+    use never::Never;
+
     impl<S: ?Sized + Stream> Stream for ::std::boxed::Box<S> {
         type Item = S::Item;
         type Error = S::Error;
@@ -98,4 +101,13 @@ if_std! {
             self.0.poll_next(cx)
         }
     }
+
+    impl<T> Stream for ::std::collections::VecDeque<T> {
+        type Item = T;
+        type Error = Never;
+
+        fn poll_next(&mut self, _cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
+            Ok(Async::Ready(self.pop_front()))
+        }
+    }
 }

futures-sink/src/lib.rs

@@ -51,6 +51,28 @@ if_std! {
         }
     }
 
+    impl<T> Sink for ::std::collections::VecDeque<T> {
+        type SinkItem = T;
+        type SinkError = Never;
+
+        fn poll_ready(&mut self, _: &mut task::Context) -> Poll<(), Self::SinkError> {
+            Ok(Async::Ready(()))
+        }
+
+        fn start_send(&mut self, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+            self.push_back(item);
+            Ok(())
+        }
+
+        fn poll_flush(&mut self, _: &mut task::Context) -> Poll<(), Self::SinkError> {
+            Ok(Async::Ready(()))
+        }
+
+        fn poll_close(&mut self, _: &mut task::Context) -> Poll<(), Self::SinkError> {
+            Ok(Async::Ready(()))
+        }
+    }
+
     impl<S: ?Sized + Sink> Sink for ::std::boxed::Box<S> {
         type SinkItem = S::SinkItem;
         type SinkError = S::SinkError;
@@ -98,7 +120,7 @@ if_std! {
 /// required methods, and a host of default methods for working in a
 /// higher-level way. The `Sink::send_all` combinator is of particular
 /// importance: you can use it to send an entire stream to a sink, which is
-/// the simplest way to ultimately consume a sink.
+/// the simplest way to ultimately consume a stream.
 pub trait Sink {
     /// The type of value that the sink accepts.
     type SinkItem;
@@ -133,7 +155,7 @@ pub trait Sink {
     /// send**.
     ///
     /// Implementations of `poll_ready` and `start_send` will usually involve
-    /// flushing behind the scenes in order to make room for new messages. 
+    /// flushing behind the scenes in order to make room for new messages.
     /// It is only necessary to call `poll_flush` if you need to guarantee that
     /// *all* of the items placed into the `Sink` have been sent.
     ///

futures-util/src/future/flatten_sink.rs

@@ -0,0 +1,80 @@
+use core::fmt;
+
+use futures_core::{task, Async, Future};
+use futures_sink::Sink;
+
+#[derive(Debug)]
+enum State<F> where F: Future, <F as Future>::Item: Sink {
+    Waiting(F),
+    Ready(F::Item),
+    Closed,
+}
+
+/// Future for the `flatten_sink` combinator, flattening a
+/// future-of-a-sink to get just the result of the final sink as a sink.
+///
+/// This is created by the `Future::flatten_sink` method.
+pub struct FlattenSink<F> where F: Future, <F as Future>::Item: Sink {
+    st: State<F>
+}
+
+impl<F> fmt::Debug for FlattenSink<F>
+    where F: Future + fmt::Debug,
+          <F as Future>::Item: Sink<SinkError=F::Error> + fmt::Debug,
+{
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("FlattenStream")
+            .field("state", &self.st)
+            .finish()
+    }
+}
+
+impl<F> Sink for FlattenSink<F> where F: Future, <F as Future>::Item: Sink<SinkError=F::Error> {
+    type SinkItem = <<F as Future>::Item as Sink>::SinkItem;
+    type SinkError = <<F as Future>::Item as Sink>::SinkError;
+
+    fn poll_ready(&mut self, cx: &mut task::Context) -> Result<Async<()>, Self::SinkError> {
+        let mut resolved_stream = match self.st {
+            State::Ready(ref mut s) => return s.poll_ready(cx),
+            State::Waiting(ref mut f) => match f.poll(cx)? {
+                Async::Pending => return Ok(Async::Pending),
+                Async::Ready(s) => s,
+            },
+            State::Closed => panic!("poll_ready called after eof"),
+        };
+        let result = resolved_stream.poll_ready(cx);
+        self.st = State::Ready(resolved_stream);
+        result
+    }
+
+    fn start_send(&mut self, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+        match self.st {
+            State::Ready(ref mut s) => s.start_send(item),
+            State::Waiting(_) => panic!("poll_ready not called first"),
+            State::Closed => panic!("start_send called after eof"),
+        }
+    }
+
+    fn poll_flush(&mut self, cx: &mut task::Context) -> Result<Async<()>, Self::SinkError> {
+        match self.st {
+            State::Ready(ref mut s) => s.poll_flush(cx),
+            // if sink not yet resolved, nothing written ==> everything flushed
+            State::Waiting(_) => Ok(Async::Ready(())),
+            State::Closed => panic!("poll_flush called after eof"),
+        }
+    }
+
+    fn poll_close(&mut self, cx: &mut task::Context) -> Result<Async<()>, Self::SinkError> {
+        if let State::Ready(ref mut s) = self.st {
+            try_ready!(s.poll_close(cx));
+        }
+        self.st = State::Closed;
+        return Ok(Async::Ready(()));
+    }
+}
+
+pub fn new<F>(fut: F) -> FlattenSink<F> where F: Future, <F as Future>::Item: Sink {
+    FlattenSink {
+        st: State::Waiting(fut)
+    }
+}

futures-util/src/future/join_all.rs

@@ -143,22 +143,3 @@ impl<F: IntoFuture> FromIterator<F> for JoinAll<F::Future> {
         join_all(iter)
     }
 }
-
-#[test]
-fn join_all_from_iter() {
-    let f_ok = |a| {
-        let r: Result<i32, ()> = Ok(a);
-        r.into_future()
-    };
-
-    pub fn assert_done<T, F>(_f: F, _result: Result<T::Item, T::Error>)
-        where T: Future,
-              T::Item: Eq + fmt::Debug,
-              T::Error: Eq + fmt::Debug,
-              F: FnOnce() -> T,
-    {
-        //FIXME: write a real test somehow somewhere
-    }
-
-    assert_done(|| vec![f_ok(1), f_ok(2)].into_iter().collect::<JoinAll<_>>(), Ok(vec![1, 2]))
-}

futures-util/src/future/mod.rs

@@ -6,6 +6,7 @@
 use core::result;
 
 use futures_core::{Future, IntoFuture, Stream};
+use futures_sink::Sink;
 
 // Primitive futures
 mod empty;
@@ -20,6 +21,7 @@ pub use self::loop_fn::{loop_fn, Loop, LoopFn};
 // combinators
 mod and_then;
 mod flatten;
+mod flatten_sink;
 mod flatten_stream;
 mod fuse;
 mod into_stream;
@@ -39,6 +41,7 @@ mod chain;
 
 pub use self::and_then::AndThen;
 pub use self::flatten::Flatten;
+pub use self::flatten_sink::FlattenSink;
 pub use self::flatten_stream::FlattenStream;
 pub use self::fuse::Fuse;
 pub use self::into_stream::IntoStream;
@@ -632,6 +635,22 @@ pub trait FutureExt: Future {
                         _>(f)
     }
 
+    /// Flatten the execution of this future when the successful result of this
+    /// future is a sink.
+    ///
+    /// This can be useful when sink initialization is deferred, and it is
+    /// convenient to work with that sink as if sink was available at the
+    /// call site.
+    ///
+    /// Note that this function consumes this future and returns a wrapped
+    /// version of it.
+    fn flatten_sink(self) -> FlattenSink<Self>
+        where <Self as Future>::Item: Sink<SinkError=Self::Error>,
+              Self: Sized
+    {
+        flatten_sink::new(self)
+    }
+
     /// Flatten the execution of this future when the successful result of this
     /// future is a stream.
     ///

futures-util/src/future/shared.rs

@@ -234,8 +234,17 @@ impl Wake for Notifier {
     }
 }
 
-unsafe impl<F: Future> Sync for Inner<F> {}
-unsafe impl<F: Future> Send for Inner<F> {}
+unsafe impl<F> Sync for Inner<F> 
+    where F: Future + Send,
+          F::Item: Send + Sync,
+          F::Error: Send + Sync
+{}
+
+unsafe impl<F> Send for Inner<F>
+    where F: Future + Send,
+          F::Item: Send + Sync,
+          F::Error: Send + Sync
+{}
 
 impl<F> fmt::Debug for Inner<F>
     where F: Future + fmt::Debug,

futures-util/src/stream/mod.rs

@@ -881,9 +881,6 @@ pub trait StreamExt: Stream {
     /// stream will be polled in a round-robin fashion, and whenever a stream is
     /// ready to yield an item that item is yielded.
     ///
-    /// The `select` function is similar to `merge` except that it requires both
-    /// streams to have the same item and error types.
-    ///
     /// Error are passed through from either stream.
     fn select<S>(self, other: S) -> Select<Self, S>
         where S: Stream<Item = Self::Item, Error = Self::Error>,

futures/tests/all.rs

@@ -268,6 +268,14 @@ fn join_all_iter_lifetime() {
     assert_done(|| sizes(vec![&[1,2,3], &[], &[0]]), Ok(vec![3, 0, 1]));
 }
 
+#[test]
+fn join_all_from_iter() {
+    assert_done(
+        || vec![f_ok(1), f_ok(2)].into_iter().collect::<JoinAll<_>>(),
+        Ok(vec![1, 2]),
+    )
+}
+
 #[test]
 fn select2() {
     assert_done(|| f_ok(2).select(empty()).then(unselect), Ok(2));

futures/tests/async_await/pinned.rs

@@ -99,6 +99,8 @@ fn main() {
     assert_eq!(block_on_stable(qux2(17)), Ok(17));
     assert_eq!(block_on(boxed(17)), Ok(17));
     assert_eq!(block_on(boxed_send(17)), Ok(17));
+    assert_eq!(block_on(boxed_borrow(&17)), Ok(17));
+    assert_eq!(block_on(boxed_send_borrow(&17)), Ok(17));
     assert_eq!(block_on_stable(uses_async_for()), Ok(vec![0, 1]));
 }
 

futures/tests/future_flatten_stream.rs

@@ -29,7 +29,7 @@ impl<T, E> Stream for PanickingStream<T, E> {
     type Item = T;
     type Error = E;
 
-    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
+    fn poll_next(&mut self, _: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
         panic!()
     }
 }

futures/tests/select_all.rs

@@ -1,7 +1,6 @@
 extern crate futures;
 
 use futures::executor::block_on;
-use futures::prelude::*;
 use futures::future::{ok, select_all, err};
 
 #[test]

futures/tests/sink.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::rc::Rc;
 use std::cell::{Cell, RefCell};
 use std::sync::atomic::{Ordering, AtomicBool};
+use std::collections::VecDeque;
 
 use futures::prelude::*;
 use futures::future::ok;
@@ -26,6 +27,17 @@ fn vec_sink() {
     assert_done(move || v.flush(), Ok(vec![0, 1]));
 }
 
+#[test]
+fn vecdeque_sink() {
+    let mut deque = VecDeque::new();
+    deque.start_send(2).unwrap();
+    deque.start_send(3).unwrap();
+
+    assert_eq!(deque.pop_front(), Some(2));
+    assert_eq!(deque.pop_front(), Some(3));
+    assert_eq!(deque.pop_front(), None);
+}
+
 #[test]
 fn send() {
     let v = Vec::new();
@@ -395,7 +407,7 @@ fn fanout_backpressure() {
     let sink = left_send.fanout(right_send);
 
     let sink = block_on(StartSendFut::new(sink, 0)).unwrap();
- 
+
     flag_cx(|flag, cx| {
         let mut task = sink.send(2);
         assert!(!flag.get());