Как вернуть захваченную переменную из замыкания `FnMut`, которое является одновременно и захватчиком

Question

У меня есть функция, collect_n, которая возвращает Future, который многократно polls a 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

Как мне выполнить sh, что я пытаюсь сделать

Answer 1

Это невозможно, если ваша переменная не является разделяемой, как Rc или Arc, так как FnMut может вызываться несколько раз, возможно, что ваше закрытие должно вернуть захваченную переменную более одного раза. Но после возврата вы теряете право собственности на переменную, поэтому вы не можете вернуть ее обратно, из-за безопасности Rust не позволяет вам сделать это.

Согласно вашей логике c мы знаем, что однажды ваша 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, иначе для последующих опросов будет c.

Другое решение будет использовать 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 ( Not captured ), мы можем вернуть его столько раз, сколько захотим, поэтому компилятор снова доволен. Благодаря fuse() это закрытие не будет вызываться после успешного возврата, в противном случае оно вернет Receiver с отброшенными Sender.

См. Также:

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