Merge pull request #557 from alexcrichton/add-iter-ok

Add an `iter_ok` adaptor for non-Result iterators

Author: alexcrichton

src/stream/iter.rs

@@ -1,3 +1,6 @@
+#![deprecated(note = "implemention moved to `iter_ok`")]
+#![allow(deprecated)]
+
 use {Async, Poll};
 use stream::Stream;
 

src/stream/iter_ok.rs

@@ -0,0 +1,48 @@
+use core::marker;
+
+use {Async, Poll};
+use stream::Stream;
+
+/// A stream which is just a shim over an underlying instance of `Iterator`.
+///
+/// This stream will never block and is always ready.
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct IterOk<I, E> {
+    iter: I,
+    _marker: marker::PhantomData<fn() -> E>,
+}
+
+/// Converts an `Iterator` over into a `Stream` which is always ready
+/// to yield the next value.
+///
+/// Iterators in Rust don't express the ability to block, so this adapter
+/// simply always calls `iter.next()` and returns that.
+///
+/// ```rust
+/// use futures::*;
+///
+/// let mut stream = stream::iter_ok::<_, ()>(vec![17, 19]);
+/// assert_eq!(Ok(Async::Ready(Some(17))), stream.poll());
+/// assert_eq!(Ok(Async::Ready(Some(19))), stream.poll());
+/// assert_eq!(Ok(Async::Ready(None)), stream.poll());
+/// ```
+pub fn iter_ok<I, E>(i: I) -> IterOk<I::IntoIter, E>
+    where I: IntoIterator,
+{
+    IterOk {
+        iter: i.into_iter(),
+        _marker: marker::PhantomData,
+    }
+}
+
+impl<I, E> Stream for IterOk<I, E>
+    where I: Iterator,
+{
+    type Item = I::Item;
+    type Error = E;
+
+    fn poll(&mut self) -> Poll<Option<I::Item>, E> {
+        Ok(Async::Ready(self.iter.next()))
+    }
+}

src/stream/mod.rs

@@ -18,9 +18,13 @@
 use {IntoFuture, Poll};
 
 mod iter;
+#[allow(deprecated)]
 pub use self::iter::{iter, Iter};
 #[cfg(feature = "with-deprecated")]
+#[allow(deprecated)]
 pub use self::Iter as IterStream;
+mod iter_ok;
+pub use self::iter_ok::{iter_ok, IterOk};
 
 mod repeat;
 pub use self::repeat::{repeat, Repeat};

src/stream/once.rs

@@ -1,15 +1,12 @@
-use core;
-
-use Poll;
-use stream;
+use {Poll, Async};
 use stream::Stream;
 
 /// A stream which emits single element and then EOF.
 ///
 /// This stream will never block and is always ready.
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
-pub struct Once<T, E>(stream::Iter<core::iter::Once<Result<T, E>>>);
+pub struct Once<T, E>(Option<Result<T, E>>);
 
 /// Creates a stream of single element
 ///
@@ -21,14 +18,18 @@ pub struct Once<T, E>(stream::Iter<core::iter::Once<Result<T, E>>>);
 /// assert_eq!(Ok(Async::Ready(None)), stream.poll());
 /// ```
 pub fn once<T, E>(item: Result<T, E>) -> Once<T, E> {
-    Once(stream::iter(core::iter::once(item)))
+    Once(Some(item))
 }
 
 impl<T, E> Stream for Once<T, E> {
     type Item = T;
     type Error = E;
 
     fn poll(&mut self) -> Poll<Option<T>, E> {
-        self.0.poll()
+        match self.0.take() {
+            Some(Ok(e)) => Ok(Async::Ready(Some(e))),
+            Some(Err(e)) => Err(e),
+            None => Ok(Async::Ready(None)),
+        }
     }
 }

src/sync/mpsc/mod.rs

@@ -484,6 +484,24 @@ impl<T> Sender<T> {
             Async::Ready(())
         }
     }
+
+    /// Attempts to send a message on this `Sender<T>` without blocking.
+    ///
+    /// This function, unlike `start_send`, is safe to call whether or not it's
+    /// being called on a task or not. Note that this function, however, will
+    /// *not* attempt to block the current task if the message cannot be sent.
+    ///
+    /// It is not recommended to call this function from inside of a future,
+    /// only from an external thread where you've otherwise arranged to be
+    /// notified when the channel is no longer full.
+    pub fn try_send(&self, msg: T) -> Result<(), SendError<T>> {
+        match self.inc_num_messages(false) {
+            Some(_park_self) => {}
+            None => return Err(SendError(msg)),
+        };
+        self.queue_push_and_signal(Some(msg));
+        Ok(())
+    }
 }
 
 impl<T> Sink for Sender<T> {

tests/bilock.rs

@@ -59,7 +59,7 @@ fn concurrent() {
         a: Some(a),
         remaining: N,
     };
-    let b = stream::iter((0..N).map(Ok::<_, ()>)).fold(b, |b, _n| {
+    let b = stream::iter_ok::<_, ()>((0..N)).fold(b, |b, _n| {
         b.lock().map(|mut b| {
             *b += 1;
             b.unlock()

tests/future_flatten_stream.rs

@@ -9,14 +9,13 @@ use futures::stream;
 
 #[test]
 fn successful_future() {
-    let stream_items = vec![Ok(17), Err(true), Ok(19)];
-    let future_of_a_stream = ok::<_, bool>(stream::iter(stream_items));
+    let stream_items = vec![17, 19];
+    let future_of_a_stream = ok::<_, bool>(stream::iter_ok(stream_items));
 
     let stream = future_of_a_stream.flatten_stream();
 
     let mut iter = stream.wait();
     assert_eq!(Ok(17), iter.next().unwrap());
-    assert_eq!(Err(true), iter.next().unwrap());
     assert_eq!(Ok(19), iter.next().unwrap());
     assert_eq!(None, iter.next());
 }

tests/mpsc.rs

@@ -354,3 +354,23 @@ fn is_ready<T>(res: &AsyncSink<T>) -> bool {
         _ => false,
     }
 }
+
+#[test]
+fn try_send() {
+    const N: usize = 3000;
+    let (tx, rx) = mpsc::channel(0);
+
+    let t = thread::spawn(move || {
+        for i in 0..N {
+            loop {
+                if tx.try_send(i).is_ok() {
+                    break
+                }
+            }
+        }
+    });
+    for (i, j) in rx.wait().enumerate() {
+        assert_eq!(i, j.unwrap());
+    }
+    t.join().unwrap();
+}

tests/sink.rs

@@ -44,14 +44,14 @@ fn send() {
 fn send_all() {
     let v = Vec::new();
 
-    let (v, _) = v.send_all(stream::iter(vec![Ok(0), Ok(1)])).wait().unwrap();
+    let (v, _) = v.send_all(stream::iter_ok(vec![0, 1])).wait().unwrap();
     assert_eq!(v, vec![0, 1]);
 
-    let (v, _) = v.send_all(stream::iter(vec![Ok(2), Ok(3)])).wait().unwrap();
+    let (v, _) = v.send_all(stream::iter_ok(vec![2, 3])).wait().unwrap();
     assert_eq!(v, vec![0, 1, 2, 3]);
 
     assert_done(
-        move || v.send_all(stream::iter(vec![Ok(4), Ok(5)])).map(|(v, _)| v),
+        move || v.send_all(stream::iter_ok(vec![4, 5])).map(|(v, _)| v),
         Ok(vec![0, 1, 2, 3, 4, 5]));
 }
 
@@ -171,7 +171,7 @@ fn with_as_map() {
 // test simple use of with_flat_map
 fn with_flat_map() {
     let sink = Vec::new().with_flat_map(|item| {
-        stream::iter(vec!(item; item).into_iter().map(Ok))
+        stream::iter_ok(vec![item; item])
     });
     let sink = sink.send(0).wait().unwrap();
     let sink = sink.send(1).wait().unwrap();

tests/split.rs

@@ -1,7 +1,7 @@
 extern crate futures;
 
 use futures::{Future, StartSend, Sink, Stream, Poll};
-use futures::stream::iter;
+use futures::stream::iter_ok;
 
 struct Join<T, U>(T, U);
 
@@ -37,7 +37,7 @@ impl<T, U: Sink> Sink for Join<T, U> {
 fn test_split() {
     let mut dest = Vec::new();
     {
-        let j = Join(iter(vec![Ok(10), Ok(20), Ok(30)]), &mut dest);
+        let j = Join(iter_ok(vec![10, 20, 30]), &mut dest);
         let (sink, stream) = j.split();
         let j = sink.reunite(stream).expect("test_split: reunite error");
         let (sink, stream) = j.split();

tests/stream.rs

@@ -4,13 +4,39 @@ extern crate futures;
 use futures::{Async, Future, Poll, Sink, Stream};
 use futures::executor;
 use futures::future::{err, ok};
-use futures::stream::{empty, iter, poll_fn, Peekable};
+use futures::stream::{empty, iter_ok, poll_fn, Peekable};
 use futures::sync::oneshot;
 use futures::sync::mpsc;
 
 mod support;
 use support::*;
 
+pub struct Iter<I> {
+    iter: I,
+}
+
+pub fn iter<J, T, E>(i: J) -> Iter<J::IntoIter>
+    where J: IntoIterator<Item=Result<T, E>>,
+{
+    Iter {
+        iter: i.into_iter(),
+    }
+}
+
+impl<I, T, E> Stream for Iter<I>
+    where I: Iterator<Item=Result<T, E>>,
+{
+    type Item = T;
+    type Error = E;
+
+    fn poll(&mut self) -> Poll<Option<T>, E> {
+        match self.iter.next() {
+            Some(Ok(e)) => Ok(Async::Ready(Some(e))),
+            Some(Err(e)) => Err(e),
+            None => Ok(Async::Ready(None)),
+        }
+    }
+}
 
 fn list() -> Box<Stream<Item=i32, Error=u32> + Send> {
     let (tx, rx) = mpsc::channel(1);
@@ -310,44 +336,45 @@ fn chunks_panic_on_cap_zero() {
 
 #[test]
 fn select() {
-    let a = iter(vec![Ok::<_, u32>(1), Ok(2), Ok(3)]);
-    let b = iter(vec![Ok(4), Ok(5), Ok(6)]);
+    let a = iter_ok::<_, u32>(vec![1, 2, 3]);
+    let b = iter_ok(vec![4, 5, 6]);
     assert_done(|| a.select(b).collect(), Ok(vec![1, 4, 2, 5, 3, 6]));
 
-    let a = iter(vec![Ok::<_, u32>(1), Ok(2), Ok(3)]);
-    let b = iter(vec![Ok(1), Ok(2)]);
+    let a = iter_ok::<_, u32>(vec![1, 2, 3]);
+    let b = iter_ok(vec![1, 2]);
     assert_done(|| a.select(b).collect(), Ok(vec![1, 1, 2, 2, 3]));
 
-    let a = iter(vec![Ok(1), Ok(2)]);
-    let b = iter(vec![Ok::<_, u32>(1), Ok(2), Ok(3)]);
+    let a = iter_ok(vec![1, 2]);
+    let b = iter_ok::<_, u32>(vec![1, 2, 3]);
     assert_done(|| a.select(b).collect(), Ok(vec![1, 1, 2, 2, 3]));
 }
 
 #[test]
 fn forward() {
     let v = Vec::new();
-    let v = iter(vec![Ok::<_, ()>(0), Ok(1)]).forward(v).wait().unwrap().1;
+    let v = iter_ok::<_, ()>(vec![0, 1]).forward(v).wait().unwrap().1;
     assert_eq!(v, vec![0, 1]);
 
-    let v = iter(vec![Ok::<_, ()>(2), Ok(3)]).forward(v).wait().unwrap().1;
+    let v = iter_ok::<_, ()>(vec![2, 3]).forward(v).wait().unwrap().1;
     assert_eq!(v, vec![0, 1, 2, 3]);
 
-    assert_done(move || iter(vec![Ok(4), Ok(5)]).forward(v).map(|(_, s)| s),
+    assert_done(move || iter_ok(vec![4, 5]).forward(v).map(|(_, s)| s),
                 Ok::<_, ()>(vec![0, 1, 2, 3, 4, 5]));
 }
 
 #[test]
+#[allow(deprecated)]
 fn concat() {
-    let a = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Ok(vec![4, 5, 6]), Ok(vec![7, 8, 9])]);
-    assert_done(move || a.concat2(), Ok(vec![1, 2, 3, 4, 5, 6, 7, 8, 9]));
+    let a = iter_ok::<_, ()>(vec![vec![1, 2, 3], vec![4, 5, 6], vec![7, 8, 9]]);
+    assert_done(move || a.concat(), Ok(vec![1, 2, 3, 4, 5, 6, 7, 8, 9]));
 
     let b = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Err(()), Ok(vec![7, 8, 9])]);
-    assert_done(move || b.concat2(), Err(()));
+    assert_done(move || b.concat(), Err(()));
 }
 
 #[test]
 fn concat2() {
-    let a = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Ok(vec![4, 5, 6]), Ok(vec![7, 8, 9])]);
+    let a = iter_ok::<_, ()>(vec![vec![1, 2, 3], vec![4, 5, 6], vec![7, 8, 9]]);
     assert_done(move || a.concat2(), Ok(vec![1, 2, 3, 4, 5, 6, 7, 8, 9]));
 
     let b = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Err(()), Ok(vec![7, 8, 9])]);

tests/stream_catch_unwind.rs

@@ -5,8 +5,7 @@ use futures::stream::Stream;
 
 #[test]
 fn panic_in_the_middle_of_the_stream() {
-    let stream = stream::iter::<_, Option<i32>, bool>(vec![
-        Some(10), None, Some(11)].into_iter().map(Ok));
+    let stream = stream::iter_ok::<_, bool>(vec![Some(10), None, Some(11)]);
 
     // panic on second element
     let stream_panicking = stream.map(|o| o.unwrap());
@@ -19,8 +18,7 @@ fn panic_in_the_middle_of_the_stream() {
 
 #[test]
 fn no_panic() {
-    let stream = stream::iter::<_, _, bool>(vec![
-        10, 11, 12].into_iter().map(Ok));
+    let stream = stream::iter_ok::<_, bool>(vec![10, 11, 12]);
 
     let mut iter = stream.catch_unwind().wait();
 

tests/unsync.rs

@@ -8,7 +8,7 @@ use futures::{Future, Stream, Sink, Async};
 use futures::unsync::oneshot;
 use futures::unsync::mpsc::{self, SendError};
 use futures::future::lazy;
-use futures::stream::{iter, unfold};
+use futures::stream::{iter_ok, unfold};
 
 use support::local_executor::Core;
 
@@ -66,7 +66,7 @@ fn mpsc_tx_err() {
 fn mpsc_backpressure() {
     let (tx, rx) = mpsc::channel::<i32>(1);
     lazy(move || {
-        iter(vec![1, 2, 3].into_iter().map(Ok))
+        iter_ok(vec![1, 2, 3])
             .forward(tx)
             .map_err(|e: SendError<i32>| panic!("{}", e))
             .join(rx.take(3).collect().map(|xs| {
@@ -79,7 +79,7 @@ fn mpsc_backpressure() {
 fn mpsc_unbounded() {
     let (tx, rx) = mpsc::unbounded::<i32>();
     lazy(move || {
-        iter(vec![1, 2, 3].into_iter().map(Ok))
+        iter_ok(vec![1, 2, 3])
             .forward(tx)
             .map_err(|e: SendError<i32>| panic!("{}", e))
             .join(rx.take(3).collect().map(|xs| {
@@ -103,7 +103,7 @@ fn mpsc_send_unpark() {
     let core = Core::new();
     let (tx, rx) = mpsc::channel::<i32>(1);
     let (donetx, donerx) = oneshot::channel();
-    core.spawn(iter(vec![1, 2].into_iter().map(Ok)).forward(tx)
+    core.spawn(iter_ok(vec![1, 2]).forward(tx)
         .then(|x: Result<_, SendError<i32>>| {
             assert!(x.is_err());
             donetx.send(()).unwrap();