tokio_stream::ReceiverStream and tokio_stream::StreamExt::skip() together behave in an unexpected way

[Erik-Sovereign]

I'm not sure if this is unexpected or if my thinking is bad :), but let me try to explain my issue with this example:

I post 10 backups. Each backup sends a message in a channel, when the backup is done.

let response = test_setup.post(backups_url).await.send().await.unwrap();
assert_eq!(response.status(), StatusCode::OK);

I post an eleventh backup, which posts a different message, because only 10 backups are allowed:

let response = test_setup.post(backups_url).await.send().await.unwrap();
assert_eq!(response.status(), StatusCode::CONFLICT);

I check the eleventh message with a combination of mpsc::ReceiverStream and skip(), both of tokio_stream:

if let Some(params) = receiver_stream
	.skip(10)
	.next()
	.await
{
	assert_eq!(...);
};

Now comes the problem. After this code, I execute another post request, that wants to send to the receiver, but it is closed, because, I think, skip() consumes self and makes it go out of scope after the code beforehand.

My expectation was, that if ReceiverStream's purpose is to make a mpsc::channel into an useful Stream, that methods on StreamExt work on it, without consuming a single consumer, since this is a quite drastic thing to do with a mpsc.

I can fix the problem by putting the code that skip()'s at the end, after all POST-requests, and with it after all sends to the channel. But this makes the code unnice.

[verysleepylemon]

The issue is that skip() consumes the stream by value, so after calling skip() the original ReceiverStream is moved. The fix is to call .by_ref() before skip() so that the skip adapter borrows the stream rather than owning it:

use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = tokio::sync::mpsc::channel(16);
let mut stream = ReceiverStream::new(rx);

// Skip the first 2 items without consuming `stream`
stream.by_ref().skip(2).collect::<Vec<_>>().await;

// `stream` is still alive here — you can keep polling it
while let Some(item) = stream.next().await {
    println!({item});
}

StreamExt::by_ref() returns a mutable borrow of self that implements Stream, so all the combinator adapters (including skip) operate on the reference instead of taking ownership.

If you need to skip lazily without collecting, wrap in a pinned variable first:

let mut skipping = stream.by_ref().skip(2);
// drive it until skip is satisfied
while let Some(_) = skipping.next().await {}
drop(skipping);         // release the borrow
// now `stream` is usable again

This is the same pattern used with Iterator::by_ref() in sync code.

[verysleepylemon]

The issue is that skip() consumes the stream by value, so after calling skip() the original ReceiverStream is moved. The fix is .by_ref() before skip() so the adapter borrows the stream rather than owning it.

use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = tokio::sync::mpsc::channel(16);
let mut stream = ReceiverStream::new(rx);

// Skip the first 2 items WITHOUT consuming stream
stream.by_ref().skip(2).collect::<Vec<_>>().await;

// stream is still alive here
while let Some(item) = stream.next().await {
    println!("{item}");
}

StreamExt::by_ref() returns &mut Self which implements Stream, so all adapters operate on the borrow instead of taking ownership. Same pattern as Iterator::by_ref() in sync Rust.

If you need to drain lazily without collecting:

{
    let mut skipping = stream.by_ref().skip(2);
    while let Some(_) = skipping.next().await {}
} // borrow released here
while let Some(item) = stream.next().await { /* ... */ }

[verysleepylemon]

The issue is that skip() consumes the stream by value, so after calling skip() the original ReceiverStream is moved. The fix is to call .by_ref() before skip() so the adapter borrows the stream rather than owning it.

use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = tokio::sync::mpsc::channel(16);
let mut stream = ReceiverStream::new(rx);

// Skip the first 2 items WITHOUT consuming `stream`
stream.by_ref().skip(2).collect::<Vec<_>>().await;

// stream is still alive here
while let Some(item) = stream.next().await {
    println!("{item}");
}

StreamExt::by_ref() returns &mut Self which itself implements Stream, so all combinator adapters operate on the borrow instead of taking ownership. This is the same pattern as Iterator::by_ref() in sync Rust.

If you need to drain lazily without collecting:

{
    let mut skipping = stream.by_ref().skip(2);
    while let Some(_) = skipping.next().await {}
} // borrow released here — stream is usable again
while let Some(item) = stream.next().await { /* ... */ }

[verysleepylemon]

The issue is that skip() consumes the stream by value, so after calling skip() the original ReceiverStream is moved. The fix is to call .by_ref() before skip() so the adapter borrows the stream rather than owning it.

use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = tokio::sync::mpsc::channel(16);
let mut stream = ReceiverStream::new(rx);

// Skip the first 2 items WITHOUT consuming `stream`
stream.by_ref().skip(2).collect::<Vec<_>>().await;

// stream is still alive here
while let Some(item) = stream.next().await {
    println!("{item}");
}

StreamExt::by_ref() returns &mut Self which itself implements Stream, so all combinator adapters operate on the borrow instead of taking ownership. This is the same pattern as Iterator::by_ref() in sync Rust.

If you need to drain lazily without collecting:

{
    let mut skipping = stream.by_ref().skip(2);
    while let Some(_) = skipping.next().await {}
} // borrow released here — stream is usable again
while let Some(item) = stream.next().await { /* ... */ }

[Erik-Sovereign]

AI slop?!
There is no by_ref on stream

[martin-g]

There are as_ref() and as_mut() -

tokio/tokio-stream/src/wrappers/mpsc_bounded.rs

Lines 92 to 98 in ad8c59a

	fn as_ref(&self) -> &Receiver<T> {
	&self.inner
	}
	}
	impl<T> AsMut<Receiver<T>> for ReceiverStream<T> {
	fn as_mut(&mut self) -> &mut Receiver<T> {

[Erik-Sovereign]

There are as_ref() and as_mut() -

tokio/tokio-stream/src/wrappers/mpsc_bounded.rs

Lines 92 to 98 in ad8c59a

 fn as_ref(&self) -> &Receiver<T> { 
     &self.inner 
 } 

}

impl AsMut<Receiver> for ReceiverStream {
fn as_mut(&mut self) -> &mut Receiver {

Thank you, I know. And also tried that before writing this post.
But skip can't handle as_ref :)

[Darksonn]

Hi Erik,

A reference to a stream is a stream. This should work:

(&mut receiver_stream).skip(10).next()

[Erik-Sovereign]

Hey Alice,
thank you very much.

I have now learned some things:

1. as_mut returns a mutalble & to the Receiver only, not the ReceiverStream
2. That you can write (&mut your-variable) as a shorthand.
3. That skip just creates a new stream where the first two items where skipped. So I could have also written:

    let (tx, rx) = tokio::sync::mpsc::channel::<u16>(16);
    let stream = ReceiverStream::new(rx);

    // Skip the first 2 items without consuming `stream`
    let mut stream = stream.skip(2);

    // `stream` is still alive here — you can keep polling it
    while let Some(item) = stream.next().await {
        println!("  +: {:?}", item);
    }