Add StreamExt::flat_map

futures-util/src/stream/mod.rs

@@ -13,9 +13,9 @@ pub use futures_core::stream::{FusedStream, Stream, TryStream};
 #[allow(clippy::module_inception)]
 mod stream;
 pub use self::stream::{
-    Chain, Collect, Concat, Enumerate, Filter, FilterMap, Flatten, Fold, ForEach, Fuse, Inspect,
-    Map, Next, Peek, Peekable, Scan, SelectNextSome, Skip, SkipWhile, StreamExt, StreamFuture, Take,
-    TakeWhile, Then, Zip,
+    Chain, Collect, Concat, Enumerate, Filter, FilterMap, FlatMap, Flatten, Fold, ForEach, Fuse,
+    Inspect, Map, Next, Peek, Peekable, Scan, SelectNextSome, Skip, SkipWhile, StreamExt,
+    StreamFuture, Take, TakeWhile, Then, Zip,
 };
 
 #[cfg(feature = "std")]

futures-util/src/stream/stream/flat_map.rs

@@ -14,7 +14,7 @@ pub struct FlatMap<St, U, F> {
     inner_stream: Option<U>,
 }
 
-impl<St: Unpin, U, F> Unpin for FlatMap<St, U, F> {}
+impl<St: Unpin, U: Unpin, F> Unpin for FlatMap<St, U, F> {}
 
 impl<St, U, F> fmt::Debug for FlatMap<St, U, F>
 where
@@ -130,7 +130,10 @@ where
 
         if let Some(inner_stream_size_hint) = self.inner_stream.as_ref().map(|st| st.size_hint()) {
             (
-                stream_size_hint.0 + inner_stream_size_hint.0,
+                stream_size_hint
+                    .0
+                    .checked_add(inner_stream_size_hint.0)
+                    .unwrap_or(stream_size_hint.0),
                 if no_stream_items_left {
                     inner_stream_size_hint.1
                 } else {

futures-util/src/stream/stream/mod.rs

@@ -551,11 +551,11 @@ pub trait StreamExt: Stream {
     /// Maps a stream like [`StreamExt::map`] but flattens nested `Stream`s.
     ///
     /// [`StreamExt::map`] is very useful, but if it produces a `Stream` instead,
-    /// there's an extra layer of indirection. `flat_map()` will remove this extra
-    /// layer on its own.
+    /// you should chain cominators like `.map(f).flatten()` while this combinator
+    /// provides ability to write `.flat_map(f)` instead of chaining.
     ///
     /// The provided closure which produce inner streams is executed over all elements
-    /// of this stream as they are made available.
+    /// of stream as last inner stream is terminated and next stream item is available.
     ///
     /// Note that this function consumes the stream passed into it and returns a
     /// wrapped version of it, similar to the existing `flat_map` methods in the

futures/src/lib.rs

@@ -442,8 +442,8 @@ pub mod stream {
         try_unfold, TryUnfold,
 
         StreamExt,
-        Chain, Collect, Concat, Enumerate, Filter, FilterMap, Flatten, Fold,
-        Forward, ForEach, Fuse, StreamFuture, Inspect, Map, Next,
+        Chain, Collect, Concat, Enumerate, Filter, FilterMap, FlatMap, Flatten,
+        Fold, Forward, ForEach, Fuse, StreamFuture, Inspect, Map, Next,
         SelectNextSome, Peek, Peekable, Scan, Skip, SkipWhile, Take, TakeWhile,
         Then, Zip,