migrate to tokio

src/main.rs

@@ -1,53 +1,72 @@
-use std::net::TcpListener;
-use std::thread;
-use std::sync::{Arc, Mutex};
-use std::collections::HashMap;
+use std::{
+    sync::{Arc, Mutex},
+    io::Error as IoError,
+    collections::HashMap,
+};
+
+use tokio::net::{TcpListener};
 
 mod message;
-use message::Message;
+use message::{Message, MessageWebSocket};
 mod code_generator;
 use code_generator::CodeGenerator;
-mod websocket;
-use websocket::WebsocketWrapper;
 
-fn main() {
-    let code_generator = Arc::new(Mutex::new(CodeGenerator::default()));
-    let server = TcpListener::bind("127.0.0.1:8080").unwrap();
-
-    let rooms = Arc::new(Mutex::new(HashMap::new()));
+#[tokio::main]
+async fn main() -> Result<(), IoError> {
+    let global_state = Arc::new(GlobalState::default());
     
-    for stream in server.incoming() {
-        let code_generator = Arc::clone(&code_generator);
-        let rooms = Arc::clone(&rooms);
-        thread::spawn (move || {
-            let ws = tungstenite::accept(stream.unwrap()).unwrap();
-            let mut ws = WebsocketWrapper::new(ws);
-            
-            ws.send(msg!("HOSTJOIN"));
-            let msg = ws.recv();
-            match msg {                
-                None => (),
-                Some(msg) => match msg.command.as_str() {
-                    "HOST" => {
-                        let code = code_generator.lock().unwrap().generate();
-                        let mut room = Room::default();
-                        let player = Player {};
-                        room.players.push(player);
-                        rooms.lock().unwrap().insert(code.clone(), room);
-                        ws.send(msg!("CODE", code));
-                    }
-                    _ => unimplemented!(),
-                }
-            }
+    let socket = TcpListener::bind("127.0.0.1:8080").await;
+    let listener = socket.expect("Could not bind to localhost:8080");
+    println!("Server running");
 
+    while let Ok((stream, addr)) = listener.accept().await {
+        let global_state = Arc::clone(&global_state);
+        let mut local_state = LocalState::default();
+        tokio::spawn(async move {
+            // Upgrade to a WS connection
+            let mut ws = MessageWebSocket(tokio_tungstenite::accept_async(stream)
+                                          .await
+                                          .expect("Could not establish connection"));
+            
+            println!("Connected to {}", addr);
+
+            while let Ok(msg) = ws.next().await {
+                dispatch(msg, &global_state, &mut local_state, &mut ws).await;
+            }
         });
     }
+    Ok(())
+}
+
+async fn dispatch(
+    msg: Message,
+    global_state: &GlobalState,
+    local_state: &mut LocalState,
+    ws: &mut MessageWebSocket
+) {
+    match msg.command.as_str() {
+        "HOST" => {
+            let room_code = global_state.code_generator.lock().unwrap().generate();
+            let game_controller = GameController::default();
+            global_state.rooms.lock().unwrap().insert(room_code, game_controller);
+        }
+        "JOIN" => {
+            let room_code = &msg.args[0];
+            let room = global_state.rooms.lock().unwrap().get(room_code);
+        }
+        _ => unimplemented!(),
+    }
 }
 
 #[derive(Default)]
-struct Room {
-    players: Vec<Player>,
+struct GlobalState {
+    code_generator: Arc<Mutex<CodeGenerator>>,
+    rooms: Arc<Mutex<HashMap<String, GameController>>>,
 }
 
-struct Player {
+#[derive(Default)]
+struct GameController {} // TODO
+
+#[derive(Default)]
+struct LocalState {
 }

src/message.rs

@@ -1,3 +1,12 @@
+use std::convert::{From, TryFrom};
+
+use tokio::net::TcpStream;
+use tokio_tungstenite::{
+    WebSocketStream,
+    tungstenite::Message as WsMessage,
+};
+use futures::{StreamExt, SinkExt};
+
 #[derive(PartialEq, Debug)]
 pub struct Message {
     pub command: String,
@@ -10,6 +19,8 @@ pub type Result<T> = std::result::Result<T, Error>;
 #[derive(PartialEq, Debug)]
 pub enum Error {
     BadParse,
+    NonText,
+    Unknown,
 }
 
 impl Message {
@@ -33,12 +44,49 @@ impl Message {
     }
 }
 
+impl TryFrom<WsMessage> for Message {
+    type Error = Error;
+
+    fn try_from(ws_message: WsMessage) -> Result<Self> {
+        let text = match ws_message {
+            WsMessage::Text(text) => text,
+            _ => return Err(Error::NonText)
+        };
+        Message::parse(text)
+    }
+}
+
+impl From<Message> for WsMessage {
+    fn from(message: Message) -> Self {
+        WsMessage::Text(message.to_string())
+    }
+}
+
+pub struct MessageWebSocket(pub WebSocketStream<TcpStream>);
+
+impl MessageWebSocket {
+    pub async fn next(&mut self) -> Result<Message> {
+        if let Some(Ok(msg)) = self.0.next().await {
+            msg.try_into()
+        } else {
+            Err(Error::Unknown)
+        }
+    }
+    pub async fn send(&mut self, msg: Message) -> Result<()> {
+        if self.0.send(msg.into()).await.is_err() {
+            Err(Error::Unknown)
+        } else {
+            Ok(())
+        }
+    }
+}
+
 #[macro_export]
 macro_rules! msg {
     ( $command:expr) => {
 	{
 	    let command = $command.to_string();
-	    let args = Vec::new();
+	    let args = vec![];
 	    Message { command, args }
 	}
     };
@@ -62,7 +110,6 @@ impl std::fmt::Display for Message {
     }
 }
 
-
 #[cfg(test)]
 mod test {
     use super::*;

src/websocket.rs

@@ -1,36 +0,0 @@
-use std::net::TcpStream;
-
-use tungstenite::protocol::{ WebSocket, Message as WsMessage };
-
-use crate::message::Message;
-
-pub struct WebsocketWrapper {
-    websocket: WebSocket<TcpStream>,
-}
-
-impl WebsocketWrapper {
-    pub fn new(websocket: WebSocket<TcpStream>) -> Self {
-        WebsocketWrapper { websocket }
-    }
-    pub fn send(&mut self, msg: Message) {
-        self.websocket.write_message(WsMessage::Text(msg.to_string())).unwrap();
-    }
-    pub fn recv(&mut self) -> Option<Message> {
-        match self.websocket.read_message() {
-            Err(_) | Ok(WsMessage::Close(_)) => None,
-            Ok(WsMessage::Text(text)) => match Message::parse(text) {
-                Ok(msg) => Some(msg),
-                Err(_) => {
-                    self.websocket.write_message(WsMessage::Text("ERROR: bad_format".to_string())).unwrap();
-                    self.recv()
-                }
-            }
-            _ => {
-                self.websocket.write_message(WsMessage::Text("ERROR: bad_command".to_string())).unwrap();
-                self.recv()
-            }
-        }
-    }
-}
-        
-