CODEWITHSUNDEEP
multithreadingrust
ncoish
ncoish

Reputation: 343

Calling an FnMut callback from another thread

I am writing a Phoenix client library for Rust, taking advantage of the async websocket client from rust-websockets. Right now I am having trouble figuring out how to pass callback functions into the thread that is handling the websocket traffic. I have a simplified struct:

pub struct Socket {
    endpoint:               String,
    connected:              Arc<AtomicBool>,
    state_change_close:     Option<Box<FnMut(String)>>,
}

This struct has a connect function laid out as follows:

pub fn connect(&mut self) -> Result<(), String> {
    if self.connected.load(Ordering::Relaxed) {
        return Ok(())
    }
    
    // Copy endpoint string, otherwise we get an error on thread::spawn
    let connection_string = self.endpoint.clone();
    let (usr_msg, stdin_ch) = mpsc::channel(0);

    let connection_thread = thread::spawn(move || {
        // tokio core for running event loop
        let mut core = Core::new().unwrap();
        let runner = ClientBuilder::new(&connection_string)
            .unwrap()
            .add_protocol("rust-websocket")
            .async_connect_insecure(&core.handle())
            .and_then(|(duplex, _)| {
                let (sink, stream) = duplex.split();
                stream.filter_map(|message| {
                    println!("Received Message: {:?}", message);
                    match message {
                        OwnedMessage::Close(e) => {
                            // This is the line where I am trying to call the callback
                            if let Some(ref mut func) = self.state_change_close {
                                (func)(e.unwrap().reason);
                            }
                            Some(OwnedMessage::Close(e))
                        },
                        _ => None,
                    }
                })
                .select(stdin_ch.map_err(|_| WebSocketError::NoDataAvailable))
                .forward(sink)
            });
        
        // Start the event loop
        core.run(runner).unwrap();
    });
    self.connected.store(true, Ordering::Relaxed);
    return Ok(())
}

When I try to compile this code I get the following error:

error[E0277]: the trait bound `std::ops::FnMut(std::string::String) + 'static: std::marker::Send` is not satisfied
  --> src\socket.rs:99:29
   |
99 |     let connection_thread = thread::spawn(move || {
   |                             ^^^^^^^^^^^^^ the trait `std::marker::Send` is not implemented for `std::ops::FnMut(std::string::String) + 'static`
   |

I have tried changing the type of state_change_close to a Mutex<Option<...>> to avoid thread safety issues, but that did not help with this problem. Is what I'm trying to do possible?

Upvotes: 3

Views: 1730

Answers (1)

ncoish
ncoish

Reputation: 343

After doing some more research I realized that I just had to modify Option<Box<FnMut(String)>> to be Option<Box<FnMut(String) + Send>> and copy that around my code to everywhere that the callback might be set. Learning more about trait objects!

Upvotes: 7

Related Questions