how too stop yield in tokio-stream

[ansjsun]

Hi Developer I have some question in my work.

when sink has end but, scan is yield ing. how to set timeout for yield i like this?

use async_stream::stream;
use tokio_stream::StreamExt;

#[tokio::main]
async fn main() {
    let (tx, rx) = tokio::sync::watch::channel(1);

    let mut scan = Box::pin(stream! {
        println!("--------scan start");
            for i in 0..10 {
                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                println!("scan: {:?}", i);

                tokio::select! {
                    _ = yield i => {
                        println!("recived yield signal");
                    }
                    _ = rx.changed() => {
                        println!("recived end signal");
                    }
                }
            }
        println!("--------scan end")
    });

    let mut sink = Box::pin(stream! {
        println!("--------sink start");
        while let Some(i) = scan.next().await {

            if i>5 {
                break;
            }
            println!("sink: {:?}", i);
            yield i;
        }
        println!("--------sink end")
    });

    while let Some(i) = sink.next().await {}

    tx.send(1);

    tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
}