Repository: alexcrichton/futures-rs
Pull request #153 (2016-09-14 04:38:49)

Future::flatten_to_stream

stepancheg: > I wonder if we can come up with a perhaps more ergonomic name? Something like `streamed` or similar?

I don't like `flatten_to_stream` name either, but I like `flat` part of it, because it emphasises the fact that it converts a collection of collections into a collection as in `flatten` or `flat_map` on iterators.

Moreover, I think the better purpose of `streamed` identifier is for a combinator that [converts a future into single element stream](https://github.com/stepancheg/grpc-rust/blob/master/src/futures_misc/future_to_stream_once.rs).
stepancheg: Updated the patch, addresses all the issues except:
* name is kept `flatten_to_stream`
* `FailedState` is kept separate enum

(see comments above)
alexcrichton: Perhaps this could be called `into_stream`?
alexcrichton: Hm I guess you could conceptually think of any future as a stream of one element, which arguably is the most generic way to think about futures/streams and what one might expect `into_stream` to do instead, so maybe not.

I still feel that `flatten_to_stream` is a bit long here, but I agree that keeping the name "flat" is probably good as it's what's happening here.

Hm @aturon or @carllerche, any ideas on naming here?

Patch:
diff --git a/src/flatten_stream.rs b/src/flatten_stream.rs
new file mode 100644
index 00000000..f175c2a7
--- /dev/null
+++ b/src/flatten_stream.rs
@@ -0,0 +1,86 @@
+use {Async, Future, Poll};
+use stream::Stream;
+
+/// Future for the `flatten_stream` combinator, flattening a
+/// future-of-a-stream to get just the result of the final stream as a stream.
+///
+/// This is created by the `Future::flatten_stream` method.
+#[must_use = "streams do nothing unless polled"]
+pub struct FlattenStream<F>
+    where F: Future,
+          <F as Future>::Item: Stream<Error=F::Error>,
+{
+    state: State<F>
+}
+
+pub fn new<F>(f: F) -> FlattenStream<F>
+    where F: Future,
+          <F as Future>::Item: Stream<Error=F::Error>,
+{
+    FlattenStream {
+        state: State::Future(f)
+    }
+}
+
+enum State<F>
+    where F: Future,
+          <F as Future>::Item: Stream<Error=F::Error>,
+{
+    // future is not yet called or called and not ready
+    Future(F),
+    // future resolved to Stream
+    Stream(F::Item),
+    // EOF after future resolved to error
+    Eof,
+    // after EOF after future resolved to error
+    Done,
+}
+
+impl<F> Stream for FlattenStream<F>
+    where F: Future,
+          <F as Future>::Item: Stream<Error=F::Error>,
+{
+    type Item = <F::Item as Stream>::Item;
+    type Error = <F::Item as Stream>::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        loop {
+            let (next_state, ret_opt) = match self.state {
+                State::Future(ref mut f) => {
+                    match f.poll() {
+                        Ok(Async::NotReady) => {
+                            // State is not changed, early return.
+                            return Ok(Async::NotReady)
+                        },
+                        Ok(Async::Ready(stream)) => {
+                            // Future resolved to stream.
+                            // We do not return, but poll that
+                            // stream in the next loop iteration.
+                            (State::Stream(stream), None)
+                        }
+                        Err(e) => {
+                            (State::Eof, Some(Err(e)))
+                        }
+                    }
+                }
+                State::Stream(ref mut s) => {
+                    // Just forward call to the stream,
+                    // do not track its state.
+                    return s.poll();
+                }
+                State::Eof => {
+                    (State::Done, Some(Ok(Async::Ready(None))))
+                }
+                State::Done => {
+                    panic!("poll called after eof");
+                }
+            };
+
+            self.state = next_state;
+            if let Some(ret) = ret_opt {
+                return ret;
+            }
+        }
+    }
+}
+
diff --git a/src/lib.rs b/src/lib.rs
index 406be4e6..fa85aa16 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -184,6 +184,7 @@ pub use lazy::{lazy, Lazy};
 // combinators
 mod and_then;
 mod flatten;
+mod flatten_stream;
 mod fuse;
 mod join;
 mod map;
@@ -193,6 +194,7 @@ mod select;
 mod then;
 pub use and_then::AndThen;
 pub use flatten::Flatten;
+pub use flatten_stream::FlattenStream;
 pub use fuse::Fuse;
 pub use join::{Join, Join3, Join4, Join5};
 pub use map::Map;
@@ -731,6 +733,40 @@ pub trait Future {
                         _>(f)
     }
 
+    /// Flatten the execution of this future when the successful result of this
+    /// future is a stream.
+    ///
+    /// This can be useful when stream initialization is deferred, and it is
+    /// convenient to work with that stream as if stream was available at the
+    /// call site.
+    ///
+    /// Note that this function consumes this future and returns a wrapped
+    /// version of it.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use futures::*;
+    /// use futures::stream::Stream;
+    ///
+    /// let stream_items = vec![Ok(17), Err(true), Ok(19)];
+    /// let future_of_a_stream = finished::<_, bool>(stream::iter(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());
+    /// ```
+    fn flatten_stream(self) -> FlattenStream<Self>
+        where <Self as Future>::Item: stream::Stream<Error=Self::Error>,
+              Self: Sized
+    {
+        flatten_stream::new(self)
+    }
+
     /// Fuse a future such that `poll` will never again be called once it has
     /// completed.
     ///


Test patch:
diff --git a/tests/future_flatten_stream.rs b/tests/future_flatten_stream.rs
new file mode 100644
index 00000000..a2cc5017
--- /dev/null
+++ b/tests/future_flatten_stream.rs
@@ -0,0 +1,43 @@
+extern crate core;
+extern crate futures;
+
+use core::marker;
+
+use futures::*;
+use futures::stream::Stream;
+
+#[test]
+fn successful_future() {
+    let stream_items = vec![Ok(17), Err(true), Ok(19)];
+    let future_of_a_stream = finished::<_, bool>(stream::iter(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());
+}
+
+struct PanickingStream<T, E> {
+    _marker: marker::PhantomData<(T, E)>
+}
+
+impl<T, E> Stream for PanickingStream<T, E> {
+    type Item = T;
+    type Error = E;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        panic!()
+    }
+}
+
+#[test]
+fn failed_future() {
+    let future_of_a_stream = failed::<PanickingStream<bool, u32>, _>(10);
+    let stream = future_of_a_stream.flatten_stream();
+    let mut iter = stream.wait();
+    assert_eq!(Err(10), iter.next().unwrap());
+    assert_eq!(None, iter.next());
+}