Repository: alexcrichton/futures-rs
Pull request #394 (2017-02-20 20:24:53)

add Stream::concat
A possible solution to #390. Another name I considered was `extend`, but maybe that could be confused with some how extending the actual stream.

carllerche: Given that the container is provided, I would suggest that `extend` is not the correct fn name.

It seems that this could be better functionality provided as a separate function? Something like:

```rust
future::extend(my_buffer, my_stream) -> impl Future<Item = MyBuffer>`
```
alexcrichton: Yeah I'm not totally sure on the naming here, I typically expect "concatenate" to not take a buffer as an argument and maybe even take a separator, but it does neither here! I sorta like the name 'extend' actually as you're extending a provided buffer.

@carllerche I'd prefer a trait method for this as it's a bit more ergonomic to call (one less import), but do you think it fits more with the other free-function combinators that we have?
carllerche: I generally prefer mirroring std as much as possible. In this case, the goal is to provide an equivalent to the Extend trait for futures. That would mean creating a new trait: FutureExtend or something like that. 

I also prefer to not use a few traits (Future, Stream) as essentially a flat namespace for all functions. It hurts discoverability. 

I do think that before commiting to an API it would be worth exploring how specialization could potentially handle the case of the Stream of bytes knowing the final lengh. 
alexcrichton: @carllerche I'm not sure I follow the point about `FutureExtend`? With this use case at least it seems like there's not much of a use case for appending an element to a collection taking a long time (e.g. needing to itself return a future). I'd imagine that `Extend` (or some similar trait) ends up being fine in this case.

I agree that a flat namespace isn't great, but we also don't want to leave useful functionality for everyone to pull in an extra crate, so we need to make sure we've at least got the essentials in this crate.
seanmonstar: As for the name `concat`, it came to mind because of my JavaScript background, where `Array.concat` takes any number of arrays and concatenates them all into a single array: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/concat

I can see `stream.extend(vec)` making sense, but I can also see someone reading it like `stream.extend(other_stream)` (if they hadn't checked the docs).

As for the failing tests, I noticed that `catch_unwind` was failing for me locally on nightly also. I haven't looked to see if this was a bug in nightly, or if futures needs to be updated.


carllerche: After re-reading, I think that I am (mostly) happy w/ the PR. Maybe, the one thing I would suggest is since this is `concat`, to not pass in an argument and instead, extend onto the first value returned by the stream.

I *believe* that the `Concat` type could, in the future, be specialized to be smart and `reserve` the amount of space necessary if the stream can provide that info. For example a HTTP body stream impl could specialize the `Future::poll` implementation and reserve the amount of space specified by content-length.

So, given all that, w/ the change I suggested above, I'm 👍 

@alexcrichton Re my comment about `Extend`, it isn't actually applicable to this specific case because this is a "flat extend" (concat). (I was misunderstanding).

However, to elaborate. [`Extend`](https://doc.rust-lang.org/std/iter/trait.Extend.html) takes an `Iterator`. Given that `Stream` is the async version of `Iterator`, `FutureExtend` would take a stream, and extend self w/ the received values. It needs to be "async" since values are received from the stream "async".
seanmonstar: I want to make the changes needed to get this done, but I admit I'm not sure what exactly they are. Sounds like Alex prefers `Extend`, Carl prefers `Concat`.

Additionally, there's a suggestion to not provide a target to fill up, but to make use of the first item in the stream. This could allow something neat, like if it is a `Stream` of `Bytes`, the `Bytes` could maybe just expand its length instead of copying the other `Bytes`. However, providing the target directly allows someone to `reserve` beforehand the size they expect (which is usually pretty easy to know in something like HTTP, by checking the `Content-Length` header first).
alexcrichton: Oh I don't feel too strongly about `Extend`, `Concat` I could live with. @carllerche how do you feel about taking in an argument to do optional reservation if necessary?
carllerche: If `concat` uses the first item, reservation could be done w/ `chain`:

```rust
let body = BytesMut::with_capacity(content_length)
   .chain(body_stream)
   .concat();
```

Reservation could also be done in the future w/ specialization.

Given those two options, I would opt for the more ergonomic API: not taking an argument.

Patch:
diff --git a/src/stream/concat.rs b/src/stream/concat.rs
new file mode 100644
index 00000000..99323cfc
--- /dev/null
+++ b/src/stream/concat.rs
@@ -0,0 +1,81 @@
+use core::mem;
+
+use {Poll, Async};
+use future::Future;
+use stream::Stream;
+
+/// A stream combinator to concatenate the results of a stream into the first
+/// yielded item.
+///
+/// This structure is produced by the `Stream::concat` method.
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Concat<S>
+    where S: Stream,
+{
+    stream: S,
+    extend: Inner<S::Item>,
+}
+
+pub fn new<S>(s: S) -> Concat<S>
+    where S: Stream,
+          S::Item: Extend<<S::Item as IntoIterator>::Item> + IntoIterator,
+{
+    Concat {
+        stream: s,
+        extend: Inner::First,
+    }
+}
+
+impl<S> Future for Concat<S>
+    where S: Stream,
+          S::Item: Extend<<S::Item as IntoIterator>::Item> + IntoIterator,
+
+{
+    type Item = S::Item;
+    type Error = S::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {
+            match self.stream.poll() {
+                Ok(Async::Ready(Some(i))) => {
+                    match self.extend {
+                        Inner::First => {
+                            self.extend = Inner::Extending(i);
+                        },
+                        Inner::Extending(ref mut e) => {
+                            e.extend(i);
+                        },
+                        Inner::Done => unreachable!(),
+                    }
+                },
+                Ok(Async::Ready(None)) => return Ok(Async::Ready(expect(self.extend.take()))),
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Err(e) => {
+                    self.extend.take();
+                    return Err(e)
+                }
+            }
+        }
+    }
+}
+
+#[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")
+}
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 19dee75b..16c56789 100755
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -27,6 +27,7 @@ pub use self::repeat::{repeat, Repeat};
 
 mod and_then;
 mod chain;
+mod concat;
 mod empty;
 mod filter;
 mod filter_map;
@@ -52,6 +53,7 @@ mod zip;
 mod forward;
 pub use self::and_then::AndThen;
 pub use self::chain::Chain;
+pub use self::concat::Concat;
 pub use self::empty::{Empty, empty};
 pub use self::filter::Filter;
 pub use self::filter_map::FilterMap;
@@ -532,6 +534,40 @@ pub trait Stream {
         collect::new(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.
+    ///
+    /// # 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.concat();
+    /// assert_eq!(result.wait(), Ok(vec![7, 8, 9, 4, 5, 6, 1, 2, 3]));
+    /// ```
+    fn concat(self) -> Concat<Self>
+        where Self: Sized,
+              Self::Item: Extend<<Self::Item as IntoIterator>::Item> + IntoIterator,
+    {
+        concat::new(self)
+    }
+
     /// Execute an accumulating computation over a stream, collecting all the
     /// values into one final result.
     ///


Test patch:
diff --git a/tests/stream.rs b/tests/stream.rs
index 6b1bfae6..9f9971bc 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -321,3 +321,12 @@ fn forward() {
     assert_done(move || iter(vec![Ok(4), Ok(5)]).forward(v).map(|(_, s)| s),
                 Ok::<_, ()>(vec![0, 1, 2, 3, 4, 5]));
 }
+
+#[test]
+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.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.concat(), Err(()));
+}