异步和流的生命周期

Question

我正在尝试创建一个消耗Stream的函数，并在连续出现max_consecutive_fails失败时截断它。但是，事情做得不好（E0495）。我将Stream s更改为Iterator s（并删除了async s），它确实起作用了。为什么会这样？如何重构该代码（起作用）？

use futures::stream::Stream;
pub fn max_fail<'a, T>(stream : impl Stream<Item = Option<T>> +'a , max_consecutive_fails: usize) -> impl Stream +'a where T : 'a
{
    use futures::stream::StreamExt;
    let mut consecutive_fails = 0;
    stream.take_while(move |x| async {
        if x.is_some(){
            consecutive_fails = 0;
            true
        }
        else{
            consecutive_fails += 1;
            consecutive_fails != max_consecutive_fails
        }
    })
}

以下是我试图指出问题所在的最小化示例，但我仍然无法理解rustc错误消息。

use futures::stream::Stream;
pub fn minified_example<'a>(stream: impl Stream<Item = bool> + 'a) -> impl Stream + 'a
{
    use futures::stream::StreamExt;
    stream.take_while( |x| async { *x })
}

Answer 1

异步块（async { ... }）类似于闭包捕获环境的方式。默认情况下，来自其他作用域的变量的每次使用都是通过引用进行的，这意味着该块创建的impl core::future::Future不能超过其捕获的变量。]​​>

您需要将x移至带有async move { ... }的块中（与闭包一样）