Build out skeleton of server message processing flow

src/main.rs

@@ -4,7 +4,10 @@ extern crate log;
 use anyhow::{Context, Result};
 use future::ready;
 use futures::prelude::*;
-use net::{server::greet, ClientMessage, ServerMessage};
+use net::{
+    server::{Handle, Server, run_client},
+    ClientMessage, ServerMessage,
+};
 use serde_json::{from_str, to_string};
 use std::net::ToSocketAddrs;
 use stream::FuturesUnordered;
@@ -28,11 +31,15 @@ async fn main() -> Result<()> {
     env_logger::init();
     let args = Args::from_args();
 
+    // create singleton server (use weak-table in the future for multiple rooms)
+    let game_server = Server::create();
+
     // dispatch websockets
-    let socket_handler = ws().map(|upgrade: Ws| {
+    let socket_handler = ws().map(move |upgrade: Ws| {
-        upgrade.on_upgrade(|ws| {
+        let handle = game_server.clone();
+        upgrade.on_upgrade(move |ws| {
             async {
-                if let Err(error) = handle_socket(ws).await {
+                if let Err(error) = handle_socket(handle, ws).await {
                     warn!("Websocket connection lost: {:#}", error);
                 }
             }
@@ -45,7 +52,7 @@ async fn main() -> Result<()> {
         .context("Couldn't parse the listen address")?;
     let servers = FuturesUnordered::new();
     for addr in addrs {
-        let (_, server) = serve(socket_handler).try_bind_ephemeral(addr)?;
+        let (_, server) = serve(socket_handler.clone()).try_bind_ephemeral(addr)?;
         servers.push(server);
     }
 
@@ -54,7 +61,7 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
-async fn handle_socket(websocket: WebSocket) -> Result<()> {
+async fn handle_socket(game_server: Handle, websocket: WebSocket) -> Result<()> {
     let (sink, source) = websocket.split();
 
     let mut sink = sink.with(|msg: ServerMessage| {
@@ -77,5 +84,5 @@ async fn handle_socket(websocket: WebSocket) -> Result<()> {
         })
     });
 
-    run_client(&mut source, &mut sink).await
+    run_client(game_server, &mut source, &mut sink).await
 }

src/net/server.rs

@@ -25,9 +25,31 @@ where
         .context("Greeting client")
 }
 
+pub struct Server {
+}
+
+#[derive(Clone)]
+pub struct Handle {
+    server: Arc<Mutex<Server>>,
+}
+
+impl Server {
+    pub fn create() -> Handle {
+        Handle {
+            server: Arc::new(Mutex::new(Server {
+            })),
+        }
+    }
+
+    pub fn process_message(&mut self, player: usize, msg: ClientMessage) {
+        debug!("Client#{} message: {:?}", player, &msg);
+    }
+}
 where
 {
+
 pub async fn run_client(
+    handle: Handle,
     source: &mut (impl Stream<Item = Result<ClientMessage, Error>> + Send + Unpin),
     mut sink: &mut (impl Sink<ServerMessage, Error = Error> + Send + Unpin),
 ) -> Result<()> {
@@ -39,7 +61,7 @@ pub async fn run_client(
     let input_task = async {
         loop {
             match source.next().await {
-                Some(Ok(msg)) => debug!("Client message: {:?}", &msg),
+                Some(Ok(msg)) => handle.server.lock().await.process_message(0, msg),
                 Some(Err(error)) => return Err(error),
                 None => break Ok(()),
             }