try_stream! / stream! with error close stream

[BratSinot]

Greetings!

Why Err close stream? I can't use .filter(..) in that case.

use async_stream::{stream, try_stream};
use futures_util::{pin_mut, Stream, StreamExt};
use std::future::ready;

fn error_or_ok(rcx: usize) -> Result<usize, String> {
    if rcx % 2 == 0 {
        Ok(rcx)
    } else {
        Err(format!("Error{rcx}"))
    }
}

fn return_stream() -> impl Stream<Item = Result<usize, String>> {
    //stream! {
    try_stream! {
        let mut rcx = 0usize;
        loop {
            //yield Ok(error_or_ok(rcx)?); // with stream! closes
            //yield {move || { Ok(error_or_ok(rcx)?) }}(); // with stream! ok
            yield error_or_ok(rcx)?; // with try_stream! closes
            rcx += 1;
        }
    }
    .filter(|data| {
        let ret = if let Err(error) = data {
            eprintln!("Got error: `{error}`.");
            false
        } else {
            true
        };

        ready(ret)
    })
}

#[tokio::main]
async fn main() {
    let stream = return_stream();
    pin_mut!(stream);

    println!("Some: `{:?}`", stream.next().await);
    println!("None: `{:?}`", stream.next().await);
    println!("Some: `{:?}`", stream.next().await);
    println!("None: `{:?}`", stream.next().await);
}

[Darksonn]

If you return from the code in the stream, then the stream will end. The question mark operator will make you return. Thus, if you use the question mark, the stream will end.

If you use stream! and don't use the question mark, then it should work.

[nanne007]

Came across the same problem.
Although the reply of @Darksonn answers the question, but how can I emit an Err value instead of return in try_stream!.
I have to fallback to use stream! with is kind of tedious. I thought try_stream! is meant to solve this.

[gabrielgrover]

Ran into this issue as well. The docs make it seem like a try operator merely emits an Err as a stream item, but I guess it shuts down the stream.