server.rs

extern crate websocket;

use std::sync::{Arc, Mutex};
use std::sync::mpsc;
use std::thread;
use websocket::{Server, Message, Sender, Receiver};
use websocket::message::Type;
use websocket::header::WebSocketProtocol;

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

	let (dispatch_send, dispatch_recv) = mpsc::channel();
	let client_senders: Arc<Mutex<Vec<mpsc::Sender<&'static str>>>> = Arc::new(Mutex::new(vec![]));

	// dispatcher thread
	{
		let client_senders = client_senders.clone();
		thread::spawn(move || {
			while let Ok(msg) = dispatch_recv.recv() {
				for client in client_senders.lock().unwrap().iter() {
					client.send("got a message");
				}
			}
		});
	}

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

		// 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("Hello".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());
				}
			});

			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();
						}
						_ => {
							dispatcher.send("got a message");
						}
					}
				}
				if let Ok(message) = client_recv.try_recv() {
					println!("{}", message);
				}
			}
		});
	}
}