如何从`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()
}

基本上我们捕获的是container而不是receiver。所以这让编译器很高兴。我使用fuse() 来确保在返回Async::Ready 后不会再次调用闭包，否则会为进一步的投票而恐慌。

另一种解决方案是使用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()

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

另见：

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