Is there a convenient way/pattern to create streams that can capture their environment?

[tamasfe]

I'm looking for a way to do the following:

fn get_foo(&self, some_param: &str) -> BoxStream<'_, Foo> {
    let some_other_param = get_other_param();
    query!(
        r#"
        SELECT * FROM "foo"
        WHERE
            "foo" = $1
            AND "bar" = $2
        "#,
        some_param,
        &some_other_param
    )
    .fetch_many(&self.db)
    .map(row_to_foo)
    .map(do_something_else)
    .filter(whatever)
    .boxed()
}

The issue is that the stream created by fetch_many holds a reference to pretty much everything it needs. Creating ad-hoc futures is supported by the language, so whenever this happens I can just do:

fn get_foo(&self) -> BoxFuture<'_, Vec<Foo>> {
    async move {
        // Clone/capture whatever is needed.
        // ...
    }.boxed()
}

I'm looking for something similar for streams in stable Rust without having to implement a custom self-referential stream type, or alternatively some library that does the boilerplate for such wrapper types.

[semicoleon]

I haven't used it myself, so I'm not sure if it fully covers your use case, but I know async-stream exists

[tamasfe]

It is close enough to what I'm looking for, thank you!