rust - 如何在Rust中使用StreamExt::scan方法更改异步 block 内的状态？

Question

我正在尝试使用StreamExt的扫描方法。如果我没有异步块，它会完美地工作。

use futures::{stream, StreamExt};

#[tokio::main]
async fn main() {
    stream::iter(1..10)
        .scan(0, |s, i| {
            *s += i;
            futures::future::ready(Some(*s))
        })
        .for_each(|x| async move {
            println!("{:?}", x);
        })
        .await;
}

但是，如果我确实有 async块，则无法编译。

use futures::{stream, StreamExt};

#[tokio::main]
async fn main() {
    stream::iter(1..10)
        .scan(0, |s, i| async move {
            *s += i;
            Some(*s)
        })
        .for_each(|x| async move {
            println!("{:?}", x);
        })
        .await;
}

错误是:

error: borrowed data cannot be stored outside of its closure
  --> src/main.rs:6:36
   |
5  |  /     stream::iter(1..10)
6  |  |         .scan(0, |s, i| async move {
   |  |__________________------____________^
   | ||                  |
   | ||                  ...because it cannot outlive this closure
7  | ||             *s += i;
8  | ||             Some(*s)
9  | ||         })
   | ||_________^ cannot be stored outside of its closure
...   |
12 |  |         })
13 |  |         .await;
   |  |______________- borrowed data cannot be stored into here...

如何解决这个问题并改变 async块中的状态？

Answer 1

您不能跨async边界共享引用。一旦代码在async上下文中执行，就无法再跟踪生存期，因为编译器不知道将来什么时候完成。
一种解决方案是使用引用计数的智能指针和内部可变性:

use futures::{stream, StreamExt};
use std::cell::Cell;
use std::rc::Rc;

#[tokio::main]
async fn main() {
    stream::iter(1..10)
        .scan(Rc::new(Cell::new(0)), |s, i| {
            let s = s.clone();
            async move {
                s.set(s.get() + i);
                Some(s.get())
            }
        })
        .for_each(|x| async move {
            println!("{:?}", x);
        })
        .await;
}

闭包参数中的 s是 &mut Rc<Cell<i32>>，无法在 async边界上移动。但是克隆的版本是一个 Rc<Cell<i32>>，可以将其移动到此处，因为没有生命周期需要跟踪。