围绕异步和流的生命

问题描述

我正在尝试制作一个使用 Stream 的函数，并在连续发生 max_consecutive_fails 失败时截断它.但是，事情做得不好(E0495).我将 Stream s更改为 Iterator s(并删除了 async s)，它确实起作用了.为什么会这样?如何重构该代码(起作用)?

I am trying to make a function that consumes Stream and truncate it when there are max_consecutive_fails consecutive fails. However, things didn't work well(E0495). I changed Streams to Iterators (and removed asyncs) and it simply worked. Why does this happen? How can I refactor this code (to work)?

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错误消息.

The below one is the minimized example I tried to point out what the problem is, but I still wasn't able to understand the rustc error message.

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 })
}

推荐答案

异步块( async {...} )类似于闭包捕获环境的方式.默认情况下，来自其他作用域的变量的每次使用都是参考，这意味着该块创建的 impl core :: future :: Future 不能超过其捕获的变量.

Async blocks (async { ... }) are similar to closures in the way they capture their environment. By default, every use of a variable from the other scope is by reference, which means the impl core::future::Future created by the block cannot outlive the variables it captures.

您需要通过 async move {...} (就像闭包一样)将 x 移入块中

You need to move x into the block with async move { ... } (Just like with closures)

这篇关于围绕异步和流的生命的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！