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

[muhamadazmy]

[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.

[muhamadazmy]

Use case

use std::time::Duration;

use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<u64>(20);

    let mut receiver = ReceiverStream::new(rx);

    let chunked = (&mut receiver).chunks_timeout(3, Duration::from_secs(3));
    tokio::pin!(chunked);

    tx.send(10).await.unwrap();
    tx.send(20).await.unwrap();

    tokio::select! {
        Some(batch) = chunked.next() => {
            println!("Got: {:?}", batch);
        }
        _ = tokio::time::sleep(Duration::from_secs(1)) => {
            // another condition
        }
    }

    let reminder = chunked.into_reminder();

    // use `receiver` here
}

[ADD-SP]

There is a similar interface in the standard library: std::slice::ChunksExactMut:into_remainder. This might be a better interface for your requirements. What do you think?

[muhamadazmy]

Thank you @ADD-SP for your review. The original plan was to completely consume the stream indeed and just return the reminder. This however won't work with the Pinned ChunksTimeout stream which is required when polling next() in a select statement.

This means an into_reminder won't actually be useful because the only way for this to be useful is if the ChunksTimeout stream is not pinned. And in that case, the reminder will always guaranteed to be empty since there is no way to interrupt a next().await call (unless you are select statement which requires the stream to be pinned)

I will still rename the method to reminder instead of remaining to match the std interface.

[ADD-SP]

Suggested change

	pub fn reminder(mut self: Pin<&mut Self>) -> Vec<S::Item> {
	pub fn into_remainder(mut self: Pin<&mut Self>) -> Vec<S::Item> {

Does this make more sense?

[ADD-SP]

Suggested change

	/// Drains the buffered items, returning them without waiting for the timeout or capacity limit.
	/// Consumes the [`ChunksTimeout`] and then returns all buffered items.

[muhamadazmy]

Thank you so much @ADD-SP for your review. I applied all your comments. Let me know if you have more comments.