make writing packets work synchronously*

azalea-client/src/chat.rs

@@ -95,7 +95,7 @@ impl Client {
     /// not the command packet. The [`Client::chat`] function handles checking
     /// whether the message is a command and using the proper packet for you,
     /// so you should use that instead.
-    pub async fn send_chat_packet(&self, message: &str) -> Result<(), std::io::Error> {
+    pub fn send_chat_packet(&self, message: &str) {
         // TODO: chat signing
         // let signature = sign_message();
         let packet = ServerboundChatPacket {
@@ -111,12 +111,11 @@ impl Client {
             last_seen_messages: LastSeenMessagesUpdate::default(),
         }
         .get();
-        self.write_packet(packet).await
     }
 
     /// Send a command packet to the server. The `command` argument should not
     /// include the slash at the front.
-    pub async fn send_command_packet(&self, command: &str) -> Result<(), std::io::Error> {
+    pub fn send_command_packet(&self, command: &str) {
         // TODO: chat signing
         let packet = ServerboundChatCommandPacket {
             command: command.to_string(),
@@ -131,7 +130,7 @@ impl Client {
             last_seen_messages: LastSeenMessagesUpdate::default(),
         }
         .get();
-        self.write_packet(packet).await
+        self.write_packet(packet);
     }
 
     /// Send a message in chat.
@@ -143,11 +142,11 @@ impl Client {
     /// # Ok(())
     /// # }
     /// ```
-    pub async fn chat(&self, message: &str) -> Result<(), std::io::Error> {
+    pub async fn chat(&self, message: &str) {
         if let Some(command) = message.strip_prefix('/') {
-            self.send_command_packet(command).await
+            self.send_command_packet(command);
         } else {
-            self.send_chat_packet(message).await
+            self.send_chat_packet(message);
         }
     }
 }

azalea-client/src/client.rs

@@ -213,10 +213,12 @@ impl Client {
 
         let world = client.world();
 
+        let (packet_writer_sender, packet_writer_receiver) = mpsc::unbounded_channel();
+
         let local_player = crate::local_player::LocalPlayer::new(
             entity,
             game_profile,
-            write_conn,
+            packet_writer_sender,
             world.clone(),
             ecs.resource_mut::<EntityInfos>().deref_mut(),
             tx,
@@ -228,7 +230,14 @@ impl Client {
             run_schedule_sender,
         };
 
-        tokio::spawn(packet_receiver.clone().read_task(read_conn));
+        let read_packets_task = tokio::spawn(packet_receiver.clone().read_task(read_conn));
+        let write_packets_task = tokio::spawn(
+            packet_receiver
+                .clone()
+                .write_task(write_conn, packet_writer_receiver),
+        );
+        client.tasks.lock().push(read_packets_task);
+        client.tasks.lock().push(write_packets_task);
 
         ecs.entity_mut(entity)
             .insert((local_player, packet_receiver));
@@ -357,25 +366,15 @@ impl Client {
     }
 
     /// Write a packet directly to the server.
-    pub async fn write_packet(&self, packet: ServerboundGamePacket) -> Result<(), std::io::Error> {
-        self.local_player_mut(&self.ecs.lock())
-            .write_packet_async(packet)
-            .await
+    pub fn write_packet(&self, packet: ServerboundGamePacket) {
+        self.local_player_mut(&self.ecs.lock()).write_packet(packet)
     }
 
-    /// Disconnect this client from the server, ending all tasks.
+    /// Disconnect this client from the server by ending all tasks.
+    ///
+    /// The OwnedReadHalf for the TCP connection is in one of the tasks, so it
+    /// automatically closes the connection when that's dropped.
     pub async fn disconnect(&self) -> Result<(), std::io::Error> {
-        if let Err(e) = self
-            .local_player_mut(&self.ecs.lock())
-            .write_conn
-            .shutdown()
-            .await
-        {
-            warn!(
-                "Error shutting down connection, but it might be fine: {}",
-                e
-            );
-        }
         let tasks = self.tasks.lock();
         for task in tasks.iter() {
             task.abort();
@@ -448,7 +447,7 @@ impl Client {
                 "Sending client information (already logged in): {:?}",
                 client_information_packet
             );
-            self.write_packet(client_information_packet).await?;
+            self.write_packet(client_information_packet);
         }
 
         Ok(())

azalea-client/src/local_player.rs

@@ -20,7 +20,7 @@ use crate::{ClientInformation, Event, PlayerInfo, WalkDirection};
 pub struct LocalPlayer {
     pub profile: GameProfile,
 
-    pub write_conn: WriteConnection<ServerboundGamePacket>,
+    pub packet_writer: mpsc::UnboundedSender<ServerboundGamePacket>,
 
     // pub world: Arc<RwLock<PartialWorld>>,
     pub physics_state: PhysicsState,
@@ -64,7 +64,7 @@ impl LocalPlayer {
     pub fn new(
         entity: Entity,
         profile: GameProfile,
-        write_conn: WriteConnection<ServerboundGamePacket>,
+        packet_writer: mpsc::UnboundedSender<ServerboundGamePacket>,
         world: Arc<RwLock<World>>,
         entity_infos: &mut EntityInfos,
         tx: mpsc::UnboundedSender<Event>,
@@ -74,7 +74,7 @@ impl LocalPlayer {
         LocalPlayer {
             profile,
 
-            write_conn,
+            packet_writer,
 
             physics_state: PhysicsState::default(),
             client_information: ClientInformation::default(),
@@ -93,18 +93,9 @@ impl LocalPlayer {
         }
     }
 
-    /// Write a packet directly to the server.
-    pub async fn write_packet_async(
-        &mut self,
-        packet: ServerboundGamePacket,
-    ) -> Result<(), std::io::Error> {
-        self.write_conn.write(packet).await?;
-        Ok(())
-    }
-
     /// Spawn a task to write a packet directly to the server.
     pub fn write_packet(&mut self, packet: ServerboundGamePacket) {
-        tokio::spawn(self.write_packet_async(packet));
+        self.packet_writer.send(packet);
     }
 }
 

azalea-client/src/packet_handling.rs

@@ -1,6 +1,9 @@
 use std::sync::Arc;
 
-use azalea_protocol::{connect::ReadConnection, packets::game::ClientboundGamePacket};
+use azalea_protocol::{
+    connect::{ReadConnection, WriteConnection},
+    packets::game::{ClientboundGamePacket, ServerboundGamePacket},
+};
 use bevy_ecs::{component::Component, prelude::Entity, query::Changed, system::Query};
 use parking_lot::Mutex;
 use tokio::sync::mpsc;
@@ -129,7 +132,7 @@ pub fn handle_packet(entity: Entity, packet: &ClientboundGamePacket) {
     //             "Sending client information because login: {:?}",
     //             client_information_packet
     //         );
-    //         client.write_packet(client_information_packet.get()).await?;
+    //         client.write_packet(client_information_packet.get());
 
     //         // brand
     //         client
@@ -613,4 +616,17 @@ impl PacketReceiver {
             self.run_schedule_sender.send(());
         }
     }
+
+    /// Consume the [`ServerboundGamePacket`] queue and actually write the
+    /// packets to the server. It's like this so writing packets doesn't need to
+    /// be awaited.
+    pub async fn write_task(
+        self,
+        mut write_conn: WriteConnection<ServerboundGamePacket>,
+        mut write_receiver: mpsc::UnboundedReceiver<ServerboundGamePacket>,
+    ) {
+        while let Some(packet) = write_receiver.recv().await {
+            write_conn.write(packet);
+        }
+    }
 }