Move crate::play to crate::app::socket

4 files changed, 57 insertions, 47 deletions

diff --git a/src/app.rs b/src/app.rs
index 43e2ac6..c04bac3 100644
--- a/src/app.rs
+++ b/src/app.rs
@@ -4,6 +4,8 @@ use crate::session::Session;
 use std::collections::HashMap;
 use std::sync::{Mutex, MutexGuard, RwLock};
 
+pub mod socket;
+
 /// Provider of the state that Tabletop Ambulator needs to keep track of.
 ///
 /// Provides the app state, including runtime data such as the active game
diff --git a/src/play.rs b/src/app/socket.rs
index 0ec9758..08e1db6 100644
--- a/src/play.rs
+++ b/src/app/socket.rs
@@ -1,54 +1,46 @@
 use crate::AppState;
 use crate::session::{HandObject, PlayerColor};
-use axum::extract::ws::{Message, Utf8Bytes, WebSocket};
-use futures_util::{SinkExt, StreamExt};
+use axum::extract::ws::{Message, Utf8Bytes};
+use futures_util::{Sink, SinkExt, Stream, StreamExt};
 use serde::{Deserialize, Serialize};
+use std::fmt::Debug;
 use std::mem;
 use std::sync::{Arc, Mutex};
 use tokio::sync::{broadcast, mpsc, oneshot};
 
+// TODO: Use in OutgoingPlayMessage::Error, then remove allow(dead_code)
+#[allow(dead_code)]
+pub enum Error {
+    BadJson(serde_json::Error),
+    Closed,
+    InvalidSession(String),
+    InvalidColor,
+}
+
+#[derive(Clone)]
+pub enum PlayUpdate {
+    HandUpdate([Vec<HandObject>; PlayerColor::COUNT]),
+}
+
 #[derive(Deserialize)]
-enum IncomingPlayMessage {
+enum IncomingMessage {
     Initialize { id: String },
     Color(String),
 }
 
 // TODO: Maybe derive Clone, reference interior vals
 #[derive(Serialize)]
-enum OutgoingPlayMessage {
+enum OutgoingMessage {
     Initialize { colors: Vec<String> },
     Hand(Vec<HandObject>),
     Error,
 }
 
-#[derive(Clone)]
-pub enum PlayUpdate {
-    HandUpdate([Vec<HandObject>; PlayerColor::COUNT]),
-}
-
-// TODO: Use in OutgoingPlayMessage::Error, then remove allow(dead_code)
-#[allow(dead_code)]
-pub enum Error {
-    BadJson(serde_json::Error),
-    Closed,
-    InvalidSession(String),
-    InvalidColor,
-}
-
 struct PlayState {
     session: Option<String>,
     color: PlayerColor,
     update_cancel_tx: oneshot::Sender<()>,
 }
-impl PlayState {
-    pub fn new(update_cancel_tx: oneshot::Sender<()>) -> Self {
-        Self {
-            session: Default::default(),
-            color: Default::default(),
-            update_cancel_tx,
-        }
-    }
-}
 
 impl From<serde_json::Error> for Error {
     fn from(value: serde_json::Error) -> Self {
@@ -56,14 +48,27 @@ impl From<serde_json::Error> for Error {
     }
 }
 
-impl From<mpsc::error::SendError<OutgoingPlayMessage>> for Error {
-    fn from(_: mpsc::error::SendError<OutgoingPlayMessage>) -> Self {
+impl From<mpsc::error::SendError<OutgoingMessage>> for Error {
+    fn from(_: mpsc::error::SendError<OutgoingMessage>) -> Self {
         Self::Closed
     }
 }
 
-pub async fn handle_play(socket: WebSocket, app_state: Arc<AppState>) {
-    let (mut sender, mut receiver) = socket.split();
+impl PlayState {
+    pub fn new(update_cancel_tx: oneshot::Sender<()>) -> Self {
+        Self {
+            session: Default::default(),
+            color: Default::default(),
+            update_cancel_tx,
+        }
+    }
+}
+
+pub async fn handle_play<S, R>(mut sender: S, mut receiver: R, app_state: Arc<AppState>)
+where
+    S: Sink<Message, Error: Debug> + Unpin + Send + 'static,
+    R: Stream<Item = Result<Message, axum::Error>> + Unpin + Send + 'static,
+{
     let (sender_tx, mut sender_rx) = mpsc::channel(2);
 
     let (update_cancel_tx, _) = oneshot::channel();
@@ -82,7 +87,7 @@ pub async fn handle_play(socket: WebSocket, app_state: Arc<AppState>) {
                 .send(Message::Text(Utf8Bytes::from(serialized)))
                 .await
             {
-                eprintln!("Failed to send serialized websocket message: {}", err);
+                eprintln!("Failed to send serialized websocket message: {:?}", err);
                 break;
             }
         }
@@ -106,7 +111,7 @@ pub async fn handle_play(socket: WebSocket, app_state: Arc<AppState>) {
                                 break;
                             }
                             Err(_) => {
-                                let result = sender_tx.send(OutgoingPlayMessage::Error).await;
+                                let result = sender_tx.send(OutgoingMessage::Error).await;
                                 if let Err(err) = result {
                                     eprintln!(
                                         "Failed to send play message as the channel closed: {}",
@@ -119,7 +124,7 @@ pub async fn handle_play(socket: WebSocket, app_state: Arc<AppState>) {
                     }
                     Err(_) => {
                         // TODO: include error details
-                        let result = sender_tx.send(OutgoingPlayMessage::Error).await;
+                        let result = sender_tx.send(OutgoingMessage::Error).await;
                         if let Err(err) = result {
                             eprintln!("Failed to send play message as the channel closed: {}", err);
                             break;
@@ -137,13 +142,13 @@ pub async fn handle_play(socket: WebSocket, app_state: Arc<AppState>) {
 }
 
 async fn handle_play_message(
-    message: IncomingPlayMessage,
-    sender_tx: mpsc::Sender<OutgoingPlayMessage>,
+    message: IncomingMessage,
+    sender_tx: mpsc::Sender<OutgoingMessage>,
     state: &Arc<Mutex<PlayState>>,
     app_state: &Arc<AppState>,
 ) -> Result<(), Error> {
     match message {
-        IncomingPlayMessage::Initialize { id } => {
+        IncomingMessage::Initialize { id } => {
             let data_opt = app_state.with_session(id.as_str(), |session| {
                 let colors: Vec<String> = session
                     .seats
@@ -181,11 +186,11 @@ async fn handle_play_message(
             }
 
             sender_tx
-                .send(OutgoingPlayMessage::Initialize { colors })
+                .send(OutgoingMessage::Initialize { colors })
                 .await
                 .map_err(Error::from)
         }
-        IncomingPlayMessage::Color(color) => {
+        IncomingMessage::Color(color) => {
             let hand = {
                 let mut state = state.lock().unwrap();
                 state.color =
@@ -202,7 +207,7 @@ async fn handle_play_message(
             };
 
             sender_tx
-                .send(OutgoingPlayMessage::Hand(hand))
+                .send(OutgoingMessage::Hand(hand))
                 .await
                 .map_err(Error::from)
         }
@@ -212,7 +217,7 @@ async fn handle_play_message(
 async fn handle_update(
     mut update_rx: broadcast::Receiver<PlayUpdate>,
     mut cancel_rx: oneshot::Receiver<()>,
-    sender_tx: mpsc::Sender<OutgoingPlayMessage>,
+    sender_tx: mpsc::Sender<OutgoingMessage>,
     state: Arc<Mutex<PlayState>>,
 ) {
     loop {
@@ -225,7 +230,7 @@ async fn handle_update(
                         .map(|color| String::from(color.as_ref()))
                         .collect();
                     let _ = sender_tx
-                        .send(OutgoingPlayMessage::Initialize {
+                        .send(OutgoingMessage::Initialize {
                             colors,
                         })
                         .await;
@@ -233,7 +238,7 @@ async fn handle_update(
                         let color = &state.lock().unwrap().color;
                         hands[color].to_owned()
                     };
-                    let _ = sender_tx.send(OutgoingPlayMessage::Hand(hand)).await;
+                    let _ = sender_tx.send(OutgoingMessage::Hand(hand)).await;
                 }
                 Err(broadcast::error::RecvError::Closed) => break,
                 Err(broadcast::error::RecvError::Lagged(_)) => continue,
diff --git a/src/main.rs b/src/main.rs
index 38bbbf2..a4e51c1 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -7,20 +7,20 @@
 #![warn(missing_docs, missing_debug_implementations)]
 
 mod app;
-mod play;
 mod session;
 mod template;
 
 use crate::app::AppState;
-use crate::play::handle_play;
 use crate::session::{HandObject, PlayerColor};
 use crate::template::{IndexTemplate, SessionTemplate};
+use app::socket::handle_play;
 use askama::Template;
 use axum::extract::{Path, Query, State, WebSocketUpgrade};
 use axum::http::{StatusCode, header};
 use axum::response::{ErrorResponse, Html, IntoResponse, Redirect, Response};
 use axum::routing::{any, get, put};
 use axum::{Json, Router};
+use futures_util::StreamExt;
 use rust_embed::Embed;
 use std::array;
 use std::collections::HashMap;
@@ -132,5 +132,8 @@ async fn update_hands(
 }
 
 async fn upgrade_play(ws: WebSocketUpgrade, State(state): State<Arc<AppState>>) -> Response {
-    ws.on_upgrade(|socket| handle_play(socket, state))
+    ws.on_upgrade(|socket| {
+        let (sender, receiver) = socket.split();
+        handle_play(sender, receiver, state)
+    })
 }
diff --git a/src/session.rs b/src/session.rs
index c15bfda..246551e 100644
--- a/src/session.rs
+++ b/src/session.rs
@@ -1,4 +1,4 @@
-use crate::play::PlayUpdate;
+use crate::app::socket::PlayUpdate;
 use serde::{Deserialize, Serialize};
 use std::array;
 use std::ops::Index;