stream: work around the rustc bug in `StreamExt::collect` (#7754)

diff --git a/tokio-stream/src/stream_ext.rs b/tokio-stream/src/stream_ext.rs
index 7e915d9909ee9402f09a8642e8939f58d9ee8ee9..fe589869215de3978799d513a7229db926652968 100644 (file)
--- a/tokio-stream/src/stream_ext.rs
+++ b/tokio-stream/src/stream_ext.rs
@@ -916,7 +916,7 @@ pub trait StreamExt: Stream {
     /// assert_eq!(Err("no"), values);
     /// # }
     /// ```
-    fn collect<T>(self) -> Collect<Self, T>
+    fn collect<T>(self) -> Collect<Self, T, T::InternalCollection>
     where
         T: FromStream<Self::Item>,
         Self: Sized,

diff --git a/tokio-stream/src/stream_ext/collect.rs b/tokio-stream/src/stream_ext/collect.rs
index eb9e2197f1414af484cea86b22dd21d3df4cb83d..8140f0b4e4bc6c1319f87b1094f396d7b7bcbaeb 100644 (file)
--- a/tokio-stream/src/stream_ext/collect.rs
+++ b/tokio-stream/src/stream_ext/collect.rs
@@ -1,7 +1,7 @@
 use crate::Stream;
 
 use core::future::Future;
-use core::marker::PhantomPinned;
+use core::marker::{PhantomData, PhantomPinned};
 use core::mem;
 use core::pin::Pin;
 use core::task::{ready, Context, Poll};
@@ -12,14 +12,12 @@ pin_project! {
     /// Future returned by the [`collect`](super::StreamExt::collect) method.
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     #[derive(Debug)]
-    pub struct Collect<T, U>
-    where
-        T: Stream,
-        U: FromStream<T::Item>,
+    pub struct Collect<T, U, C>
     {
         #[pin]
         stream: T,
-        collection: U::InternalCollection,
+        collection: C,
+        _output: PhantomData<U>,
         // Make this future `!Unpin` for compatibility with async trait methods.
         #[pin]
         _pin: PhantomPinned,
@@ -38,24 +36,25 @@ pin_project! {
 /// enhancements to the Rust language.
 pub trait FromStream<T>: sealed::FromStreamPriv<T> {}
 
-impl<T, U> Collect<T, U>
+impl<T, U> Collect<T, U, U::InternalCollection>
 where
     T: Stream,
     U: FromStream<T::Item>,
 {
-    pub(super) fn new(stream: T) -> Collect<T, U> {
+    pub(super) fn new(stream: T) -> Collect<T, U, U::InternalCollection> {
         let (lower, upper) = stream.size_hint();
         let collection = U::initialize(sealed::Internal, lower, upper);
 
         Collect {
             stream,
             collection,
+            _output: PhantomData,
             _pin: PhantomPinned,
         }
     }
 }
 
-impl<T, U> Future for Collect<T, U>
+impl<T, U> Future for Collect<T, U, U::InternalCollection>
 where
     T: Stream,
     U: FromStream<T::Item>,