rust - 如何从 `FnMut` 闭包中返回捕获的变量，同时也是一个捕获者

Question

我有一个函数，collect_n ，返回 Future反复polls一个 futures::sync::mpsc::Receiver并将结果收集到一个向量中，并使用该向量进行解析。问题是它消耗了 Receiver ，所以 Receiver不能再次使用。我正在尝试编写一个拥有 Receiver 所有权的版本。但是当返回 Future 时，将所有权返回给调用者解决。

这是我写的:

/// Like collect_n but returns the mpsc::Receiver it consumes so it can be reused.
pub fn collect_n_reusable<T>(
    mut rx: mpsc::Receiver<T>,
    n: usize,
) -> impl Future<Item = (mpsc::Receiver<T>, Vec<T>), Error = ()> {
    let mut events = Vec::new();

    future::poll_fn(move || {
        while events.len() < n {
            let e = try_ready!(rx.poll()).unwrap();
            events.push(e);
        }
        let ret = mem::replace(&mut events, Vec::new());
        Ok(Async::Ready((rx, ret)))
    })
}

这会导致编译错误:

error[E0507]: cannot move out of `rx`, a captured variable in an `FnMut` closure
   --> src/test_util.rs:200:26
    |
189 |     mut rx: mpsc::Receiver<T>,
    |     ------ captured outer variable
...
200 |         Ok(Async::Ready((rx, ret)))
    |                          ^^ move occurs because `rx` has type `futures::sync::mpsc::Receiver<T>`, which does not implement the `Copy` trait

我怎样才能完成我想做的事情？

Answer 1

除非您的变量像 Rc 一样可共享，否则这是不可能的。或 Arc , 自 FnMut可以多次调用您的闭包可能需要返回多个捕获的变量。但是在返回后，你失去了变量的所有权，所以你不能把它返回，由于安全性，Rust 不允许你这样做。

根据您的逻辑，我们知道一旦您的Future准备好了就不需要再次轮询了，所以我们可以不使用智能指针来创建解决方案。让我们考虑一个像下面这样的容器对象，我使用了 Option :

use futures::sync::mpsc;
use futures::{Future, Async, try_ready};
use futures::stream::Stream;
use core::mem;

pub fn collect_n_reusable<T>(
    mut rx: mpsc::Receiver<T>,
    n: usize,
) -> impl Future<Item = (mpsc::Receiver<T>, Vec<T>), Error = ()> {
    let mut events = Vec::new();
    let mut rx = Some(rx); //wrapped with an Option 
    futures::future::poll_fn(move || {
        while events.len() < n {
            let e = try_ready!(rx.as_mut().unwrap().poll()).unwrap();//used over an Option
            events.push(e);
        }
        let ret = mem::replace(&mut events, Vec::new());

        //We took it from option and returned.
        //Careful this will panic if the return line got execute more than once.
        Ok(Async::Ready((rx.take().unwrap(), ret)))
    })
    .fuse()
}

基本上我们捕获了 容器 不是 接收器 .所以这让编译器很高兴。我用了 fuse()确保在返回 Async::Ready 后不会再次调用闭包，否则它会为进一步的民意调查而 panic 。

另一种解决方案是使用 std::mem::swap :

futures::future::poll_fn(move || {
    while events.len() < n {
        let e = try_ready!(rx.poll()).unwrap();
        events.push(e);
    }
    let mut place_holder_receiver = futures::sync::mpsc::channel(0).1; //creating object with same type to swap with actual one. 
    let ret = mem::replace(&mut events, Vec::new());
    mem::swap(&mut place_holder_receiver, &mut rx); //Swapping the actual receiver with the placeholder

    Ok(Async::Ready((place_holder_receiver, ret))) //so we can return placeholder in here
})
.fuse()

只是我已经交换了实际的 receiver与 place_holder_receiver .由于占位符是在 FnMut 中创建的( 未捕获 )我们可以根据需要多次返回它，因此编译器再次感到高兴。感谢 fuse()返回成功后不会调用这个闭包，否则会返回 Receiver s 带有一个丢弃的 Sender .

另见:

https://docs.rs/futures/0.3.4/futures/future/trait.FutureExt.html#method.fuse