Rust编译错误：无法在FnMut闭包中移出`client`变量

我在使用mastodon-async crate处理Mastodon用户时间线事件，需要将特定事件内容发送到MQTT主题，参考官方示例开发后遇到编译错误，尝试用fold替代try_for_each也未能解决。

问题代码 #

use std::thread;
use std::time::Duration;

use anyhow::{anyhow, Result};
use futures_util::TryStreamExt;
use log::{info, warn};
use mastodon_async::helpers::toml;
use mastodon_async::Mastodon;
use mastodon_async::prelude::{Event, Status};
use rumqttc::{AsyncClient, EventLoop, MqttOptions, QoS};

async fn get_mastodon_data() -> Result<Mastodon> {
    if let Ok(data) = toml::from_file("weaselbot-mastodon.toml") {
        Ok(Mastodon::from(data))
    } else {
        Err(anyhow!("Mastodon doesn't seems to have been registered, please execute the register command !"))
    }
}

fn process_weasel_toot(status: Status) -> Result<String> {
    if !status.media_attachments.is_empty() {
        for attachement in &status.media_attachments {
            match &attachement.remote_url {
                Some(url) => {
                    let url = url.clone();
                    return Ok(url);
                }
                None => warn!("can't find remote url !"),
            }
        }
    } else {
        warn!("status without any attachments");
    }

    Err(anyhow!("Status can't be processed for media"))
}

async fn evt(mut eventloop: EventLoop) {
    while let notification = eventloop.poll().await {
        match notification {
            _ => ()
        }
    }
}

pub async fn mastodon_run() -> Result<()> {
    let mastodon = get_mastodon_data().await?;
    let stream = mastodon.stream_user().await?;


    let mut mqttoptions = MqttOptions::new("mastodon", "127.0.0.1", 1883);
    mqttoptions.set_keep_alive(Duration::from_secs(5));

    let (client, eventloop) = AsyncClient::new(mqttoptions, 10);
    // process eventloop
    thread::spawn(move || evt(eventloop));

    info!("Waiting for Mastodon events...");
    stream.try_for_each(|event| async move {
        match event {
            Event::Update(status) => {
                info!("found update event !");
                match process_weasel_toot(status) {
                    Ok(url) => {
                        info!("Found url {}", url);
                        client.publish("mastodon", QoS::AtLeastOnce, false, url);
                    }
                    _ => (),
                };
            }
            _ => (),
        }
        Ok(())
    }).await?;

    Ok(())
}


#[tokio::main]
async fn main() -> Result<()>
{
    mastodon_run().await
}

依赖配置 #

[dependencies]
log = "0.4.17"
anyhow = "1.0.69"
toml = "0.5.11"
rumqttc = "0.20.0"
tokio = { version = "1.26.0", features = ["macros", "rt-multi-thread"] }
futures-util = "0.3.25"
mastodon-async = { version = "1.1.0", features = ["toml", "mt"] }

编译错误信息 #

error[E0507]: cannot move out of `client`, a captured variable in an `FnMut` closure
  --> src/main.rs:59:41
   |
54 |       let (client, eventloop) = AsyncClient::new(mqttoptions, 10);
   |            ------ captured outer variable
...
59 |       stream.try_for_each(|event| async move {
   |  _________________________-------____________^
   | |                         |
   | |                         captured by this `FnMut` closure
60 | |         match event {
61 | |             Event::Update(status) => {
62 | |                 info!("found update event !");
...  |
66 | |                         client.publish("mastodon", QoS::AtLeastOnce, false, url);
   | |                         ------
   | |                         |
   | |                         variable moved due to use in generator
   | |                         move occurs because `client` has type `rumqttc::AsyncClient`, which does not implement the `Copy` trait
...  |
73 | |         Ok(())
74 | |     }).await?;
   | |_____^ move out of `client` occurs here

For more information about this error, try `rustc --explain E0507`.

解决方案 #

错误原因 #

rumqttc::AsyncClient未实现Copy trait，try_for_each的闭包是FnMut类型，每次迭代都会尝试获取client的所有权，导致编译报错。同时原代码用thread::spawn运行异步函数evt是错误的，普通线程无法执行异步任务，需要用Tokio的异步任务调度器。

修改步骤 #

1. 引入Arc共享AsyncClient的所有权，AsyncClient本身实现了Send和Sync，支持线程安全共享。
2. 将AsyncClient包装进Arc，在闭包中克隆Arc避免所有权移动问题。
3. 替换thread::spawn为tokio::spawn，正确运行异步的MQTT事件循环。

修改后的完整代码 #

use std::sync::Arc;
use std::time::Duration;

use anyhow::{anyhow, Result};
use futures_util::TryStreamExt;
use log::{info, warn};
use mastodon_async::helpers::toml;
use mastodon_async::Mastodon;
use mastodon_async::prelude::{Event, Status};
use rumqttc::{AsyncClient, EventLoop, MqttOptions, QoS};
use tokio::spawn;

async fn get_mastodon_data() -> Result<Mastodon> {
    if let Ok(data) = toml::from_file("weaselbot-mastodon.toml") {
        Ok(Mastodon::from(data))
    } else {
        Err(anyhow!("Mastodon doesn't seem to have been registered, please execute the register command!"))
    }
}

fn process_weasel_toot(status: Status) -> Result<String> {
    if !status.media_attachments.is_empty() {
        for attachment in &status.media_attachments {
            match &attachment.remote_url {
                Some(url) => {
                    let url = url.clone();
                    return Ok(url);
                }
                None => warn!("Can't find remote URL!"),
            }
        }
    } else {
        warn!("Status without any attachments");
    }

    Err(anyhow!("Status can't be processed for media"))
}

async fn evt(mut eventloop: EventLoop) {
    while let _notification = eventloop.poll().await {
        // 可根据需求添加MQTT事件处理逻辑，比如重连、接收消息等
    }
}

pub async fn mastodon_run() -> Result<()> {
    let mastodon = get_mastodon_data().await?;
    let stream = mastodon.stream_user().await?;

    let mut mqttoptions = MqttOptions::new("mastodon", "127.0.0.1", 1883);
    mqttoptions.set_keep_alive(Duration::from_secs(5));

    let (client, eventloop) = AsyncClient::new(mqttoptions, 10);
    // 用Tokio异步任务运行MQTT事件循环
    spawn(evt(eventloop));

    // 将client包装进Arc实现共享所有权
    let client = Arc::new(client);

    info!("Waiting for Mastodon events...");
    stream.try_for_each(|event| {
        // 克隆Arc，避免所有权移动
        let client = Arc::clone(&client);
        async move {
            match event {
                Event::Update(status) => {
                    info!("Found update event!");
                    if let Ok(url) = process_weasel_toot(status) {
                        info!("Found URL: {}", url);
                        // 等待publish完成，确保消息发送成功并处理错误
                        client.publish("mastodon", QoS::AtLeastOnce, false, url).await?;
                    }
                }
                _ => (),
            }
            Ok(())
        }
    }).await?;

    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    mastodon_run().await
}

额外说明 #

• Arc是原子引用计数指针，允许多个异步任务共享同一个AsyncClient实例，克隆仅增加引用计数，不会复制对象本身。
• 原代码中client.publish未等待异步操作完成，修改后添加.await?确保消息发送完成，同时处理MQTT可能返回的错误。
• 修正了拼写错误（attachement改为attachment），优化了日志信息格式。

内容的提问来源于stack exchange，提问作者iXô