rust - future 生成器关闭时出错 : Captured variable cannot escape `FnMut` closure body

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 上有一些点击，但我在代码中没有看到变量正在转义的任何地方。异步块不会同时运行，所以我看不出任何问题。此外，我觉得我在做一些非常简单的事情:我得到了一个类型，它允许我将数据发送回客户端，但是当在 async 块中使用对它的引用时，它会产生编译错误。该错误仅在我在异步代码中使用 outgoing 或 db 变量时发生。
这是我的代码(错误在 handle_connection 函数中):
main.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?