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

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,

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>,