Repository: alexcrichton/futures-rs (rust)

Issue:
Stream::concat fails for empty streams
Hi!

`Stream::concat` seems to fail when the stream it is wrapping is empty (i.e. produces `Ready(None)` on the first poll).

To reproduce, compile and run this example:

```rust
extern crate futures;

use futures::{Future, Stream};

fn main() {
    let e = futures::stream::empty::<Vec<u8>, ()>();
    let mut c = e.concat();
    println!("Poll: {:?}", c.poll());
}
```

Result:

```
thread 'main' panicked at 'cannot poll Concat again', src/libcore/option.rs:705
stack backtrace:
   1:        0x10e381fff - std::sys::backtrace::tracing::imp::write::h6f1d53a70916b90d
   2:        0x10e383c1f - std::panicking::default_hook::{{closure}}::h137e876f7d3b5850
   3:        0x10e383126 - std::panicking::default_hook::h0ac3811ec7cee78c
   4:        0x10e3836cc - std::panicking::rust_panic_with_hook::hc303199e04562edf
   5:        0x10e383564 - std::panicking::begin_panic::h6ed03353807cf54d
   6:        0x10e383482 - std::panicking::begin_panic_fmt::hc321cece241bb2f5
   7:        0x10e3833e7 - rust_begin_unwind
   8:        0x10e3a7870 - core::panicking::panic_fmt::h27224b181f9f037f
   9:        0x10e3a78dd - core::option::expect_failed::h8606bc228cd3f504
  10:        0x10e379353 - <core::option::Option<T>>::expect::h78e9cd75a38fc6ae
  11:        0x10e37b6e4 - futures::stream::concat::expect::hcbba6ade024283b3
  12:        0x10e37ba5a - <futures::stream::concat::Concat<S> as futures::future::Future>::poll::h1c1ca2be81aca657
  13:        0x10e37c04f - rust_repro_futures_concat_empty_bug::main::ha26b06194c09e10f
  14:        0x10e38423a - __rust_maybe_catch_panic
  15:        0x10e382c3e - std::rt::lang_start::h538f8960e7644c80
  16:        0x10e37c189 - main
```

Expected behaviour:

`Concat<S>` implements `Future<Item=Option<S::Item>, Error=S::Error>`, to make it clear what  happens when the stream is empty and allow the user to statically verify that the case is handled.

If it's not an option to break the API, I would like it if at least this behaviour would be more clearly documented, and produces an error message that is more helpful in debugging the problem. Maybe we can also introduce a `concat_option` (or similar) method to provide a safer interface.

BTW I'd be excited to take on fixing this problem as my first contribution to `futures`.

Pull request:
implement Stream::concat_safe
Fixes #450.

I went with an `Option` here rather than leveraging `Default` because I see it as more flexible. It is easy to recover the `Default` behaviour if one wishes via:

```rust
s.concat_safe().map(|x| x.unwrap_or(default()))
```

However, the reverse (recovering the `Option` behaviour from an implementation using `Default`) is not so easy.

As soon as we can break the API (with futures 0.2, I'm assuming), we could move `concat` to use `Default` instead if we think this will provide additional convenience when used with types like `Vec`. In this case we might want to choose a different name for `concat_safe` at this point though, maybe `concat_option`? Alternatively we could introduce `concat_default` and make `concat_safe` the new concat.

Let me know what you think.

Patch:
diff --git a/src/stream/concat.rs b/src/stream/concat.rs
index 7ee0af98..1a1dedd9 100644
--- a/src/stream/concat.rs
+++ b/src/stream/concat.rs
@@ -1,4 +1,6 @@
 use core::mem;
+use core::fmt::{Debug, Formatter, Result as FmtResult};
+use core::default::Default;
 
 use {Poll, Async};
 use future::Future;
@@ -8,13 +10,67 @@ use stream::Stream;
 /// yielded item.
 ///
 /// This structure is produced by the `Stream::concat` method.
-#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Concat2<S>
+    where S: Stream,
+{
+    inner: ConcatSafe<S>
+}
+
+impl<S: Debug> Debug for Concat2<S> where S: Stream, S::Item: Debug {
+    fn fmt(&self, fmt: &mut Formatter) -> FmtResult {
+        fmt.debug_struct("Concat2")
+            .field("inner", &self.inner)
+            .finish()
+    }
+}
+
+pub fn new2<S>(s: S) -> Concat2<S>
+    where S: Stream,
+          S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator + Default,
+{
+    Concat2 {
+        inner: new_safe(s)
+    }
+}
+
+impl<S> Future for Concat2<S>
+    where S: Stream,
+          S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator + Default,
+
+{
+    type Item = S::Item;
+    type Error = S::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.inner.poll().map(|a| {
+            match a {
+                Async::NotReady => Async::NotReady,
+                Async::Ready(None) => Async::Ready(Default::default()),
+                Async::Ready(Some(e)) => Async::Ready(e)
+            }
+        })
+    }
+}
+
+
+/// A stream combinator to concatenate the results of a stream into the first
+/// yielded item.
+///
+/// This structure is produced by the `Stream::concat` method.
 #[must_use = "streams do nothing unless polled"]
 pub struct Concat<S>
     where S: Stream,
 {
-    stream: S,
-    extend: Inner<S::Item>,
+    inner: ConcatSafe<S>
+}
+
+impl<S: Debug> Debug for Concat<S> where S: Stream, S::Item: Debug {
+    fn fmt(&self, fmt: &mut Formatter) -> FmtResult {
+        fmt.debug_struct("Concat")
+            .field("inner", &self.inner)
+            .finish()
+    }
 }
 
 pub fn new<S>(s: S) -> Concat<S>
@@ -22,8 +78,7 @@ pub fn new<S>(s: S) -> Concat<S>
           S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator,
 {
     Concat {
-        stream: s,
-        extend: Inner::First,
+        inner: new_safe(s)
     }
 }
 
@@ -35,6 +90,44 @@ impl<S> Future for Concat<S>
     type Item = S::Item;
     type Error = S::Error;
 
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.inner.poll().map(|a| {
+            match a {
+                Async::NotReady => Async::NotReady,
+                Async::Ready(None) => panic!("attempted concatenation of empty stream"),
+                Async::Ready(Some(e)) => Async::Ready(e)
+            }
+        })
+    }
+}
+
+
+#[derive(Debug)]
+struct ConcatSafe<S>
+    where S: Stream,
+{
+    stream: S,
+    extend: Inner<S::Item>,
+}
+
+fn new_safe<S>(s: S) -> ConcatSafe<S>
+    where S: Stream,
+          S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator,
+{
+    ConcatSafe {
+        stream: s,
+        extend: Inner::First,
+    }
+}
+
+impl<S> Future for ConcatSafe<S>
+    where S: Stream,
+          S::Item: Extend<<<S as Stream>::Item as IntoIterator>::Item> + IntoIterator,
+
+{
+    type Item = Option<S::Item>;
+    type Error = S::Error;
+
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         loop {
             match self.stream.poll() {
@@ -49,10 +142,16 @@ impl<S> Future for Concat<S>
                         Inner::Done => unreachable!(),
                     }
                 },
-                Ok(Async::Ready(None)) => return Ok(Async::Ready(expect(self.extend.take()))),
+                Ok(Async::Ready(None)) => {
+                    match mem::replace(&mut self.extend, Inner::Done) {
+                        Inner::First => return Ok(Async::Ready(None)),
+                        Inner::Extending(e) => return Ok(Async::Ready(Some(e))),
+                        Inner::Done => panic!("cannot poll Concat again")
+                    }
+                },
                 Ok(Async::NotReady) => return Ok(Async::NotReady),
                 Err(e) => {
-                    self.extend.take();
+                    self.extend = Inner::Done;
                     return Err(e)
                 }
             }
@@ -60,22 +159,10 @@ impl<S> Future for Concat<S>
     }
 }
 
+
 #[derive(Debug)]
 enum Inner<E> {
     First,
     Extending(E),
     Done,
-}
-
-impl<E> Inner<E> {
-    fn take(&mut self) -> Option<E> {
-        match mem::replace(self, Inner::Done) {
-            Inner::Extending(e) => Some(e),
-            _ => None,
-        }
-    }
-}
-
-fn expect<T>(opt: Option<T>) -> T {
-    opt.expect("cannot poll Concat again")
-}
+}
\ No newline at end of file
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index c6d114a7..a0e50e57 100755
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -54,7 +54,7 @@ mod zip;
 mod forward;
 pub use self::and_then::AndThen;
 pub use self::chain::Chain;
-pub use self::concat::Concat;
+pub use self::concat::{Concat, Concat2};
 pub use self::empty::{Empty, empty};
 pub use self::filter::Filter;
 pub use self::filter_map::FilterMap;
@@ -540,9 +540,48 @@ pub trait Stream {
     /// destination, returning a future representing the end result.
     ///
     /// This combinator will extend the first item with the contents
-    /// of all the successful results of the stream. If an error
-    /// occurs, all the results will be dropped and the error will be
-    /// returned.
+    /// of all the successful results of the stream. If the stream is
+    /// empty, the default value will be returned. If an error occurs,
+    /// all the results will be dropped and the error will be returned.
+    ///
+    /// The name `concat2` is an intermediate measure until the release of
+    /// futures 0.2, at which point it will be renamed back to `concat`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use std::thread;
+    ///
+    /// use futures::{Future, Sink, Stream};
+    /// use futures::sync::mpsc;
+    ///
+    /// let (mut tx, rx) = mpsc::channel(1);
+    ///
+    /// thread::spawn(move || {
+    ///     for i in (0..3).rev() {
+    ///         let n = i * 3;
+    ///         tx = tx.send(vec![n + 1, n + 2, n + 3]).wait().unwrap();
+    ///     }
+    /// });
+    /// let result = rx.concat2();
+    /// assert_eq!(result.wait(), Ok(vec![7, 8, 9, 4, 5, 6, 1, 2, 3]));
+    /// ```
+    fn concat2(self) -> Concat2<Self>
+        where Self: Sized,
+              Self::Item: Extend<<<Self as Stream>::Item as IntoIterator>::Item> + IntoIterator + Default,
+    {
+        concat::new2(self)
+    }
+
+    /// Concatenate all results of a stream into a single extendable
+    /// destination, returning a future representing the end result.
+    ///
+    /// This combinator will extend the first item with the contents
+    /// of all the successful results of the stream. If an error occurs,
+    /// all the results will be dropped and the error will be returned.
+    ///
+    /// It's important to note that this function will panic if the stream
+    /// is empty, which is the reason for its deprecation.
     ///
     /// # Examples
     ///
@@ -563,6 +602,7 @@ pub trait Stream {
     /// let result = rx.concat();
     /// assert_eq!(result.wait(), Ok(vec![7, 8, 9, 4, 5, 6, 1, 2, 3]));
     /// ```
+    #[deprecated(since="0.1.14", note="please use `Stream::concat2` instead")]
     fn concat(self) -> Concat<Self>
         where Self: Sized,
               Self::Item: Extend<<<Self as Stream>::Item as IntoIterator>::Item> + IntoIterator,


Test patch:
diff --git a/tests/stream.rs b/tests/stream.rs
index ea32b2e9..2a776dd2 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -4,7 +4,7 @@ extern crate futures;
 use futures::{Poll, Future, Stream, Sink};
 use futures::executor;
 use futures::future::{ok, err};
-use futures::stream::{iter, Peekable, BoxStream};
+use futures::stream::{empty, iter, Peekable, BoxStream};
 use futures::sync::oneshot;
 use futures::sync::mpsc;
 
@@ -344,3 +344,15 @@ fn concat() {
     let b = iter(vec![Ok::<_, ()>(vec![1, 2, 3]), Err(()), Ok(vec![7, 8, 9])]);
     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])]);
+    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])]);
+    assert_done(move || b.concat2(), Err(()));
+
+    let c = empty::<Vec<()>, ()>();
+    assert_done(move || c.concat2(), Ok(vec![]))
+}