From ec98b41c85bb8c2891755fc6b86366fb85413a5e Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Thu, 17 Oct 2019 23:56:01 +0900
Subject: [PATCH 01/25] feat: Add FlattenCompat struct

---
 src/lib.rs                   |  1 +
 src/stream/stream/flatten.rs | 24 ++++++++++++++++++++++++
 src/stream/stream/mod.rs     |  1 +
 3 files changed, 26 insertions(+)
 create mode 100644 src/stream/stream/flatten.rs

diff --git a/src/lib.rs b/src/lib.rs
index e138c87d..afbad31e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -47,6 +47,7 @@
 #![doc(test(attr(allow(unused_extern_crates, unused_variables))))]
 #![doc(html_logo_url = "https://async.rs/images/logo--hero.svg")]
 #![recursion_limit = "1024"]
+#![feature(associated_type_bounds)]
 
 use cfg_if::cfg_if;
 
diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
new file mode 100644
index 00000000..fffda4dd
--- /dev/null
+++ b/src/stream/stream/flatten.rs
@@ -0,0 +1,24 @@
+use std::pin::Pin;
+
+use crate::stream::{IntoStream, Stream};
+use crate::task::{Context, Poll};
+
+/// Real logic of both `Flatten` and `FlatMap` which simply delegate to
+/// this type.
+#[derive(Clone, Debug)]
+struct FlattenCompat<S, U> {
+    stream: S,
+    frontiter: Option<U>,
+}
+impl<S, U> FlattenCompat<S, U> {
+    pin_utils::unsafe_unpinned!(stream: S);
+    pin_utils::unsafe_unpinned!(frontiter: Option<U>);
+
+    /// Adapts an iterator by flattening it, for use in `flatten()` and `flat_map()`.
+    pub fn new(stream: S) -> FlattenCompat<S, U> {
+        FlattenCompat {
+            stream,
+            frontiter: None,
+        }
+    }
+}
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index 764dc978..7047b033 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -106,6 +106,7 @@ cfg_if! {
 cfg_if! {
     if #[cfg(any(feature = "unstable", feature = "docs"))] {
         mod merge;
+        mod flatten;
 
         use std::pin::Pin;
 

From bb1416420d047638d2cc8be21920206c3053870a Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Thu, 17 Oct 2019 23:56:32 +0900
Subject: [PATCH 02/25] feat: Add Stream trait for FlattenCompat

---
 src/stream/stream/flatten.rs | 49 ++++++++++++++++++++++++++++++++++++
 1 file changed, 49 insertions(+)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index fffda4dd..7265d17f 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -22,3 +22,52 @@ impl<S, U> FlattenCompat<S, U> {
         }
     }
 }
+
+impl<S, U> Stream for FlattenCompat<S, U>
+where
+    S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
+    U: Stream + std::marker::Unpin,
+{
+    type Item = U::Item;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        loop {
+            if let Some(ref mut inner) = self.as_mut().frontiter() {
+                if let item @ Some(_) = futures_core::ready!(Pin::new(inner).poll_next(cx)) {
+                    return Poll::Ready(item);
+                }
+            }
+
+            match futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx)) {
+                None => return Poll::Ready(None),
+                Some(inner) => *self.as_mut().frontiter() = Some(inner.into_stream()),
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::FlattenCompat;
+
+    use crate::prelude::*;
+    use crate::task;
+
+    use std::collections::VecDeque;
+
+    #[test]
+    fn test_poll_next() -> std::io::Result<()> {
+        let inner1: VecDeque<u8> = vec![1, 2, 3].into_iter().collect();
+        let inner2: VecDeque<u8> = vec![4, 5, 6].into_iter().collect();
+
+        let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
+
+        task::block_on(async move {
+            let flat = FlattenCompat::new(s);
+            let v: Vec<u8> = flat.collect().await;
+
+            assert_eq!(v, vec![1, 2, 3, 4, 5, 6]);
+            Ok(())
+        })
+    }
+}

From 2dee2897509f0ac5c1a8e26d768bfdf3cbe54099 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 18 Oct 2019 10:43:21 +0900
Subject: [PATCH 03/25] Add FlatMap struct

---
 src/stream/stream/flatten.rs | 37 ++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 7265d17f..e922e94e 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -1,8 +1,44 @@
 use std::pin::Pin;
 
+use crate::prelude::*;
+use crate::stream::stream::map::Map;
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 
+#[allow(missing_debug_implementations)]
+pub struct FlatMap<S: Stream, U: IntoStream, F> {
+    inner: FlattenCompat<Map<S, F, S::Item, U>, U>,
+}
+
+impl<S, U, F> FlatMap<S, U, F>
+where
+    S: Stream,
+    U: IntoStream,
+    F: FnMut(S::Item) -> U,
+{
+    pin_utils::unsafe_pinned!(inner: FlattenCompat<Map<S, F, S::Item, U>, U>);
+
+    pub fn new(stream: S, f: F) -> FlatMap<S, U, F> {
+        FlatMap {
+            inner: FlattenCompat::new(stream.map(f)),
+        }
+    }
+}
+
+impl<S, U, F> Stream for FlatMap<S, U, F>
+where
+    S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
+    S::Item: std::marker::Unpin,
+    U: Stream + std::marker::Unpin,
+    F: FnMut(S::Item) -> U + std::marker::Unpin,
+{
+    type Item = U::Item;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.as_mut().inner().poll_next(cx)
+    }
+}
+
 /// Real logic of both `Flatten` and `FlatMap` which simply delegate to
 /// this type.
 #[derive(Clone, Debug)]
@@ -10,6 +46,7 @@ struct FlattenCompat<S, U> {
     stream: S,
     frontiter: Option<U>,
 }
+
 impl<S, U> FlattenCompat<S, U> {
     pin_utils::unsafe_unpinned!(stream: S);
     pin_utils::unsafe_unpinned!(frontiter: Option<U>);

From 2187a2a31d5f9c72e8ee32d2beae3c129bd8e80f Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 18 Oct 2019 10:43:36 +0900
Subject: [PATCH 04/25] feat: Add Stream::flat_map

---
 src/stream/stream/mod.rs | 41 ++++++++++++++++++++++++++++++++++++++--
 1 file changed, 39 insertions(+), 2 deletions(-)

diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index 7047b033..2f1a89f9 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -30,6 +30,7 @@ mod filter;
 mod filter_map;
 mod find;
 mod find_map;
+mod flatten;
 mod fold;
 mod for_each;
 mod fuse;
@@ -77,6 +78,7 @@ use try_for_each::TryForEeachFuture;
 
 pub use chain::Chain;
 pub use filter::Filter;
+pub use flatten::FlatMap;
 pub use fuse::Fuse;
 pub use inspect::Inspect;
 pub use map::Map;
@@ -93,6 +95,7 @@ use std::marker::PhantomData;
 
 use cfg_if::cfg_if;
 
+use crate::stream::IntoStream;
 use crate::utils::extension_trait;
 
 cfg_if! {
@@ -106,7 +109,6 @@ cfg_if! {
 cfg_if! {
     if #[cfg(any(feature = "unstable", feature = "docs"))] {
         mod merge;
-        mod flatten;
 
         use std::pin::Pin;
 
@@ -496,7 +498,6 @@ extension_trait! {
             #
             # }) }
             ```
-            
         "#]
         fn last(
             self,
@@ -570,6 +571,42 @@ extension_trait! {
             Filter::new(self, predicate)
         }
 
+        #[doc= r#"
+            Creates an stream that works like map, but flattens nested structure.
+
+            # Examples
+
+            Basic usage:
+
+            ```
+            # async_std::task::block_on(async {
+
+            use std::collections::VecDeque;
+            use async_std::prelude::*;
+            use async_std::stream::IntoStream;
+
+            let inner1: VecDeque<u8> = vec![1,2,3].into_iter().collect();
+            let inner2: VecDeque<u8> = vec![4,5,6].into_iter().collect();
+
+            let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
+
+            let flat= s.flat_map(|s| s.into_stream() );
+            let v: Vec<u8> = flat.collect().await;
+
+            assert_eq!(v, vec![1,2,3,4,5,6]);
+
+            # });
+            ```
+        "#]
+        fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F>
+            where
+                Self: Sized,
+                U: IntoStream,
+                F: FnMut(Self::Item) -> U,
+        {
+            FlatMap::new(self, f)
+        }
+
         #[doc = r#"
             Both filters and maps a stream.
 

From cd862083a5db3d04c4ff2cbbaa3eb96e50f04ccd Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 18 Oct 2019 12:19:38 +0900
Subject: [PATCH 05/25] Add Flatten struct

---
 src/stream/stream/flatten.rs | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index e922e94e..06e9ec2e 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -39,6 +39,21 @@ where
     }
 }
 
+pub struct Flatten<S: Stream>
+where
+    S::Item: IntoStream,
+{
+    inner: FlattenCompat<S, <S::Item as IntoStream>::IntoStream>,
+}
+
+impl<S: Stream<Item: IntoStream>> Flatten<S> {
+    pin_utils::unsafe_pinned!(inner: FlattenCompat<S, <S::Item as IntoStream>::IntoStream>);
+
+    pub fn new(stream: S) -> Flatten<S> {
+        Flatten { inner: FlattenCompat::new(stream) }
+    }
+}
+
 /// Real logic of both `Flatten` and `FlatMap` which simply delegate to
 /// this type.
 #[derive(Clone, Debug)]

From 8138afbfadd0b7f2640a28d9c0af0f59a59e3967 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 18 Oct 2019 12:20:02 +0900
Subject: [PATCH 06/25] feat: Add Stream trait for Flatten

---
 src/stream/stream/flatten.rs | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 06e9ec2e..5c761672 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -54,6 +54,19 @@ impl<S: Stream<Item: IntoStream>> Flatten<S> {
     }
 }
 
+impl<S, U> Stream for Flatten<S>
+where
+    S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
+    U: Stream + std::marker::Unpin,
+{
+    type Item = U::Item;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.as_mut().inner().poll_next(cx)
+    }
+}
+
+
 /// Real logic of both `Flatten` and `FlatMap` which simply delegate to
 /// this type.
 #[derive(Clone, Debug)]

From 176359afae6adf0f4202fde938f0857c9759c9d9 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 18 Oct 2019 12:20:28 +0900
Subject: [PATCH 07/25] Add Stream::flatten

---
 src/stream/stream/mod.rs | 36 +++++++++++++++++++++++++++++++++---
 1 file changed, 33 insertions(+), 3 deletions(-)

diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index 2f1a89f9..63d1d8cc 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -78,7 +78,7 @@ use try_for_each::TryForEeachFuture;
 
 pub use chain::Chain;
 pub use filter::Filter;
-pub use flatten::FlatMap;
+pub use flatten::{FlatMap, Flatten};
 pub use fuse::Fuse;
 pub use inspect::Inspect;
 pub use map::Map;
@@ -590,8 +590,7 @@ extension_trait! {
 
             let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
 
-            let flat= s.flat_map(|s| s.into_stream() );
-            let v: Vec<u8> = flat.collect().await;
+            let v :Vec<_> = s.flat_map(|s| s.into_stream()).collect().await;
 
             assert_eq!(v, vec![1,2,3,4,5,6]);
 
@@ -607,6 +606,37 @@ extension_trait! {
             FlatMap::new(self, f)
         }
 
+        #[doc = r#"
+            Creates an stream that flattens nested structure.
+
+            # Examples
+
+            Basic usage:
+
+            ```
+            # async_std::task::block_on(async {
+
+            use std::collections::VecDeque;
+            use async_std::prelude::*;
+
+            let inner1: VecDeque<u8> = vec![1,2,3].into_iter().collect();
+            let inner2: VecDeque<u8> = vec![4,5,6].into_iter().collect();
+            let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
+
+            let v: Vec<_> = s.flatten().collect().await;
+
+            assert_eq!(v, vec![1,2,3,4,5,6]);
+
+            # });
+        "#]
+        fn flatten(self) -> Flatten<Self>
+        where
+            Self: Sized,
+            Self::Item: IntoStream,
+        {
+            Flatten::new(self)
+        }
+
         #[doc = r#"
             Both filters and maps a stream.
 

From 410d16eaf6f5933bc5516a709c27134f227111e4 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 18 Oct 2019 13:20:44 +0900
Subject: [PATCH 08/25] Add docs + To unstable feature

---
 src/stream/stream/flatten.rs | 11 +++++++++++
 src/stream/stream/mod.rs     | 11 +++++++----
 2 files changed, 18 insertions(+), 4 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 5c761672..b9700876 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -5,6 +5,11 @@ use crate::stream::stream::map::Map;
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 
+/// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
+/// documentation for more.
+///
+/// [`flat_map`]: trait.Stream.html#method.flat_map
+/// [`Stream`]: trait.Stream.html
 #[allow(missing_debug_implementations)]
 pub struct FlatMap<S: Stream, U: IntoStream, F> {
     inner: FlattenCompat<Map<S, F, S::Item, U>, U>,
@@ -39,6 +44,12 @@ where
     }
 }
 
+/// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
+/// documentation for more.
+///
+/// [`flatten`]: trait.Stream.html#method.flatten
+/// [`Stream`]: trait.Stream.html
+#[allow(missing_debug_implementations)]
 pub struct Flatten<S: Stream>
 where
     S::Item: IntoStream,
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index 6802de90..8ad8cdc0 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -30,7 +30,6 @@ mod filter;
 mod filter_map;
 mod find;
 mod find_map;
-mod flatten;
 mod fold;
 mod for_each;
 mod fuse;
@@ -78,7 +77,6 @@ use try_for_each::TryForEeachFuture;
 
 pub use chain::Chain;
 pub use filter::Filter;
-pub use flatten::{FlatMap, Flatten};
 pub use fuse::Fuse;
 pub use inspect::Inspect;
 pub use map::Map;
@@ -93,17 +91,18 @@ pub use zip::Zip;
 use std::cmp::Ordering;
 use std::marker::PhantomData;
 
-use crate::stream::IntoStream;
-
 cfg_unstable! {
     use std::pin::Pin;
 
     use crate::future::Future;
     use crate::stream::FromStream;
+    use crate::stream::into_stream::IntoStream;
 
     pub use merge::Merge;
+    pub use flatten::{FlatMap, Flatten};
 
     mod merge;
+    mod flatten;
 }
 
 extension_trait! {
@@ -589,6 +588,8 @@ extension_trait! {
             # });
             ```
         "#]
+        #[cfg(feature = "unstable")]
+        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F>
             where
                 Self: Sized,
@@ -621,6 +622,8 @@ extension_trait! {
 
             # });
         "#]
+        #[cfg(feature = "unstable")]
+        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         fn flatten(self) -> Flatten<Self>
         where
             Self: Sized,

From 271b6f4a1c6a8096ba68682430a984001de19d56 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Fri, 25 Oct 2019 23:58:08 +0900
Subject: [PATCH 09/25] fix: Using pin_project!

---
 src/stream/stream/flatten.rs | 85 +++++++++++++++++++-----------------
 1 file changed, 44 insertions(+), 41 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index b9700876..d44fab0e 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -1,18 +1,22 @@
 use std::pin::Pin;
+use pin_project_lite::pin_project;
 
 use crate::prelude::*;
 use crate::stream::stream::map::Map;
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 
-/// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
-/// documentation for more.
-///
-/// [`flat_map`]: trait.Stream.html#method.flat_map
-/// [`Stream`]: trait.Stream.html
-#[allow(missing_debug_implementations)]
-pub struct FlatMap<S: Stream, U: IntoStream, F> {
-    inner: FlattenCompat<Map<S, F, S::Item, U>, U>,
+pin_project! {
+    /// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
+    /// documentation for more.
+    ///
+    /// [`flat_map`]: trait.Stream.html#method.flat_map
+    /// [`Stream`]: trait.Stream.html
+    #[allow(missing_debug_implementations)]
+    pub struct FlatMap<S: Stream, U: IntoStream, F> {
+        #[pin]
+        inner: FlattenCompat<Map<S, F, S::Item, U>, U>,
+    }
 }
 
 impl<S, U, F> FlatMap<S, U, F>
@@ -21,7 +25,6 @@ where
     U: IntoStream,
     F: FnMut(S::Item) -> U,
 {
-    pin_utils::unsafe_pinned!(inner: FlattenCompat<Map<S, F, S::Item, U>, U>);
 
     pub fn new(stream: S, f: F) -> FlatMap<S, U, F> {
         FlatMap {
@@ -33,33 +36,33 @@ where
 impl<S, U, F> Stream for FlatMap<S, U, F>
 where
     S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
-    S::Item: std::marker::Unpin,
     U: Stream + std::marker::Unpin,
-    F: FnMut(S::Item) -> U + std::marker::Unpin,
+    F: FnMut(S::Item) -> U,
 {
     type Item = U::Item;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.as_mut().inner().poll_next(cx)
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().inner.poll_next(cx)
     }
 }
 
-/// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
-/// documentation for more.
-///
-/// [`flatten`]: trait.Stream.html#method.flatten
-/// [`Stream`]: trait.Stream.html
-#[allow(missing_debug_implementations)]
-pub struct Flatten<S: Stream>
-where
-    S::Item: IntoStream,
-{
-    inner: FlattenCompat<S, <S::Item as IntoStream>::IntoStream>,
+pin_project!{
+    /// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
+    /// documentation for more.
+    ///
+    /// [`flatten`]: trait.Stream.html#method.flatten
+    /// [`Stream`]: trait.Stream.html
+    #[allow(missing_debug_implementations)]
+    pub struct Flatten<S: Stream>
+    where
+        S::Item: IntoStream,
+    {
+        #[pin]
+        inner: FlattenCompat<S, <S::Item as IntoStream>::IntoStream>,
+    }
 }
 
 impl<S: Stream<Item: IntoStream>> Flatten<S> {
-    pin_utils::unsafe_pinned!(inner: FlattenCompat<S, <S::Item as IntoStream>::IntoStream>);
-
     pub fn new(stream: S) -> Flatten<S> {
         Flatten { inner: FlattenCompat::new(stream) }
     }
@@ -72,24 +75,23 @@ where
 {
     type Item = U::Item;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.as_mut().inner().poll_next(cx)
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().inner.poll_next(cx)
     }
 }
 
 
-/// Real logic of both `Flatten` and `FlatMap` which simply delegate to
-/// this type.
-#[derive(Clone, Debug)]
-struct FlattenCompat<S, U> {
-    stream: S,
-    frontiter: Option<U>,
+pin_project! {
+    /// Real logic of both `Flatten` and `FlatMap` which simply delegate to
+    /// this type.
+    #[derive(Clone, Debug)]
+    struct FlattenCompat<S, U> {
+        stream: S,
+        frontiter: Option<U>,
+    }
 }
 
 impl<S, U> FlattenCompat<S, U> {
-    pin_utils::unsafe_unpinned!(stream: S);
-    pin_utils::unsafe_unpinned!(frontiter: Option<U>);
-
     /// Adapts an iterator by flattening it, for use in `flatten()` and `flat_map()`.
     pub fn new(stream: S) -> FlattenCompat<S, U> {
         FlattenCompat {
@@ -106,17 +108,18 @@ where
 {
     type Item = U::Item;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
         loop {
-            if let Some(ref mut inner) = self.as_mut().frontiter() {
+            if let Some(inner) = this.frontiter {
                 if let item @ Some(_) = futures_core::ready!(Pin::new(inner).poll_next(cx)) {
                     return Poll::Ready(item);
                 }
             }
 
-            match futures_core::ready!(Pin::new(&mut self.stream).poll_next(cx)) {
+            match futures_core::ready!(Pin::new(&mut this.stream).poll_next(cx)) {
                 None => return Poll::Ready(None),
-                Some(inner) => *self.as_mut().frontiter() = Some(inner.into_stream()),
+                Some(inner) => *this.frontiter = Some(inner.into_stream()),
             }
         }
     }

From 00e7e58bf3194c8a9e0b912061cdf0741a1834be Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 00:26:53 +0900
Subject: [PATCH 10/25] fix type def

---
 src/stream/stream/flatten.rs | 23 ++++++++++-------------
 src/stream/stream/mod.rs     | 12 ++++--------
 2 files changed, 14 insertions(+), 21 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index d44fab0e..bc1bee34 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -13,27 +13,27 @@ pin_project! {
     /// [`flat_map`]: trait.Stream.html#method.flat_map
     /// [`Stream`]: trait.Stream.html
     #[allow(missing_debug_implementations)]
-    pub struct FlatMap<S: Stream, U: IntoStream, F> {
+    pub struct FlatMap<S, U, T, F> {
         #[pin]
-        inner: FlattenCompat<Map<S, F, S::Item, U>, U>,
+        inner: FlattenCompat<Map<S, F, T, U>, U>,
     }
 }
 
-impl<S, U, F> FlatMap<S, U, F>
+impl<S, U, F> FlatMap<S, U, S::Item, F>
 where
     S: Stream,
     U: IntoStream,
     F: FnMut(S::Item) -> U,
 {
 
-    pub fn new(stream: S, f: F) -> FlatMap<S, U, F> {
+    pub fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
         FlatMap {
             inner: FlattenCompat::new(stream.map(f)),
         }
     }
 }
 
-impl<S, U, F> Stream for FlatMap<S, U, F>
+impl<S, U, F> Stream for FlatMap<S, U, S::Item, F>
 where
     S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
     U: Stream + std::marker::Unpin,
@@ -53,22 +53,19 @@ pin_project!{
     /// [`flatten`]: trait.Stream.html#method.flatten
     /// [`Stream`]: trait.Stream.html
     #[allow(missing_debug_implementations)]
-    pub struct Flatten<S: Stream>
-    where
-        S::Item: IntoStream,
-    {
+    pub struct Flatten<S, U> {
         #[pin]
-        inner: FlattenCompat<S, <S::Item as IntoStream>::IntoStream>,
+        inner: FlattenCompat<S, U>
     }
 }
 
-impl<S: Stream<Item: IntoStream>> Flatten<S> {
-    pub fn new(stream: S) -> Flatten<S> {
+impl<S: Stream<Item: IntoStream>> Flatten<S, S::Item> {
+    pub fn new(stream: S) -> Flatten<S, S::Item> {
         Flatten { inner: FlattenCompat::new(stream) }
     }
 }
 
-impl<S, U> Stream for Flatten<S>
+impl<S, U> Stream for Flatten<S, <S::Item as IntoStream>::IntoStream>
 where
     S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
     U: Stream + std::marker::Unpin,
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index 8ad8cdc0..c994a8ff 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -30,6 +30,7 @@ mod filter;
 mod filter_map;
 mod find;
 mod find_map;
+mod flatten;
 mod fold;
 mod for_each;
 mod fuse;
@@ -77,6 +78,7 @@ use try_for_each::TryForEeachFuture;
 
 pub use chain::Chain;
 pub use filter::Filter;
+pub use flatten::{FlatMap, Flatten};
 pub use fuse::Fuse;
 pub use inspect::Inspect;
 pub use map::Map;
@@ -99,10 +101,8 @@ cfg_unstable! {
     use crate::stream::into_stream::IntoStream;
 
     pub use merge::Merge;
-    pub use flatten::{FlatMap, Flatten};
 
     mod merge;
-    mod flatten;
 }
 
 extension_trait! {
@@ -588,9 +588,7 @@ extension_trait! {
             # });
             ```
         "#]
-        #[cfg(feature = "unstable")]
-        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-        fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F>
+        fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, Self::Item, F>
             where
                 Self: Sized,
                 U: IntoStream,
@@ -622,9 +620,7 @@ extension_trait! {
 
             # });
         "#]
-        #[cfg(feature = "unstable")]
-        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
-        fn flatten(self) -> Flatten<Self>
+        fn flatten(self) -> Flatten<Self, Self::Item>
         where
             Self: Sized,
             Self::Item: IntoStream,

From 001368d3dfa4f9f0962c52c357a776e9e66e8780 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 00:29:10 +0900
Subject: [PATCH 11/25] $cargo fmt

---
 src/stream/stream/flatten.rs | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index bc1bee34..fc3592ee 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -1,5 +1,5 @@
-use std::pin::Pin;
 use pin_project_lite::pin_project;
+use std::pin::Pin;
 
 use crate::prelude::*;
 use crate::stream::stream::map::Map;
@@ -25,7 +25,6 @@ where
     U: IntoStream,
     F: FnMut(S::Item) -> U,
 {
-
     pub fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
         FlatMap {
             inner: FlattenCompat::new(stream.map(f)),
@@ -46,7 +45,7 @@ where
     }
 }
 
-pin_project!{
+pin_project! {
     /// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
     /// documentation for more.
     ///
@@ -61,7 +60,9 @@ pin_project!{
 
 impl<S: Stream<Item: IntoStream>> Flatten<S, S::Item> {
     pub fn new(stream: S) -> Flatten<S, S::Item> {
-        Flatten { inner: FlattenCompat::new(stream) }
+        Flatten {
+            inner: FlattenCompat::new(stream),
+        }
     }
 }
 
@@ -77,7 +78,6 @@ where
     }
 }
 
-
 pin_project! {
     /// Real logic of both `Flatten` and `FlatMap` which simply delegate to
     /// this type.

From 0c5abee284eae2136ab1b795d086e69761b8a913 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 00:36:04 +0900
Subject: [PATCH 12/25] to unstable stream::flat_map, stream::flatten

---
 src/stream/stream/mod.rs | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index c994a8ff..596dc419 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -30,7 +30,6 @@ mod filter;
 mod filter_map;
 mod find;
 mod find_map;
-mod flatten;
 mod fold;
 mod for_each;
 mod fuse;
@@ -78,7 +77,6 @@ use try_for_each::TryForEeachFuture;
 
 pub use chain::Chain;
 pub use filter::Filter;
-pub use flatten::{FlatMap, Flatten};
 pub use fuse::Fuse;
 pub use inspect::Inspect;
 pub use map::Map;
@@ -101,8 +99,10 @@ cfg_unstable! {
     use crate::stream::into_stream::IntoStream;
 
     pub use merge::Merge;
+    pub use flatten::{FlatMap, Flatten};
 
     mod merge;
+    mod flatten;
 }
 
 extension_trait! {
@@ -588,6 +588,8 @@ extension_trait! {
             # });
             ```
         "#]
+        #[cfg(feature = "unstable")]
+        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, Self::Item, F>
             where
                 Self: Sized,
@@ -620,6 +622,8 @@ extension_trait! {
 
             # });
         "#]
+        #[cfg(feature = "unstable")]
+        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
         fn flatten(self) -> Flatten<Self, Self::Item>
         where
             Self: Sized,

From b66ffa670eb91356f49cc0ceb56a3fd95df3489a Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 00:42:27 +0900
Subject: [PATCH 13/25] update recursion_limit

---
 src/lib.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/lib.rs b/src/lib.rs
index 7f04e082..1e33c8fc 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -46,7 +46,7 @@
 #![doc(test(attr(deny(rust_2018_idioms, warnings))))]
 #![doc(test(attr(allow(unused_extern_crates, unused_variables))))]
 #![doc(html_logo_url = "https://async.rs/images/logo--hero.svg")]
-#![recursion_limit = "1024"]
+#![recursion_limit = "2048"]
 #![feature(associated_type_bounds)]
 
 #[macro_use]

From 7ce721f562e17db4eb7953ec5612c5edff8675d6 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 01:42:19 +0900
Subject: [PATCH 14/25] Update src/lib.rs

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/lib.rs | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/lib.rs b/src/lib.rs
index 1e33c8fc..22481b53 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -47,7 +47,6 @@
 #![doc(test(attr(allow(unused_extern_crates, unused_variables))))]
 #![doc(html_logo_url = "https://async.rs/images/logo--hero.svg")]
 #![recursion_limit = "2048"]
-#![feature(associated_type_bounds)]
 
 #[macro_use]
 mod utils;

From b7b5df13aa7843f42fe01c31fa8758727646dbd8 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 01:42:39 +0900
Subject: [PATCH 15/25] Update src/stream/stream/flatten.rs

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index fc3592ee..56277330 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -108,7 +108,7 @@ where
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();
         loop {
-            if let Some(inner) = this.frontiter {
+            if let Some(inner) = this.frontiter.as_mut().as_pin_mut() {
                 if let item @ Some(_) = futures_core::ready!(Pin::new(inner).poll_next(cx)) {
                     return Poll::Ready(item);
                 }

From 6168952d6f262ee04d4fc06533db2665e207edb3 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 01:42:57 +0900
Subject: [PATCH 16/25] Update src/stream/stream/flatten.rs

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 56277330..b244f035 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -109,7 +109,7 @@ where
         let mut this = self.project();
         loop {
             if let Some(inner) = this.frontiter.as_mut().as_pin_mut() {
-                if let item @ Some(_) = futures_core::ready!(Pin::new(inner).poll_next(cx)) {
+                if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
                     return Poll::Ready(item);
                 }
             }

From bf3508ffb254db07a187293e8cdef7df651bce6e Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 01:43:07 +0900
Subject: [PATCH 17/25] Update src/stream/stream/flatten.rs

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index b244f035..1b21746e 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -114,7 +114,7 @@ where
                 }
             }
 
-            match futures_core::ready!(Pin::new(&mut this.stream).poll_next(cx)) {
+            match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
                 None => return Poll::Ready(None),
                 Some(inner) => *this.frontiter = Some(inner.into_stream()),
             }

From 8932cecec75b78058de240d2a4c72d90b07759b3 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 01:43:14 +0900
Subject: [PATCH 18/25] Update src/stream/stream/flatten.rs

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 1b21746e..59330855 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -116,7 +116,7 @@ where
 
             match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
                 None => return Poll::Ready(None),
-                Some(inner) => *this.frontiter = Some(inner.into_stream()),
+                Some(inner) => this.frontiter.set(Some(inner.into_stream())),
             }
         }
     }

From 61b7a09c70d78329131dedc528baea5d6cda7942 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Sat, 26 Oct 2019 01:44:48 +0900
Subject: [PATCH 19/25] Fix type declaration

---
 src/stream/stream/flatten.rs | 23 ++++++++++++++++-------
 1 file changed, 16 insertions(+), 7 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 59330855..3bf6fe84 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -34,8 +34,9 @@ where
 
 impl<S, U, F> Stream for FlatMap<S, U, S::Item, F>
 where
-    S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
-    U: Stream + std::marker::Unpin,
+    S: Stream,
+    S::Item: IntoStream<IntoStream = U, Item = U::Item>,
+    U: Stream,
     F: FnMut(S::Item) -> U,
 {
     type Item = U::Item;
@@ -58,7 +59,11 @@ pin_project! {
     }
 }
 
-impl<S: Stream<Item: IntoStream>> Flatten<S, S::Item> {
+impl<S> Flatten<S, S::Item>
+where
+    S: Stream,
+    S::Item: IntoStream,
+{
     pub fn new(stream: S) -> Flatten<S, S::Item> {
         Flatten {
             inner: FlattenCompat::new(stream),
@@ -68,8 +73,9 @@ impl<S: Stream<Item: IntoStream>> Flatten<S, S::Item> {
 
 impl<S, U> Stream for Flatten<S, <S::Item as IntoStream>::IntoStream>
 where
-    S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
-    U: Stream + std::marker::Unpin,
+     S: Stream,
+     S::Item: IntoStream<IntoStream = U, Item = U::Item>,
+     U: Stream,
 {
     type Item = U::Item;
 
@@ -83,7 +89,9 @@ pin_project! {
     /// this type.
     #[derive(Clone, Debug)]
     struct FlattenCompat<S, U> {
+        #[pin]
         stream: S,
+        #[pin]
         frontiter: Option<U>,
     }
 }
@@ -100,8 +108,9 @@ impl<S, U> FlattenCompat<S, U> {
 
 impl<S, U> Stream for FlattenCompat<S, U>
 where
-    S: Stream<Item: IntoStream<IntoStream = U, Item = U::Item>> + std::marker::Unpin,
-    U: Stream + std::marker::Unpin,
+    S: Stream,
+    S::Item: IntoStream<IntoStream = U, Item = U::Item>,
+    U: Stream,
 {
     type Item = U::Item;
 

From 13a08b0d54f5c983883ce3efca60278d38a8b6d7 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sun, 27 Oct 2019 12:35:14 +0900
Subject: [PATCH 20/25] Narrow the disclosure range of FlatMap::new

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 3bf6fe84..3efb3523 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -25,7 +25,7 @@ where
     U: IntoStream,
     F: FnMut(S::Item) -> U,
 {
-    pub fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
+    pub(super) fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
         FlatMap {
             inner: FlattenCompat::new(stream.map(f)),
         }

From 37f14b0195e63838f3e4485ba14bb8e54cc59789 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sun, 27 Oct 2019 12:35:32 +0900
Subject: [PATCH 21/25] Narrow the disclosure range of Flatten::new

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 3efb3523..d81e79fb 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -64,7 +64,7 @@ where
     S: Stream,
     S::Item: IntoStream,
 {
-    pub fn new(stream: S) -> Flatten<S, S::Item> {
+    pub(super) fn new(stream: S) -> Flatten<S, S::Item> {
         Flatten {
             inner: FlattenCompat::new(stream),
         }

From a42ae2f3d925499ea8732a017baf7f68f146cf27 Mon Sep 17 00:00:00 2001
From: nasa <htilcs1115@gmail.com>
Date: Sun, 27 Oct 2019 12:35:51 +0900
Subject: [PATCH 22/25] Narrow the disclosure range of FlattenCompat::new

Co-Authored-By: Taiki Endo <te316e89@gmail.com>
---
 src/stream/stream/flatten.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index d81e79fb..2533f379 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -98,7 +98,7 @@ pin_project! {
 
 impl<S, U> FlattenCompat<S, U> {
     /// Adapts an iterator by flattening it, for use in `flatten()` and `flat_map()`.
-    pub fn new(stream: S) -> FlattenCompat<S, U> {
+    fn new(stream: S) -> FlattenCompat<S, U> {
         FlattenCompat {
             stream,
             frontiter: None,

From 688976203e0ff6f564647f2f21b89fea7fafae88 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Tue, 29 Oct 2019 09:52:50 +0900
Subject: [PATCH 23/25] fix: Split FlattenCompat logic to Flatten and FlatMap

---
 src/stream/stream/flat_map.rs |  62 +++++++++++++++++++
 src/stream/stream/flatten.rs  | 112 +++-------------------------------
 src/stream/stream/mod.rs      |   4 +-
 3 files changed, 72 insertions(+), 106 deletions(-)
 create mode 100644 src/stream/stream/flat_map.rs

diff --git a/src/stream/stream/flat_map.rs b/src/stream/stream/flat_map.rs
new file mode 100644
index 00000000..ed3268ea
--- /dev/null
+++ b/src/stream/stream/flat_map.rs
@@ -0,0 +1,62 @@
+use pin_project_lite::pin_project;
+use std::pin::Pin;
+
+use crate::prelude::*;
+use crate::stream::stream::map::Map;
+use crate::stream::{IntoStream, Stream};
+use crate::task::{Context, Poll};
+
+pin_project! {
+    /// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
+    /// documentation for more.
+    ///
+    /// [`flat_map`]: trait.Stream.html#method.flat_map
+    /// [`Stream`]: trait.Stream.html
+    #[allow(missing_debug_implementations)]
+    pub struct FlatMap<S, U, T, F> {
+        #[pin]
+        stream: Map<S, F, T, U>,
+        #[pin]
+        inner_stream: Option<U>,
+    }
+}
+
+impl<S, U, F> FlatMap<S, U, S::Item, F>
+where
+    S: Stream,
+    U: IntoStream,
+    F: FnMut(S::Item) -> U,
+{
+    pub(super) fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
+        FlatMap {
+            stream: stream.map(f),
+            inner_stream: None,
+        }
+    }
+}
+
+impl<S, U, F> Stream for FlatMap<S, U, S::Item, F>
+where
+    S: Stream,
+    S::Item: IntoStream<IntoStream = U, Item = U::Item>,
+    U: Stream,
+    F: FnMut(S::Item) -> U,
+{
+    type Item = U::Item;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+        loop {
+            if let Some(inner) = this.inner_stream.as_mut().as_pin_mut() {
+                if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
+                    return Poll::Ready(item);
+                }
+            }
+
+            match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
+                None => return Poll::Ready(None),
+                Some(inner) => this.inner_stream.set(Some(inner.into_stream())),
+            }
+        }
+    }
+}
diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 2533f379..2ea0673e 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -6,46 +6,6 @@ use crate::stream::stream::map::Map;
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 
-pin_project! {
-    /// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
-    /// documentation for more.
-    ///
-    /// [`flat_map`]: trait.Stream.html#method.flat_map
-    /// [`Stream`]: trait.Stream.html
-    #[allow(missing_debug_implementations)]
-    pub struct FlatMap<S, U, T, F> {
-        #[pin]
-        inner: FlattenCompat<Map<S, F, T, U>, U>,
-    }
-}
-
-impl<S, U, F> FlatMap<S, U, S::Item, F>
-where
-    S: Stream,
-    U: IntoStream,
-    F: FnMut(S::Item) -> U,
-{
-    pub(super) fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
-        FlatMap {
-            inner: FlattenCompat::new(stream.map(f)),
-        }
-    }
-}
-
-impl<S, U, F> Stream for FlatMap<S, U, S::Item, F>
-where
-    S: Stream,
-    S::Item: IntoStream<IntoStream = U, Item = U::Item>,
-    U: Stream,
-    F: FnMut(S::Item) -> U,
-{
-    type Item = U::Item;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project().inner.poll_next(cx)
-    }
-}
-
 pin_project! {
     /// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
     /// documentation for more.
@@ -55,7 +15,9 @@ pin_project! {
     #[allow(missing_debug_implementations)]
     pub struct Flatten<S, U> {
         #[pin]
-        inner: FlattenCompat<S, U>
+        stream: S,
+        #[pin]
+        inner_stream: Option<U>,
     }
 }
 
@@ -66,47 +28,13 @@ where
 {
     pub(super) fn new(stream: S) -> Flatten<S, S::Item> {
         Flatten {
-            inner: FlattenCompat::new(stream),
+            stream,
+            inner_stream: None,
         }
     }
 }
 
 impl<S, U> Stream for Flatten<S, <S::Item as IntoStream>::IntoStream>
-where
-     S: Stream,
-     S::Item: IntoStream<IntoStream = U, Item = U::Item>,
-     U: Stream,
-{
-    type Item = U::Item;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project().inner.poll_next(cx)
-    }
-}
-
-pin_project! {
-    /// Real logic of both `Flatten` and `FlatMap` which simply delegate to
-    /// this type.
-    #[derive(Clone, Debug)]
-    struct FlattenCompat<S, U> {
-        #[pin]
-        stream: S,
-        #[pin]
-        frontiter: Option<U>,
-    }
-}
-
-impl<S, U> FlattenCompat<S, U> {
-    /// Adapts an iterator by flattening it, for use in `flatten()` and `flat_map()`.
-    fn new(stream: S) -> FlattenCompat<S, U> {
-        FlattenCompat {
-            stream,
-            frontiter: None,
-        }
-    }
-}
-
-impl<S, U> Stream for FlattenCompat<S, U>
 where
     S: Stream,
     S::Item: IntoStream<IntoStream = U, Item = U::Item>,
@@ -117,7 +45,7 @@ where
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();
         loop {
-            if let Some(inner) = this.frontiter.as_mut().as_pin_mut() {
+            if let Some(inner) = this.inner_stream.as_mut().as_pin_mut() {
                 if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
                     return Poll::Ready(item);
                 }
@@ -125,34 +53,8 @@ where
 
             match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
                 None => return Poll::Ready(None),
-                Some(inner) => this.frontiter.set(Some(inner.into_stream())),
+                Some(inner) => this.inner_stream.set(Some(inner.into_stream())),
             }
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::FlattenCompat;
-
-    use crate::prelude::*;
-    use crate::task;
-
-    use std::collections::VecDeque;
-
-    #[test]
-    fn test_poll_next() -> std::io::Result<()> {
-        let inner1: VecDeque<u8> = vec![1, 2, 3].into_iter().collect();
-        let inner2: VecDeque<u8> = vec![4, 5, 6].into_iter().collect();
-
-        let s: VecDeque<_> = vec![inner1, inner2].into_iter().collect();
-
-        task::block_on(async move {
-            let flat = FlattenCompat::new(s);
-            let v: Vec<u8> = flat.collect().await;
-
-            assert_eq!(v, vec![1, 2, 3, 4, 5, 6]);
-            Ok(())
-        })
-    }
-}
diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs
index 596dc419..29e68fc4 100644
--- a/src/stream/stream/mod.rs
+++ b/src/stream/stream/mod.rs
@@ -99,10 +99,12 @@ cfg_unstable! {
     use crate::stream::into_stream::IntoStream;
 
     pub use merge::Merge;
-    pub use flatten::{FlatMap, Flatten};
+    pub use flatten::Flatten;
+    pub use flat_map::FlatMap;
 
     mod merge;
     mod flatten;
+    mod flat_map;
 }
 
 extension_trait! {

From ae7adf2c366e388a59ef3417dc7726892dfcf14c Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Tue, 29 Oct 2019 10:01:41 +0900
Subject: [PATCH 24/25] fix: Remove unused import

---
 src/stream/stream/flatten.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index 2ea0673e..f0049864 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -1,8 +1,8 @@
 use pin_project_lite::pin_project;
 use std::pin::Pin;
 
-use crate::prelude::*;
-use crate::stream::stream::map::Map;
+
+
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};
 

From 1554b0440743a7551ed3c23a88586a5832dca592 Mon Sep 17 00:00:00 2001
From: k-nasa <htilcs1115@gmail.com>
Date: Tue, 29 Oct 2019 10:12:22 +0900
Subject: [PATCH 25/25] $cargo fmt

---
 src/stream/stream/flatten.rs | 2 --
 1 file changed, 2 deletions(-)

diff --git a/src/stream/stream/flatten.rs b/src/stream/stream/flatten.rs
index f0049864..5e791cda 100644
--- a/src/stream/stream/flatten.rs
+++ b/src/stream/stream/flatten.rs
@@ -1,8 +1,6 @@
 use pin_project_lite::pin_project;
 use std::pin::Pin;
 
-
-
 use crate::stream::{IntoStream, Stream};
 use crate::task::{Context, Poll};