rust - Cannot move out of captured variables in an `FnMut` closure

Question

pub fn create_future(
    notificator: mpsc::Sender<usize>,
    proxy: Proxy,
) -> impl Future<Item = (), Error = ()> {
    proxy.something()
        .and_then(move |sub| {
            sub.for_each(move |a| { // <---- Closure A
                proxy.something_else(a)
                    .and_then(move |b| { // <---- Closure B
                        notificator.send(b.len());  // <---- Error!
                        Ok(())
                    })
                    .or_else(|e| {
                        panic!("oops {}", e);
                        Ok(())
                    })
            })
        })
        .map_err(|e| {
            ()
        })
}

This doesn't compile because

.and_then(move |b| {
          ^^^^^^^^^ cannot move out of captured outer variable in an `FnMut` closure

My understanding of the error is:

1. Closure B is FnMut, and it captures notificator by taking its ownership
2. In Closure B, send again needs to take the ownership
3. Now both send and Closure B are modifying notificator thus the error.

Is my understanding right? How can I solve this problem?

Answer 1

Nested closures are tricky.

Consider this:

fn use_a_fn_multiple_times(f: impl Fn(String)) {
    f("foo".to_owned());
    f("bar".to_owned());
}

fn use_fn_once(f: impl FnOnce() -> Vec<u8>) {
    println!("Bytes: {:?}", f());
}

fn main() {
  use_a_fn_multiple_times(|a: String| {
    use_fn_once(move || a.into_bytes());
  });
}

Playground

Notice that the inner closure captures a by move. This is fine. The outer closure owns a and can do with it what it wants, including moving it into the inner closure (which, because it consumes its captured value, is a FnOnce).

The outer closure is called multiple times, each time with a new string, and every time a new inner closure capturing this string is created.

But what if what you want to capture comes from even further out?

fn use_a_fn_multiple_times(f: impl Fn(String)) {
    f("foo".to_owned());
    f("bar".to_owned());
}

fn use_fn_once(f: impl FnOnce() -> Vec<u8>) {
    println!("Bytes: {:?}", f());
}

fn main() {
  let outer_s = "see:".to_owned();

  use_a_fn_multiple_times(|a: String| {
    use_fn_once(move || {
        let mut v = outer_s.into_bytes();
        v.extend(a.into_bytes());
        v
    });
  });
}

Playground

Then you get the error you're seeing (except for Fn vs FnMut, which is immaterial to the problem). The inner closure is created anew on every call to the outer closure (it has to be, because it has to capture a every time), but it tries to capture outer_s by move every time. This can't work; after the first time, outer_s is moved from and thus invalid.

To map this back to your code, it's wrong to say "Closure B captures notificator", because there isn't just one Closure B. There's as many as necessary, however often your nested and_then and for_each calls will end up in that piece of code. But only one can ever capture by move.

So to solve this, you either need to make sure there's only one Closure B, or make sure you have enough mpsc::Senders for everyone.

The first way is done by pulling the closure out of the nested context.

let closure_b = move |b| {
    notificator.send(b.len());
    Ok(())
};
proxy.something()
    .and_then(move |sub| {
        sub.for_each(move |a| { // <---- Closure A
            proxy.something_else(a)
                .and_then(closure_b)
                .or_else(|e| {
                    panic!("oops {}", e);
                    Ok(())
                })
        })
    })
    .map_err(|e| {
        ()
    })

except that won't work, since now Closure A faces the same issue, so you have to do it multiple times:

let closure_b = move |b| {
    notificator.send(b.len());
    Ok(())
};
let closure_a = move |a| {
    proxy.something_else(a)
        .and_then(closure_b)
        .or_else(|e| {
            panic!("oops {}", e);
            Ok(())
        })
};
proxy.something()
    .and_then(move |sub| {
        sub.for_each(closure_a)
    })
    .map_err(|e| {
        ()
    })

The second way involves a lot of clone() calls, and since I can't type-check your code, I won't attempt to write it.

When all is said and done, though, your code will still fail, because you're moving out of Proxy while also trying to use it.