[ChunksTimeout] Consumes the stream reminder and return the items immediately

Summary:
When the underlying stream is an exclusive reference (&mut stream), and we need to drop the `ChunksTimout` stream without losing the buffered items.

Author: muhamadazmy

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -33,6 +33,34 @@ impl<S: Stream> ChunksTimeout<S> {
             cap: max_size,
         }
     }
+
+    /// Consumes the [`ChunksTimeout`] and then returns all buffered items.
+    ///
+    /// ```
+    /// use tokio::time::Duration;
+    /// use tokio_stream::{self as stream, StreamExt};
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let chunked = stream::iter([1, 2, 3, 4])
+    ///        .throttle(Duration::from_millis(2))
+    ///        .chunks_timeout(4, Duration::from_millis(8));
+    ///
+    ///     tokio::pin!(chunked);
+    ///
+    ///     // race with another future
+    ///     tokio::select! {
+    ///         Some(_chunk) = chunked.next() => {}
+    ///         _ = tokio::time::sleep(Duration::from_millis(3)) => {}
+    ///     }
+    ///
+    ///     assert_eq!(chunked.into_remainder(), vec![1, 2]);
+    /// }
+    /// ```
+    pub fn into_remainder(mut self: Pin<&mut Self>) -> Vec<S::Item> {
+        let me = self.as_mut().project();
+        std::mem::take(me.items)
+    }
 }
 
 impl<S: Stream> Stream for ChunksTimeout<S> {

tokio-stream/tests/stream_chunks_remainder.rs

@@ -0,0 +1,25 @@
+#![warn(rust_2018_idioms)]
+#![cfg(all(feature = "time", not(target_os = "wasi")))] // Wasi does not support panic recovery
+#![cfg(panic = "unwind")]
+
+use std::error::Error;
+use tokio::time::Duration;
+use tokio_stream::{self as stream, StreamExt};
+
+#[tokio::test]
+async fn stream_chunks_remainder() -> Result<(), Box<dyn Error>> {
+    let inner = stream::iter([1, 2, 3, 4]).throttle(Duration::from_millis(5));
+    tokio::pin!(inner);
+
+    let chunked = (&mut inner).chunks_timeout(4, Duration::from_millis(20));
+    tokio::pin!(chunked);
+
+    tokio::select! {
+        Some(_chunk) = chunked.next() => {}
+        _ = tokio::time::sleep(Duration::from_millis(9)) => {}
+    }
+
+    assert_eq!(chunked.into_remainder(), vec![1, 2]);
+    assert_eq!(inner.next().await, Some(3));
+    Ok(())
+}