How to use a collection of websockets across threads?

[srahner-hicknhack]

Hi, I really like your library with its simple usage and would like to use it
for experimenting and learning about Rust and websockets by implementing a very
simple chat server.

For that I have a vector of websockets being wrapped in Arc and Mutexes in order
to be able to use them across threads. As shown in your websocket example I call
websocket::start in order to use websockets. Then I spawn a new thread
in which I receive the new websocket from the Receiver<Websocket>
and push it to a vector wrapped in an Arc such that I can store it in the vector
and handle messages for it in the thread. It is also wrapped in a Mutex to be
usable from other websocket threads.

When a new message is received from the websocket (from a HTML page) I want to
send it to all opened websockets using the vector containing the websockets.

However, this won´t work as let message = match ws_arc_.lock().unwrap().next()
locks a socket making it unavailable when looping the websockets in another thread
such that I cannot send data to it.

So, what would be the correct way to handle a collection of websockets across
different threads to be able to send data to all websockets?

fn main() {
    let clients: Arc<Mutex<Vec<Arc<Mutex<websocket::Websocket>>>>> = 
                                            Arc::new(Mutex::new(Vec::new()));
     router!(request,
        // Some more routes ...

        // create websocket
        (GET) (/ws) => {
                let (response, websocket) = try_or_400!(websocket::start::<String>(&request, None));
                let clients_ = clients.clone();
                let ip = request.remote_addr().clone();
                thread::spawn(move || {
                    let mut ws = websocket.recv().unwrap();
                    let ws_arc = Arc::new(Mutex::new(ws));
                    let ws_arc_ = ws_arc.clone();
                    clients_.lock().unwrap().push(ws_arc);
                    ws_arc_.lock().unwrap().send_text(format!("Connected {:?}", &ip).as_str());

                    loop {
                        let message = match ws_arc_.lock().unwrap().next() {
                            Some(m) => m,
                            None => break,
                        };

                        match message {
                            websocket::Message::Text(txt) => {
                                for other in clients_.lock().unwrap().iter() {
                                    other.lock().unwrap().send_text(&txt).unwrap();
                                }
                            },
                            websocket::Message::Binary(_) => {
                                println!("received binary from a websocket");
                            },
                        }
                    }
                });

                response
            },

            _ => Response::html("404 error."")
        )
    });
}

[tomaka]

That's a known problem with synchronous IO in Rust general.
Instead of sending the message directly you will have to put it in an intermediary buffer, and each websocket thread will have to read from that buffer from time to time.

[tomaka]

I admit that the API exposed by rouille is not very ergonomic for that. It just mimics the API of regular sockets and has the same limitations as regular sockets of the Rust stdlib.

[srahner-hicknhack]

Thank for your reply. How would you combine reading from an intermediary buffer with

let message = match ws_arc_.lock().unwrap().next() {
     Some(m) => m,
    None => break,
};

I would have to read from the buffer and additionally wait for new messages to be received by the websocket. However, calling next() would wait until new data is sent. Both ways of processing data would have to be combined somehow.