de_connector: Flush buff before player leaves (#734)

crates/connector/src/game/buffer.rs

@@ -67,6 +67,19 @@ impl PlayerBuffer {
         }
     }
 
+    /// Builds packages from all buffered messages and removes the packages
+    /// from the buffer.
+    pub(super) fn build_all(&mut self) -> PlayerPackageIterator<'_> {
+        PlayerPackageIterator {
+            index: 0,
+            iterators: [
+                self.unreliable.build_all(),
+                self.unordered.build_all(),
+                self.semi_ordered.build_all(),
+            ],
+        }
+    }
+
     fn builder_mut(&mut self, reliability: Reliability) -> &mut PackageBuilder {
         match reliability {
             Reliability::Unreliable => &mut self.unreliable,

crates/connector/src/game/greceiver.rs

@@ -237,22 +237,32 @@ impl GameProcessor {
 
     /// Process disconnect message.
     async fn process_leave(&mut self, meta: MessageMeta) {
-        let Some(id) = self.state.remove(meta.source).await else {
+        let Some(mut player_state) = self.state.remove(meta.source).await else {
             warn!("Tried to remove non-existent player {:?}.", meta.source);
             return;
         };
 
         self.clients.free(meta.source).await;
 
         info!(
-            "Player {id} on {:?} just left game on port {}.",
-            meta.source, self.port
+            "Player {} on {:?} just left game on port {}.",
+            player_state.id(),
+            meta.source,
+            self.port
         );
 
+        for output in player_state.buffer_mut().build_all() {
+            let _ = self.outputs.send(output).await;
+        }
+
         self.send(&FromGame::Left, Reliability::SemiOrdered, meta.source)
             .await;
-        self.send_all(&FromGame::PeerLeft(id), Reliability::SemiOrdered, None)
-            .await;
+        self.send_all(
+            &FromGame::PeerLeft(player_state.id()),
+            Reliability::SemiOrdered,
+            None,
+        )
+        .await;
     }
 
     async fn process_readiness(&mut self, meta: MessageMeta, readiness: Readiness) {

crates/connector/src/game/state.rs

@@ -47,9 +47,9 @@ impl GameState {
         self.inner.write().await.add(addr)
     }
 
-    /// Removes a single player from the game. It returns ID of the player if
-    /// the player was part of the game or None otherwise.
-    pub(super) async fn remove(&mut self, addr: SocketAddr) -> Option<Player> {
+    /// Removes a single player from the game. It returns state object of the
+    /// player if the player was part of the game or None otherwise.
+    pub(super) async fn remove(&mut self, addr: SocketAddr) -> Option<PlayerSlot> {
         self.inner.write().await.remove(addr)
     }
 
@@ -140,11 +140,11 @@ impl GameStateInner {
         }
     }
 
-    fn remove(&mut self, addr: SocketAddr) -> Option<Player> {
+    fn remove(&mut self, addr: SocketAddr) -> Option<PlayerSlot> {
         match self.players.remove_entry(&addr) {
             Some((_, player)) => {
                 self.available_ids.release(player.id);
-                Some(player.id)
+                Some(player)
             }
             None => None,
         }
@@ -272,7 +272,7 @@ pub(super) enum ReadinessUpdateError {
     Desync { game: Readiness, client: Readiness },
 }
 
-struct PlayerSlot {
+pub(super) struct PlayerSlot {
     id: Player,
     readiness: Readiness,
     buffer: PlayerBuffer,
@@ -286,6 +286,14 @@ impl PlayerSlot {
             buffer: PlayerBuffer::new(addr),
         }
     }
+
+    pub(super) fn id(&self) -> Player {
+        self.id
+    }
+
+    pub(super) fn buffer_mut(&mut self) -> &mut PlayerBuffer {
+        &mut self.buffer
+    }
 }
 
 #[cfg(test)]
@@ -314,6 +322,7 @@ mod tests {
                     .remove("127.0.0.1:1001".parse().unwrap())
                     .await
                     .unwrap()
+                    .id()
             ));
             assert!(!state.contains("127.0.0.1:1001".parse().unwrap()).await);
             assert!(state.contains("127.0.0.1:1002".parse().unwrap()).await);