CODEWITHSUNDEEP
rustrust-tokio
haolun
haolun

Reputation: 334

Is there a way to make StreamExt::next non blocking (fail fast) if the stream is empty (need to wait for the next element)?

Currently I am doing something like this

use tokio::time::timeout;

while let Ok(option_element) = timeout(Duration::from_nanos(1), stream.next()).await {
...
}

to drain the items already in the rx buffer of the stream. I don't want to wait for the next element that has not been received.

I think the timeout would slow down the while loop.

I am wondering that is there a better way to do this without the use of the timeout? Possibly like this https://github.com/async-rs/async-std/issues/579 but for the streams in futures/tokio.

Upvotes: 2

Views: 2297

Answers (1)

Alice Ryhl
Alice Ryhl

Reputation: 4219

The direct answer to your question is to use the FutureExt::now_or_never method from the futures crate as in stream.next().now_or_never().

However it is important to avoid writing a busy loop that waits on several things by calling now_or_never on each thing in a loop. This is bad because it is blocking the thread, and you should prefer a different solution such as tokio::select! to wait for multiple things. For the special case of this where you are constantly checking whether the task should shut down, see this other question instead.

On the other hand, an example where using now_or_never is perfectly fine is when you want to empty a queue for the items available now so you can batch process them in some manner. This is fine because the now_or_never loop will stop spinning as soon as it has emptied the queue.

Beware that if the stream is empty, then now_or_never will succeed because next() immediately returns None in this case.

Upvotes: 3

Related Questions