9 changed files with 258 additions and 456 deletions
Split View
Diff Options
-
+17 -213Cargo.lock
-
+1 -1Cargo.toml
-
+37 -26src/client.rs
-
+0 -186src/control/controller.rs
-
+7 -2src/control/mod.rs
-
+0 -7src/control/request.rs
-
+192 -0src/control/ws.rs
-
+4 -9src/main.rs
-
+0 -12src/result.rs
+ 17
- 213
Cargo.lock
View File
+ 1
- 1
Cargo.toml
View File
+ 37
- 26
src/client.rs
View File
+ 0
- 186
src/control/controller.rs
View File
| @ -1,186 +0,0 @@ | |||
| use std::sync::mpsc; | |||
| use std::str; | |||
| use std::thread; | |||
| use rustc_serialize::json; | |||
| use websocket; | |||
| use websocket::{Receiver, Sender}; | |||
| use config; | |||
| use result; | |||
| use super::request::*; | |||
| use super::response::*; | |||
| type WebSocketReceiver = | |||
| websocket::receiver::Receiver<websocket::WebSocketStream>; | |||
| type WebSocketSender = | |||
| websocket::sender::Sender<websocket::WebSocketStream>; | |||
| type WebSocketClient = | |||
| websocket::Client<websocket::DataFrame, WebSocketSender, WebSocketReceiver>; | |||
| pub struct Controller { | |||
| client_tx: mpsc::Sender<Request>, | |||
| client_rx: mpsc::Receiver<Response>, | |||
| } | |||
| impl Controller { | |||
| pub fn new(tx: mpsc::Sender<Request>, | |||
| rx: mpsc::Receiver<Response>) | |||
| -> Self | |||
| { | |||
| Controller { | |||
| client_tx: tx, | |||
| client_rx: rx, | |||
| } | |||
| } | |||
| pub fn run(&mut self) { | |||
| let host = config::CONTROL_HOST; | |||
| let port = config::CONTROL_PORT; | |||
| let mut server = websocket::Server::bind((host, port)).unwrap(); | |||
| info!("Controller bound to {}:{}", host, port); | |||
| loop { | |||
| info!("Waiting for controller client"); | |||
| let client = match Self::try_get_client(&mut server) { | |||
| Ok(client) => client, | |||
| Err(e) => { | |||
| error!("Error accepting control connection: {}", e); | |||
| continue; | |||
| } | |||
| }; | |||
| // Empty client_rx of any messages that client has sent while | |||
| // no-one was connected. | |||
| while let Ok(_) = self.client_rx.try_recv() { /* continue */ } | |||
| // Notify client that a controller is connected. | |||
| self.client_tx.send(Request::ConnectNotification).unwrap(); | |||
| let (sender, receiver) = client.split(); | |||
| let (sender_tx, sender_rx) = mpsc::channel(); | |||
| // Handle incoming messages from controller in a separate thread, | |||
| // and forward them to the client through client_tx. | |||
| let tx = self.client_tx.clone(); | |||
| let handle = thread::spawn(move || { | |||
| Self::receiver_loop(receiver, tx, sender_tx); | |||
| }); | |||
| // Handle messages from client and forward them to the controller. | |||
| Self::sender_loop(sender, &mut self.client_rx, sender_rx); | |||
| // Sender loop has terminated, wait for receiver loop too. | |||
| handle.join(); | |||
| // Notify client that the controller has disconnected. | |||
| self.client_tx.send(Request::DisconnectNotification).unwrap(); | |||
| } | |||
| } | |||
| fn try_get_client(server: &mut websocket::Server) | |||
| -> result::Result<WebSocketClient> | |||
| { | |||
| let connection = try!(server.accept()); | |||
| let request = try!(connection.read_request()); | |||
| let client = try!(request.accept().send()); | |||
| Ok(client) | |||
| } | |||
| fn receiver_loop( | |||
| mut receiver: WebSocketReceiver, | |||
| client_tx: mpsc::Sender<Request>, | |||
| sender_tx: mpsc::Sender<()>) | |||
| { | |||
| for message_result in receiver.incoming_messages() { | |||
| let message : websocket::message::Message = match message_result { | |||
| Ok(message) => message, | |||
| Err(e) => { | |||
| warn!("Error receving websocket message: {}", e); | |||
| continue; | |||
| } | |||
| }; | |||
| match message.opcode { | |||
| websocket::message::Type::Text => { | |||
| let payload = message.payload; | |||
| match Self::handle_text_message(&payload, &client_tx) { | |||
| Ok(()) => (), | |||
| Err(e) => { | |||
| error!("Error handling text message: {}", e); | |||
| } | |||
| } | |||
| }, | |||
| websocket::message::Type::Close => { | |||
| info!("Received close request from controller."); | |||
| break; | |||
| }, | |||
| code => warn!( | |||
| "Unhandled websocket message with code {:?}", code | |||
| ), | |||
| } | |||
| } | |||
| info!("Shutting down websocket receiver"); | |||
| receiver.shutdown().unwrap(); | |||
| // Notify sender that the websocket is closed | |||
| sender_tx.send(()); | |||
| } | |||
| fn handle_text_message( | |||
| payload_bytes: &[u8], | |||
| client_tx: &mpsc::Sender<Request>) | |||
| -> result::Result<()> | |||
| { | |||
| let payload = try!(str::from_utf8(payload_bytes)); | |||
| let control_request = try!(json::decode(payload)); | |||
| debug!("Received control request: {:?}", control_request); | |||
| try!(client_tx.send(control_request)); | |||
| Ok(()) | |||
| } | |||
| fn sender_loop( | |||
| mut sender: WebSocketSender, | |||
| client_rx: &mut mpsc::Receiver<Response>, | |||
| sender_rx: mpsc::Receiver<()>) | |||
| { | |||
| loop { | |||
| select! { | |||
| _ = sender_rx.recv() => break, | |||
| response_result = client_rx.recv() => { | |||
| let response = match response_result { | |||
| Ok(response) => response, | |||
| Err(e) => { | |||
| error!( | |||
| "Error receiving from client channel: {}", e | |||
| ); | |||
| break; | |||
| } | |||
| }; | |||
| match Self::send_response(&mut sender, response) { | |||
| Ok(()) => (), | |||
| Err(e) => | |||
| error!("Error sending control response: {}", e), | |||
| } | |||
| } | |||
| } | |||
| } | |||
| info!("Shutting down websocket sender"); | |||
| sender.shutdown_all().unwrap(); | |||
| } | |||
| fn send_response(sender: &mut WebSocketSender, response: Response) | |||
| -> result::Result<()> | |||
| { | |||
| debug!("Sending control response: {:?}", response); | |||
| let encoded = try!(json::encode(&response)); | |||
| let message = websocket::Message::text(encoded); | |||
| try!(sender.send_message(&message)); | |||
| Ok(()) | |||
| } | |||
| } | |||
+ 7
- 2
src/control/mod.rs
View File
| @ -1,7 +1,12 @@ | |||
| mod controller; | |||
| mod request; | |||
| mod response; | |||
| mod ws; | |||
| pub use self::controller::Controller; | |||
| pub use self::ws::{ | |||
| listen, | |||
| Notification, | |||
| Sender, | |||
| SendError, | |||
| }; | |||
| pub use self::request::*; | |||
| pub use self::response::*; | |||
+ 0
- 7
src/control/request.rs
View File
+ 192
- 0
src/control/ws.rs
View File
| @ -0,0 +1,192 @@ | |||
| use std::error; | |||
| use std::fmt; | |||
| use std::sync::mpsc; | |||
| use rustc_serialize::json; | |||
| use ws; | |||
| use config; | |||
| use super::request::*; | |||
| use super::response::*; | |||
| /// This enum contains the possible notifications that the control loop will | |||
| /// send to the client. | |||
| #[derive(Debug)] | |||
| pub enum Notification { | |||
| /// A new controller has connected: control messages can now be sent on the | |||
| /// given channel. | |||
| Connected(Sender), | |||
| /// The controller has disconnected. | |||
| Disconnected, | |||
| /// An irretrievable error has arisen. | |||
| Error(String), | |||
| /// The controller has sent a request. | |||
| Request(Request), | |||
| } | |||
| /// This error is returned when a `Sender` fails to send a control request. | |||
| #[derive(Debug)] | |||
| pub enum SendError { | |||
| /// Error encoding the control request. | |||
| JSONEncoderError(json::EncoderError), | |||
| /// Error sending the encoded control request to the websocket. | |||
| WebSocketError(ws::Error), | |||
| } | |||
| impl fmt::Display for SendError { | |||
| fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { | |||
| match *self { | |||
| SendError::JSONEncoderError(ref err) => | |||
| write!(fmt, "JSONEncoderError: {}", err), | |||
| SendError::WebSocketError(ref err) => | |||
| write!(fmt, "WebSocketError: {}", err), | |||
| } | |||
| } | |||
| } | |||
| impl error::Error for SendError { | |||
| fn description(&self) -> &str { | |||
| match *self { | |||
| SendError::JSONEncoderError(_) => "JSONEncoderError", | |||
| SendError::WebSocketError(_) => "WebSocketError", | |||
| } | |||
| } | |||
| fn cause(&self) -> Option<&error::Error> { | |||
| match *self { | |||
| SendError::JSONEncoderError(ref err) => Some(err), | |||
| SendError::WebSocketError(ref err) => Some(err), | |||
| } | |||
| } | |||
| } | |||
| impl From<json::EncoderError> for SendError { | |||
| fn from(err: json::EncoderError) -> Self { | |||
| SendError::JSONEncoderError(err) | |||
| } | |||
| } | |||
| impl From<ws::Error> for SendError { | |||
| fn from(err: ws::Error) -> Self { | |||
| SendError::WebSocketError(err) | |||
| } | |||
| } | |||
| /// This struct is used to send control responses to the controller. | |||
| /// It encapsulates the websocket connection so as to isolate clients from | |||
| /// the underlying implementation. | |||
| #[derive(Clone, Debug)] | |||
| pub struct Sender { | |||
| sender: ws::Sender, | |||
| } | |||
| impl Sender { | |||
| /// Queues up a control response to be sent to the controller. | |||
| pub fn send(&mut self, response: Response) -> Result<(), SendError> { | |||
| let encoded = try!(json::encode(&response)); | |||
| try!(self.sender.send(encoded)); | |||
| Ok(()) | |||
| } | |||
| } | |||
| /// This struct handles a single websocket connection. | |||
| #[derive(Debug)] | |||
| struct Handler { | |||
| /// The channel on which to send notifications to the client. | |||
| client_tx: mpsc::Sender<Notification>, | |||
| /// The channel on which to send messages to the controller. | |||
| socket_tx: ws::Sender, | |||
| } | |||
| impl Handler { | |||
| fn send_to_client(&self, notification: Notification) -> ws::Result<()> { | |||
| match self.client_tx.send(notification) { | |||
| Ok(()) => Ok(()), | |||
| Err(e)=> { | |||
| error!("Error sending notification to client: {}", e); | |||
| Err(ws::Error::new(ws::ErrorKind::Internal, "")) | |||
| } | |||
| } | |||
| } | |||
| } | |||
| impl ws::Handler for Handler { | |||
| fn on_open(&mut self, _: ws::Handshake) -> ws::Result<()> { | |||
| info!("Websocket open"); | |||
| self.send_to_client(Notification::Connected(Sender { | |||
| sender: self.socket_tx.clone() | |||
| })) | |||
| } | |||
| fn on_close(&mut self, code: ws::CloseCode, reason: &str) { | |||
| info!("Websocket closed: code: {:?}, reason: {:?}", code, reason); | |||
| self.send_to_client(Notification::Disconnected).unwrap_or(()) | |||
| } | |||
| fn on_message(&mut self, msg: ws::Message) -> ws::Result<()> { | |||
| // Get the payload string. | |||
| let payload = match msg { | |||
| ws::Message::Text(payload) => payload, | |||
| ws::Message::Binary(_) => { | |||
| error!("Received binary websocket message from controller"); | |||
| return Err(ws::Error::new( | |||
| ws::ErrorKind::Protocol, "Binary message not supported" | |||
| )) | |||
| } | |||
| }; | |||
| // Decode the json control request. | |||
| let control_request = match json::decode(&payload) { | |||
| Ok(control_request) => control_request, | |||
| Err(e) => { | |||
| error!("Received invalid JSON message from controller: {}", e); | |||
| return Err(ws::Error::new( | |||
| ws::ErrorKind::Protocol, "Invalid JSON" | |||
| )) | |||
| } | |||
| }; | |||
| debug!("Received control request: {:?}", control_request); | |||
| // Send the control request to the client. | |||
| self.send_to_client(Notification::Request(control_request)) | |||
| } | |||
| } | |||
| /// Start listening on the socket address stored in configuration, and send | |||
| /// control notifications to the client through the given channel. | |||
| pub fn listen(client_tx: mpsc::Sender<Notification>) { | |||
| let websocket_result = ws::Builder::new().with_settings(ws::Settings { | |||
| max_connections: 1, | |||
| ..ws::Settings::default() | |||
| }).build(|socket_tx| Handler { | |||
| client_tx: client_tx.clone(), | |||
| socket_tx: socket_tx, | |||
| }); | |||
| let websocket = match websocket_result { | |||
| Ok(websocket) => websocket, | |||
| Err(e) => { | |||
| error!("Unable to build websocket: {}", e); | |||
| client_tx.send(Notification::Error( | |||
| format!("Unable to build websocket: {}", e) | |||
| )).unwrap(); | |||
| return | |||
| } | |||
| }; | |||
| let listen_result = websocket.listen( | |||
| (config::CONTROL_HOST, config::CONTROL_PORT) | |||
| ); | |||
| match listen_result { | |||
| Ok(_) => (), | |||
| Err(e) => { | |||
| error!("Unable to listen on websocket: {}", e); | |||
| client_tx.send(Notification::Error( | |||
| format!("Unable to listen on websocket: {}", e) | |||
| )).unwrap(); | |||
| } | |||
| } | |||
| } | |||