Add Controller.

src/config.rs

@@ -8,3 +8,6 @@ pub const PASSWORD: &'static str = "ijklmnop";
 
 pub const SERVER_HOST : &'static str = "server.slsknet.org";
 pub const SERVER_PORT : u16 = 2242;
+
+pub const CONTROL_HOST: &'static str = "localhost";
+pub const CONTROL_PORT: u16 = 2244;

src/control.rs

@@ -1,66 +1,123 @@
 use std::io;
 use std::io::{Read, Write};
+use std::sync::mpsc;
+use std::thread;
 
 use rustc_serialize::json;
-
 use mio::tcp::TcpStream;
+use websocket;
+use websocket::{Receiver, Sender};
 
-#[derive(RustcDecodable, RustcEncodable)]
-pub enum ControlRequest {
-    LoginRequest(LoginRequest),
+use client;
+use config;
+
+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<ControlRequest>,
+    client_rx: mpsc::Receiver<ControlResponse>,
 }
 
-impl ControlRequest {
-    fn read_from<R: Read + Sized>(&self, mut reader: R) -> io::Result<Self> {
-        let mut string = String::new();
-        try!(reader.read_to_string(&mut string));
-        match json::decode(&string) {
-            Ok(request) => Ok(request),
-            Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)),
+impl Controller {
+    pub fn new(tx: mpsc::Sender<ControlRequest>,
+               rx: mpsc::Receiver<ControlResponse>)
+        -> Self
+    {
+        Controller {
+            client_tx: tx,
+            client_rx: rx,
         }
     }
-}
 
-#[derive(RustcDecodable, RustcEncodable)]
-pub enum ControlResponse {
-    LoginResponse(LoginResponse),
-}
+    pub fn run(&mut self) {
+        let host = config::CONTROL_HOST;
+        let port = config::CONTROL_PORT;
+        loop {
+            let client = Self::get_client(host, port);
+            let (mut sender, mut receiver) = client.split();
+            let tx = self.client_tx.clone();
+            thread::spawn(move || {
+                Self::receiver_loop(receiver, tx);
+            });
+            Self::sender_loop(sender, &mut self.client_rx);
+        }
+    }
 
-impl ControlResponse {
-    fn write_to<W: Write + Sized>(&self, mut writer: W) -> io::Result<()> {
-        match json::encode(self) {
-            Ok(json_string) => {
-                try!(writer.write(&json_string.into_bytes()));
-                Ok(())
-            },
+    fn get_client(host: &str, port: u16) -> WebSocketClient
+    {
+        let mut server = websocket::Server::bind((host, port)).unwrap();
+        loop {
+            match Self::try_get_client(&mut server) {
+                Ok(client) => return client,
+                Err(e) => error!("Error accepting control connection: {}", e),
+            }
+        }
+    }
 
+    fn try_get_client(server: &mut websocket::Server)
+        -> io::Result<WebSocketClient>
+    {
+        let connection = try!(server.accept());
+        let request = try!(connection.read_request());
+        match request.accept().send() {
+            Ok(client) => Ok(client),
             Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)),
         }
     }
-}
 
-pub struct ControlStream {
-    stream: TcpStream,
-}
+    fn receiver_loop(
+        mut receiver: WebSocketReceiver, tx: mpsc::Sender<ControlRequest>)
+    {
+        for message_result in receiver.incoming_messages() {
+            let message: websocket::Message = match message_result {
+                Ok(message) => message,
+                Err(e) => {
+                    warn!("Error receiving control message {}", e);
+                    continue;
+                },
+            };
+            let payload = match message.opcode {
+                websocket::message::Type::Text =>
+                    String::from_utf8(message.payload.into_owned()).unwrap(),
 
-impl ControlStream {
-    pub fn new(stream: TcpStream) -> Self {
-        ControlStream {
-            stream: stream,
+                code => {
+                    warn!("Unhandled websocket message type: {:?}", code);
+                    continue;
+                },
+            };
+            let control_request = json::decode(&payload).unwrap();
+            tx.send(control_request);
         }
     }
 
-    pub fn read_request(&mut self) -> io::Result<Option<ControlRequest>> {
-        Ok(None)
-    }
-
-    pub fn write_response(&mut self, response: &ControlResponse)
-        -> io::Result<()>
+    fn sender_loop(
+        mut sender: WebSocketSender, rx: &mut  mpsc::Receiver<ControlResponse>)
     {
-        response.write_to(&mut self.stream)
+        for control_response in rx.iter() {
+            let encoded = json::encode(&control_response).unwrap();
+            let message = websocket::Message::text(encoded);
+            sender.send_message(&message).unwrap();
+        }
     }
 }
 
+#[derive(RustcDecodable, RustcEncodable)]
+pub enum ControlRequest {
+    LoginRequest(LoginRequest),
+}
+
+#[derive(RustcDecodable, RustcEncodable)]
+pub enum ControlResponse {
+    LoginResponse(LoginResponse),
+}
+
 #[derive(RustcDecodable, RustcEncodable)]
 pub struct LoginRequest {
     username: String,