and you get a task, and you get a task, and YOU get a task, and you get a task..

Cargo.lock

@@ -133,9 +133,9 @@ dependencies = [
 
 [[package]]
 name = "async-compression"
-version = "0.4.13"
+version = "0.4.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e614738943d3f68c628ae3dbce7c3daffb196665f82f8c8ea6b65de73c79429"
+checksum = "998282f8f49ccd6116b0ed8a4de0fbd3151697920e7c7533416d6e25e76434a7"
 dependencies = [
  "brotli",
  "flate2",

server/src/game.rs

@@ -9,6 +9,9 @@ use std::{
 };
 use uuid::Uuid;
 
+// Game logic
+// NOTE: Don't try to send messages from here, that's the job of the associated handler
+
 /// Internal manifest for making a new game
 #[derive(Debug)]
 pub struct NewGameManifest {

server/src/game_handler.rs

@@ -119,10 +119,14 @@ impl GameHandler {
                                 tx.send(DmUser(SendChatMessage(message), Addr(addr))).await
                             {
                                 tracing::error!("Could not send message: {}", e);
-                            };
+                            }
                         })
                     }
-                    Ok(None) => tokio::spawn(async move { tracing::debug!("TODO") }),
+                    Ok(None) => {
+                        tokio::spawn(
+                            async move { tracing::debug!("TODO: whatever i'm supposed to do") },
+                        )
+                    }
                     Ok(Some((judge_round, czar_id))) => {
                         let tx = self.state.users_tx.clone();
 
@@ -132,7 +136,7 @@ impl GameHandler {
                                 .await
                             {
                                 tracing::error!("Could not send message: {}", e);
-                            };
+                            }
                         })
                     }
                 };
@@ -392,22 +396,22 @@ impl GameHandler {
             self.send_game_meta_update(vec![game_id]);
 
             // Broadcast game browser update
-            if let Err(e) = self
-                .state
-                .broadcast_tx
-                .send(meta_games_browser_update(&self.state))
-            {
+            let msg = meta_games_browser_update(&self.state);
+            let tx = self.state.broadcast_tx.clone();
+            tokio::spawn(async move {
+                if let Err(e) = tx.send(msg) {
                     tracing::error!("Could not broadcast game browser update: {}", e);
-            };
+                }
+            });
 
             // Broadcast server meta update
-            if let Err(e) = self
-                .state
-                .broadcast_tx
-                .send(meta_server_summary_update(&self.state))
-            {
+            let msg = meta_server_summary_update(&self.state);
+            let tx = self.state.broadcast_tx.clone();
+            tokio::spawn(async move {
+                if let Err(e) = tx.send(msg) {
                     tracing::error!("Could not broadcast server meta update: {}", e);
-            };
+                }
+            });
         } else {
             tracing::error!("Attempted to create game for nonexistent player!");
         }

server/src/message_handler.rs

@@ -33,36 +33,45 @@ impl MessageHandler {
                         msg = format! {"{0}: {1}", self.state.online_users.read().unwrap().get(&addr).unwrap().read().unwrap().name, chat_message.text};
                     }
 
-                    if let Err(e) = self
-                        .state
-                        .broadcast_tx
-                        .send(to_string::<ChatMessage>(&ChatMessage { text: msg }).unwrap())
-                    {
+                    // Broadcast incoming chat message
+                    let broadcast_message =
+                        to_string::<ChatMessage>(&ChatMessage { text: msg }).unwrap();
+                    let tx = self.state.broadcast_tx.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = tx.send(broadcast_message) {
                             tracing::error!("Error broadcasting Chat message: {}", e)
-                    };
+                        }
+                    });
                 }
                 _user_log_in_request
                     if let Ok(user_log_in) = from_str::<UserLogInRequest>(&text) =>
                 {
-                    if let Err(e) = self.state.users_tx.send(UserLogIn(user_log_in, addr)).await {
+                    let msg = UserLogIn(user_log_in, addr);
+                    let tx = self.state.users_tx.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = tx.send(msg).await {
                             tracing::error!("Error sending user login: {}", e)
-                    };
+                        }
+                    });
                 }
                 _new_game_request if let Ok(new_game) = from_str::<NewGameRequest>(&text) => {
-                    if let Err(e) = self.state.games_tx.send(NewGame(new_game, addr)).await {
+                    let msg = NewGame(new_game, addr);
+                    let tx = self.state.games_tx.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = tx.send(msg).await {
                             tracing::error!("Error requesting new game: {}", e)
-                    };
+                        }
+                    });
                 }
 
                 _join_game_request if let Ok(join_request) = from_str::<GameJoinRequest>(&text) => {
-                    if let Err(e) = self
-                        .state
-                        .games_tx
-                        .send(JoinGame(join_request.id, addr))
-                        .await
-                    {
+                    let msg = JoinGame(join_request.id, addr);
+                    let tx = self.state.games_tx.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = tx.send(msg).await {
                             tracing::error!("Error requesting game join: {}", e)
-                    };
+                        }
+                    });
                 }
 
                 _player_move_request
@@ -76,14 +85,13 @@ impl MessageHandler {
                         tracing::error!("Move request game_id is empty! Ignoring...");
                         return;
                     } else {
-                        if let Err(e) = self
-                            .state
-                            .games_tx
-                            .send(MoveRequest(move_request, addr))
-                            .await
-                        {
+                        let msg = MoveRequest(move_request, addr);
+                        let tx = self.state.games_tx.clone();
+                        tokio::spawn(async move {
+                            if let Err(e) = tx.send(msg).await {
                                 tracing::error!("Error sending move request: {}", e)
-                        };
+                            }
+                        });
                     }
                 }
 
@@ -91,23 +99,26 @@ impl MessageHandler {
                     if let Ok(judge_request) = from_str::<JudgeDecisionRequest>(&text) =>
                 {
                     if !judge_request.winning_cards.is_empty() {
-                        if let Err(e) = self
-                            .state
-                            .games_tx
-                            .send(JudgeDecision(judge_request, addr))
-                            .await
-                        {
+                        let msg = JudgeDecision(judge_request, addr);
+                        let tx = self.state.games_tx.clone();
+                        tokio::spawn(async move {
+                            if let Err(e) = tx.send(msg).await {
                                 tracing::error!("Error sending Judge Decision: {}", e)
-                        };
+                            }
+                        });
                     } else {
                         tracing::error!("Judge request received with empty cards");
                     }
                 }
 
                 _delete_game if let Ok(delete_request) = from_str::<GameDeleteRequest>(&text) => {
-                    if let Err(e) = self.state.games_tx.send(DeleteGame(delete_request)).await {
+                    let msg = DeleteGame(delete_request);
+                    let tx = self.state.games_tx.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = tx.send(msg).await {
                             tracing::error!("Error sending delete game: {}", e)
-                    };
+                        }
+                    });
                 }
 
                 _ => tracing::error!(
@@ -117,7 +128,11 @@ impl MessageHandler {
                 ),
             },
 
-            Message::Binary(data) => tracing::error!("{} sent binary: {:?}\n TODO: Probably close connection here", addr, data),
+            Message::Binary(data) => tracing::error!(
+                "{} sent binary: {:?}\n TODO: Probably close connection here",
+                addr,
+                data
+            ),
 
             Message::Close(close_frame) => {
                 self.handle_close(close_frame, addr);
@@ -135,6 +150,7 @@ impl MessageHandler {
 
     /// This runs when a connection closes
     fn handle_close(&self, close_frame: Option<CloseFrame>, addr: SocketAddr) {
+        // TODO: this is user handler's job
         let user_name = self
             .state
             .online_users
@@ -151,13 +167,13 @@ impl MessageHandler {
         let msg = ChatMessage {
             text: format!("{0} left.", &user_name),
         };
-        if let Err(e) = self
-            .state
-            .broadcast_tx
-            .send(to_string::<ChatMessage>(&msg).unwrap())
-        {
+        let chat_message = to_string::<ChatMessage>(&msg).unwrap();
+        let tx = self.state.broadcast_tx.clone();
+        tokio::spawn(async move {
+            if let Err(e) = tx.send(chat_message) {
                 tracing::error!("Error broadcasting user leave message: {}", e)
-        };
+            }
+        });
 
         // Process close frame
         if let Some(cf) = close_frame {
@@ -172,6 +188,7 @@ impl MessageHandler {
         }
 
         // Move user to offline
+        // TODO: This is user handler's job
         self.state.offline_users.write().unwrap().insert(
             user_name,
             self.state
@@ -183,15 +200,19 @@ impl MessageHandler {
         );
 
         // Send client updates
-        if let Err(e) = self
-            .state
-            .broadcast_tx
-            .send(meta_server_summary_update(&self.state))
-        {
+        let msg = meta_server_summary_update(&self.state);
+        let tx = self.state.broadcast_tx.clone();
+        tokio::spawn(async move {
+            if let Err(e) = tx.send(msg) {
                 tracing::error!("Error broadcasting server summary update: {}", e)
-        };
-        if let Err(e) = self.state.broadcast_tx.send(meta_chat_update(&self.state)) {
+            }
+        });
+        let msg = meta_chat_update(&self.state);
+        let tx = self.state.broadcast_tx.clone();
+        tokio::spawn(async move {
+            if let Err(e) = tx.send(msg) {
                 tracing::error!("Error broadcasting chat update: {}", e)
-        };
+            }
+        });
     }
 }

server/src/user_handler.rs

@@ -8,6 +8,8 @@ use serde_json::to_string;
 use std::net::SocketAddr;
 use std::sync::{Arc, RwLock};
 
+// TODO: clean up all this tx/msg mess
+
 /// Handles users
 pub struct UserHandler {
     /// Pointer to global state
@@ -83,32 +85,40 @@ impl UserHandler {
         match message {
             SendUserUpdate(message) => {
                 let msg = to_string::<UserUpdate>(&message).unwrap();
+                tokio::spawn(async move {
                     if let Err(e) = tx.send(msg).await {
                         tracing::error!("Error sending user update: {}", e)
-                };
+                    }
+                });
             }
             SendChatMessage(message) => {
                 let msg = to_string::<ChatMessage>(&message).unwrap();
+                tokio::spawn(async move {
                     if let Err(e) = tx.send(msg).await {
                         tracing::error!("Error sending chat message: {}", e)
-                };
+                    }
+                });
             }
             SendJudgeRound(message) => {
                 let msg = to_string::<JudgeRound>(&message).unwrap();
+                tokio::spawn(async move {
                     if let Err(e) = tx.send(msg).await {
                         tracing::error!("Error sending judge round: {}", e)
-                };
+                    }
+                });
             }
         }
     }
 
     /// Create, register, and hydrate new user
     async fn set_up_new_user(&self, user: User, addr: SocketAddr) {
-        let tx = user.tx.clone();
+        let user_tx = user.tx.clone();
         let new_user = Arc::new(RwLock::new(user));
 
         // Notify client of new username
-        let _ = tx.send(user_client_self_update(&new_user)).await;
+        let tx = user_tx.clone();
+        let msg = user_client_self_update(&new_user);
+        tokio::spawn(async move { tx.send(msg).await });
 
         // Register uuid
         self.state
@@ -125,23 +135,29 @@ impl UserHandler {
             .insert(addr, new_user.clone());
 
         // Hydrate client
-        let _ = tx.send(meta_games_browser_update(&self.state)).await;
-        let _ = tx.send(meta_new_game_card_packs(&self.state)).await;
+        let tx = user_tx.clone();
+        let msg = meta_games_browser_update(&self.state);
+        tokio::spawn(async move { tx.send(msg).await });
+        let tx = user_tx.clone();
+        let msg = meta_new_game_card_packs(&self.state);
+        tokio::spawn(async move { tx.send(msg).await });
 
         // Broadcast new user's existence
         // TODO: this should probably be combined and sent as one
-        let _ = &self
-            .state
-            .broadcast_tx
-            .send(meta_announce_user_join(&self.state, &addr));
-        let _ = &self
-            .state
-            .broadcast_tx
-            .send(meta_server_summary_update(&self.state));
-        let _ = &self.state.broadcast_tx.send(meta_chat_update(&self.state));
+        let tx = self.state.broadcast_tx.clone();
+        let msg = meta_announce_user_join(&self.state, &addr);
+        tokio::spawn(async move { tx.send(msg) });
+
+        let tx = self.state.broadcast_tx.clone();
+        let msg = meta_server_summary_update(&self.state);
+        tokio::spawn(async move { tx.send(msg) });
+        let tx = self.state.broadcast_tx.clone();
+        let msg = meta_chat_update(&self.state);
+        tokio::spawn(async move { tx.send(msg) });
 
         // TODO: this races the broadcasts but if it's done last it'll probably show up last...
-        let _ = tx.send(meta_motd()).await;
+        let tx = user_tx.clone();
+        tokio::spawn(async move { tx.send(meta_motd()).await });
     }
 
     /// Handle user login
@@ -199,7 +215,9 @@ impl UserHandler {
                 old_name,
                 new_name
             };
-            let _ = broadcast_tx.send(to_string(&ChatMessage { text: msg }).unwrap());
+            let tx = broadcast_tx.clone();
+            let msg = to_string(&ChatMessage { text: msg }).unwrap();
+            tokio::spawn(async move { tx.send(msg) });
         }
         // Check if name is taken by an online user
         else if self
@@ -245,8 +263,9 @@ impl UserHandler {
                 old_name,
                 new_name
             };
-            let _ =
-                broadcast_tx.send(to_string::<ChatMessage>(&ChatMessage { text: msg }).unwrap());
+            let chat_message = to_string::<ChatMessage>(&ChatMessage { text: msg }).unwrap();
+            let tx = broadcast_tx.clone();
+            tokio::spawn(async move { tx.send(chat_message) });
         }
 
         // Send the user their new name
@@ -279,15 +298,17 @@ impl UserHandler {
                     );
                     let tx = self.state.games_tx.clone();
 
-                    tokio::spawn(async move {
-                        let _ = tx.send(msg).await;
-                    });
+                    tokio::spawn(async move { tx.send(msg).await });
                 }
             }
         }
         // Send client updates
-        let _ = broadcast_tx.send(meta_games_browser_update(&self.state));
-        let _ = broadcast_tx.send(meta_chat_update(&self.state));
+        let tx = broadcast_tx.clone();
+        let msg = meta_games_browser_update(&self.state);
+        tokio::spawn(async move { tx.send(msg) });
+        let tx = broadcast_tx.clone();
+        let msg = meta_chat_update(&self.state);
+        tokio::spawn(async move { tx.send(msg) });
     }
 }