Repository: alexcrichton/futures-rs (rust)

Issue:
Change stream::iter to not require an iterator of Result
Just work over an iterator of T and don't fail!

Pull request:
Add an `iter_ok` adaptor for non-Result iterators
The `iter` adaptor has been hard to use at least for me historically as it's
pretty rare to have an iterator of results. In addition with the planned change
to have errors terminate a `Stream` it doesn't necessarily make sense to
continue with an iterator of results!

Closes #480

Patch:
diff --git a/src/stream/iter.rs b/src/stream/iter.rs
index dd327a04..0cead2dd 100644
--- a/src/stream/iter.rs
+++ b/src/stream/iter.rs
@@ -1,3 +1,6 @@
+#![deprecated(note = "implemention moved to `iter_ok`")]
+#![allow(deprecated)]
+
 use {Async, Poll};
 use stream::Stream;
 
diff --git a/src/stream/iter_ok.rs b/src/stream/iter_ok.rs
new file mode 100644
index 00000000..3bb3e41a
--- /dev/null
+++ b/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()))
+    }
+}
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index d8d22674..2b411ff6 100755
--- a/src/stream/mod.rs
+++ b/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};
diff --git a/src/stream/once.rs b/src/stream/once.rs
index 9eb1551d..24fb327b 100644
--- a/src/stream/once.rs
+++ b/src/stream/once.rs
@@ -1,7 +1,4 @@
-use core;
-
-use Poll;
-use stream;
+use {Poll, Async};
 use stream::Stream;
 
 /// A stream which emits single element and then EOF.
@@ -9,7 +6,7 @@ use stream::Stream;
 /// 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,7 +18,7 @@ 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> {
@@ -29,6 +26,10 @@ impl<T, E> Stream for Once<T, E> {
     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)),
+        }
     }
 }
diff --git a/src/sync/mpsc/mod.rs b/src/sync/mpsc/mod.rs
index a901d0ad..d4320d11 100644
--- a/src/sync/mpsc/mod.rs
+++ b/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> {


Test patch:
diff --git a/tests/bilock.rs b/tests/bilock.rs
index bc1bafa5..bb79aa29 100644
--- a/tests/bilock.rs
+++ b/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()
diff --git a/tests/future_flatten_stream.rs b/tests/future_flatten_stream.rs
index 56ca6894..f9c97a8f 100644
--- a/tests/future_flatten_stream.rs
+++ b/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());
 }
diff --git a/tests/mpsc.rs b/tests/mpsc.rs
index f92d7927..ec484110 100644
--- a/tests/mpsc.rs
+++ b/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();
+}
diff --git a/tests/sink.rs b/tests/sink.rs
index 43a0bc8b..abbc9a0c 100644
--- a/tests/sink.rs
+++ b/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();
diff --git a/tests/split.rs b/tests/split.rs
index e00c72fd..81695c51 100644
--- a/tests/split.rs
+++ b/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();
diff --git a/tests/stream.rs b/tests/stream.rs
index ccd02ce9..51e3aa98 100644
--- a/tests/stream.rs
+++ b/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])]);
diff --git a/tests/stream_catch_unwind.rs b/tests/stream_catch_unwind.rs
index f93b301f..7278fe98 100644
--- a/tests/stream_catch_unwind.rs
+++ b/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();
 
diff --git a/tests/unsync.rs b/tests/unsync.rs
index 0ccbee11..f9d5b796 100644
--- a/tests/unsync.rs
+++ b/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();