未来生成器闭包错误：捕获的变量无法逃脱“FnMut”闭包体

Question

我想创建一个简单的 websocket 服务器。我想处理传入的消息并发送响应，但出现错误：

error: captured variable cannot escape `FnMut` closure body
  --> src\main.rs:32:27
   |
32 |       incoming.for_each(|m| async {
   |  _________________________-_^
   | |                         |
   | |                         inferred to be a `FnMut` closure
33 | |         match m {
34 | |             // Error here...
35 | |             Ok(message) => do_something(message, db, &mut outgoing).await,
36 | |             Err(e) => panic!(e)
37 | |         }
38 | |     }).await;
   | |_____^ returns a reference to a captured variable which escapes the closure body
   |
   = note: `FnMut` closures only have access to their captured variables while they are executing...
   = note: ...therefore, they cannot allow references to captured variables to escape

这对 Stack Overflow 有一些影响，但我在代码中看不到任何变量正在转义的地方。异步块不会同时运行，所以我没有看到任何问题。此外，我觉得我在做一些非常简单的事情：我得到了一个允许我将数据发送回客户端的类型，但是当在异步块中使用对它的引用时，它会给出编译错误。仅当我在异步代码中使用outgoing 或db 变量时才会出现该错误。

这是我的代码（错误在 handle_connection 函数中）：

ma​​in.rs

use tokio::net::{TcpListener, TcpStream};
use std::net::SocketAddr;
use std::sync::Arc;
use futures::{StreamExt, SinkExt};
use tungstenite::Message;
use tokio_tungstenite::WebSocketStream;

struct DatabaseConnection;

#[tokio::main]
async fn main() -> Result<(), ()> {
    listen("127.0.0.1:3012", Arc::new(DatabaseConnection)).await
}

async fn listen(address: &str, db: Arc<DatabaseConnection>) -> Result<(), ()> {
    let try_socket = TcpListener::bind(address).await;
    let mut listener = try_socket.expect("Failed to bind on address");

    while let Ok((stream, addr)) = listener.accept().await {
        tokio::spawn(handle_connection(stream, addr, db.clone()));
    }

    Ok(())
}

async fn handle_connection(raw_stream: TcpStream, addr: SocketAddr, db: Arc<DatabaseConnection>) {
    let db = &*db;
    let ws_stream = tokio_tungstenite::accept_async(raw_stream).await.unwrap();

    let (mut outgoing, incoming) = ws_stream.split();

    // Adding 'move' does also not work
    incoming.for_each(|m| async {
        match m {
            // Error here...
            Ok(message) => do_something(message, db, &mut outgoing).await,
            Err(e) => panic!(e)
        }
    }).await;
}

async fn do_something(message: Message, db: &DatabaseConnection, outgoing: &mut futures_util::stream::SplitSink<WebSocketStream<TcpStream>, Message>) {
    // Do something...

    // Send some message
    let _ = outgoing.send(Message::Text("yay".to_string())).await;
}

Cargo.toml

[dependencies]
futures = "0.3.*"
futures-channel = "0.3.*"
futures-util = "0.3.*"
tokio = { version = "0.2.*", features = [ "full" ] }
tokio-tungstenite = "0.10.*"
tungstenite = "0.10.*"

使用async move 时，出现以下错误：

代码

incoming.for_each(|m| async move {
    let x = &mut outgoing;
    let b = db;
}).await;

错误

error[E0507]: cannot move out of `outgoing`, a captured variable in an `FnMut` closure
  --> src\main.rs:33:38
   |
31 |       let (mut outgoing, incoming) = ws_stream.split();
   |            ------------ captured outer variable
32 | 
33 |       incoming.for_each(|m| async move {
   |  ______________________________________^
34 | |         let x = &mut outgoing;
   | |                      --------
   | |                      |
   | |                      move occurs because `outgoing` has type `futures_util::stream::stream::split::SplitSink<tokio_tungstenite::WebSocketStream<tokio::net::tcp::stream::TcpStream>, tungstenite::protocol::message::Message>`, which does not implement the `Copy` trait
   | |                      move occurs due to use in generator
35 | |         let b = db;
36 | |     }).await;
   | |_____^ move out of `outgoing` occurs here

Answer 1

FnMut 是一个匿名结构体，因为FnMut捕获了&mut outgoing，所以它成为这个匿名结构体内部的一个字段，这个字段将在每次调用FnMut时使用，它可以被多次调用.如果您以某种方式丢失它（通过返回或移动到另一个范围等......）您的程序将无法使用该字段进行进一步调用，因为安全性 Rust 编译器不允许您这样做（对于您的两种情况）。

在您的情况下，我们可以将其用作每次调用的参数，而不是捕获&mut outgoing，这样我们将保留outgoing 的所有权。您可以使用来自 futures-rs 的 fold 来做到这一点：

incoming
    .fold(outgoing, |mut outgoing, m| async move {
        match m {
            // Error here...
            Ok(message) => do_something(message, db, &mut outgoing).await,
            Err(e) => panic!(e),
        }

        outgoing
    })
    .await;

这可能看起来有点棘手，但它确实有效，我们使用常量累加器（outgoing），它将用作FnMut 的参数。

Playground（感谢@Solomon Ucko 创建可重现的示例）

另请参阅：

• How to return the captured variable from `FnMut` closure, which is a captor at the same time
• How can I move a captured variable into a closure within a closure?