websocket - 使用 rust-websocket 聊天

Question

我正在尝试使用 Rust-Websocket 创建一个简单的聊天室，多人可以在其中相互交谈。

我查看了这些示例，“server.rs”和“websockets.html”对我来说是一个不错的起点。所以我只是尝试启动它并从网络连接。一切正常，但我只能与自己通信，而不能与其他连接通信(因为它将消息直接发送回 sender 而不是每个连接)。

所以我试图获得一个包含所有 senders/clients 的向量，这样我就可以遍历它们并将消息发送给每个人，但这似乎是有问题的。我无法与 sender 或 client 通信，因为它不是线程安全的，我也无法复制其中任何一个。

我不确定我是不是完全没有理解整个借用100%或者它不是为了做这样的交叉连接通信。

服务器.rs:
https://github.com/cyderize/rust-websocket/blob/master/examples/server.rs

websockets.html:
https://github.com/cyderize/rust-websocket/blob/master/examples/websockets.html

我可能从错误的方向来处理这个问题。与所有其他线程共享收到的消息可能更容易。我稍微考虑了一下，但我唯一能想到的是使用 channels 从线程内部向外部发送消息。有什么方法可以直接在线程之间广播消息吗？我需要做的就是将一个字符串从一个线程发送到另一个线程。

Answer 1

所以这并不像人们想象的那么简单。

基本上，我使用了一个调度程序线程，它充当所有已连接客户端的控制中心。因此，每当客户端收到一条消息时，它就会被发送到调度程序，然后调度程序将消息分发给每个连接的客户端。

我还必须在另一个线程中接收消息，因为在 rust-websocket 中没有非阻塞方式来接收消息。然后我可以使用一个永久循环来检查从 websocket 和调度程序收到的新消息。

这是我的代码最终的样子:

extern crate websocket;

use std::str;
use std::sync::{Arc, Mutex};
use std::sync::mpsc;
use std::thread;

use websocket::{Server, Message, Sender, Receiver};
use websocket::header::WebSocketProtocol;
use websocket::message::Type;


fn main() {
    let server = Server::bind("0.0.0.0:2794").unwrap();

    let (dispatcher_tx, dispatcher_rx) = mpsc::channel::<String>();
    let client_senders: Arc<Mutex<Vec<mpsc::Sender<String>>>> = Arc::new(Mutex::new(vec![]));

    // dispatcher thread
    {
        let client_senders = client_senders.clone();
        thread::spawn(move || {
            while let Ok(msg) = dispatcher_rx.recv() {
                for sender in client_senders.lock().unwrap().iter() {
                    sender.send(msg.clone()).unwrap();
                }
            }
        });
    }

    // client threads
    for connection in server {
        let dispatcher = dispatcher_tx.clone();
        let (client_tx, client_rx) = mpsc::channel();
        client_senders.lock().unwrap().push(client_tx);

        // Spawn a new thread for each connection.
        thread::spawn(move || {
            let request = connection.unwrap().read_request().unwrap(); // Get the request
            let headers = request.headers.clone(); // Keep the headers so we can check them

            request.validate().unwrap(); // Validate the request

            let mut response = request.accept(); // Form a response

            if let Some(&WebSocketProtocol(ref protocols)) = headers.get() {
                if protocols.contains(&("rust-websocket".to_string())) {
                    // We have a protocol we want to use
                    response.headers.set(WebSocketProtocol(vec!["rust-websocket".to_string()]));
                }
            }

            let mut client = response.send().unwrap(); // Send the response

            let ip = client.get_mut_sender()
                .get_mut()
                .peer_addr()
                .unwrap();

            println!("Connection from {}", ip);

            let message: Message = Message::text("SERVER: Connected.".to_string());
            client.send_message(&message).unwrap();

            let (mut sender, mut receiver) = client.split();

            let(tx, rx) = mpsc::channel::<Message>();
            thread::spawn(move || {
                for message in receiver.incoming_messages() {
                    tx.send(message.unwrap()).unwrap();
                }
            });

            loop {
                if let Ok(message) = rx.try_recv() {
                    match message.opcode {
                        Type::Close => {
                            let message = Message::close();
                            sender.send_message(&message).unwrap();
                            println!("Client {} disconnected", ip);
                            return;
                        },
                        Type::Ping => {
                            let message = Message::pong(message.payload);
                            sender.send_message(&message).unwrap();
                        },
                        _ => {
                            let payload_bytes = &message.payload;
                            let payload_string = match str::from_utf8(payload_bytes) {
                                Ok(v) => v,
                                Err(e) => panic!("Invalid UTF-8 sequence: {}", e),
                            };
                            let msg_string = format!("MESSAGE: {}: ", payload_string);
                            dispatcher.send(msg_string).unwrap();
                        }
                    }
                }
                if let Ok(message) = client_rx.try_recv() {
                    let message: Message = Message::text(message);
                    sender.send_message(&message).unwrap();
                }
            }
        });
    }
}

http://pastebin.com/H9McWLrH