Call stored closure from an async closure in rust

Question

I have a closure that mutates variables designed outside of it. How would I go about calling this closure that modifies the state from inside an async scope?

I have the following code (abstracted, to show the issue):

#[tokio::main]
async fn main() {
    let mut y = false;

    let mut incr = |z: bool| {
        y = z;
    };

    stream::iter(0..1).for_each(|_| async {
        incr(true);
    }).await;
});

Which produces the following:

error: captured variable cannot escape `FnMut` closure body
  --> src/main.rs:40:37
   |
36 |       let mut incr = |z: bool| {
   |           -------- variable defined here
...
40 |       stream::iter(0..1).for_each(|_| async {
   |  ___________________________________-_^
   | |                                   |
   | |                                   inferred to be a `FnMut` closure
41 | |         incr(true);
   | |         ---- variable captured here
42 | |     }).await;
   | |_____^ returns an `async` block that contains a reference to a captured variable, which then escapes the closure body
   |
   = note: `FnMut` closures only have access to their captured variables while they are executing...
   = note: ...therefore, they cannot allow references to captured variables to escape

Now, I believe I understand why the error is occurring. I just cannot figure out a way around this.

For context:

1. I have a websocket client and I am reading from the stream
2. Every time I receive data from the stream, I am transforming it
3. I then need to call a closure with the transformed data to be used elsewhere - essentially like an EventEmitter in JavaScript.

Am I going about this the wrong way? I am a JavaScript developer, so I'm having to change my way of thinking here.

Answer 1

In rust, usually, you need to use some kind of synchronization mechanisms. It would be really difficult to send a reference to a normal variable. So you would need to use some Arc, RwLock, Mutex depending on your needs. Also you will probably want to move them into the closure to avoid lifetimes problems.

In this case I think Arc<Cell> would do:

use std::cell::Cell;
use std::sync::Arc;
use futures::StreamExt;
use futures::stream;


#[tokio::main]
async fn main() {
    let yy = Arc::new(Cell::new(false));
    let y = yy.clone();
    let incr = move |z: bool| {
        y.set(z);
    };

    stream::iter(0..1).for_each(|_| async {
        incr(true);
    }).await;
    
    println!("{yy:?}");
}

Playground

Check out the sync for further reading. As well as cell documentation