Merge branch 'main' of https://github.com/mat-1/azalea

azalea-client/src/client.rs

@@ -4,7 +4,7 @@ use azalea_block::BlockState;
 use azalea_chat::component::Component;
 use azalea_core::{ChunkPos, ResourceLocation, Vec3};
 use azalea_protocol::{
-    connect::{Connection, ConnectionError},
+    connect::{Connection, ConnectionError, ReadConnection, WriteConnection},
     packets::{
         game::{
             clientbound_player_chat_packet::ClientboundPlayerChatPacket,
@@ -67,7 +67,8 @@ impl ChatPacket {
 #[derive(Clone)]
 pub struct Client {
     game_profile: GameProfile,
-    pub conn: Arc<tokio::sync::Mutex<Connection<ClientboundGamePacket, ServerboundGamePacket>>>,
+    pub read_conn: Arc<tokio::sync::Mutex<ReadConnection<ClientboundGamePacket>>>,
+    pub write_conn: Arc<tokio::sync::Mutex<WriteConnection<ServerboundGamePacket>>>,
     pub player: Arc<Mutex<Player>>,
     pub dimension: Arc<Mutex<Dimension>>,
     pub physics_state: Arc<Mutex<PhysicsState>>,
@@ -185,14 +186,18 @@ impl Client {
             }
         };
 
-        let conn = Arc::new(tokio::sync::Mutex::new(conn));
+        let (read_conn, write_conn) = conn.into_split();
+
+        let read_conn = Arc::new(tokio::sync::Mutex::new(read_conn));
+        let write_conn = Arc::new(tokio::sync::Mutex::new(write_conn));
 
         let (tx, rx) = mpsc::unbounded_channel();
 
         // we got the GameConnection, so the server is now connected :)
         let client = Client {
             game_profile,
-            conn,
+            read_conn,
+            write_conn,
             player: Arc::new(Mutex::new(Player::default())),
             dimension: Arc::new(Mutex::new(Dimension::default())),
             physics_state: Arc::new(Mutex::new(PhysicsState::default())),
@@ -209,9 +214,14 @@ impl Client {
         Ok((client, rx))
     }
 
+    /// Write a packet directly to the server.
+    pub async fn write_packet(&self, packet: ServerboundGamePacket) -> Result<(), std::io::Error> {
+        self.write_conn.lock().await.write(packet).await
+    }
+
     async fn protocol_loop(client: Client, tx: UnboundedSender<Event>) {
         loop {
-            let r = client.conn.lock().await.read().await;
+            let r = client.read_conn.lock().await.read().await;
             match r {
                 Ok(packet) => match Self::handle(&packet, &client, &tx).await {
                     Ok(_) => {}
@@ -323,10 +333,7 @@ impl Client {
                 }
 
                 client
-                    .conn
-                    .lock()
-                    .await
-                    .write(
+                    .write_packet(
                         ServerboundCustomPayloadPacket {
                             identifier: ResourceLocation::new("brand").unwrap(),
                             // they don't have to know :)
@@ -444,12 +451,11 @@ impl Client {
                     (new_pos, y_rot, x_rot)
                 };
 
-                let mut conn_lock = client.conn.lock().await;
-                conn_lock
-                    .write(ServerboundAcceptTeleportationPacket { id: p.id }.get())
+                client
+                    .write_packet(ServerboundAcceptTeleportationPacket { id: p.id }.get())
                     .await?;
-                conn_lock
-                    .write(
+                client
+                    .write_packet(
                         ServerboundMovePlayerPosRotPacket {
                             x: new_pos.x,
                             y: new_pos.y,
@@ -567,10 +573,7 @@ impl Client {
             ClientboundGamePacket::KeepAlive(p) => {
                 debug!("Got keep alive packet {:?}", p);
                 client
-                    .conn
-                    .lock()
-                    .await
-                    .write(ServerboundKeepAlivePacket { id: p.id }.get())
+                    .write_packet(ServerboundKeepAlivePacket { id: p.id }.get())
                     .await?;
             }
             ClientboundGamePacket::RemoveEntities(p) => {

azalea-client/src/movement.rs

@@ -121,7 +121,7 @@ impl Client {
         };
 
         if let Some(packet) = packet {
-            self.conn.lock().await.write(packet).await?;
+            self.write_packet(packet).await?;
         }
 
         Ok(())

azalea-crypto/benches/my_benchmark.rs

@@ -2,7 +2,7 @@ use azalea_crypto::{create_cipher, decrypt_packet, encrypt_packet};
 use criterion::{criterion_group, criterion_main, Criterion};
 
 fn bench(c: &mut Criterion) {
-    let (mut enc, mut dec) = create_cipher(b"0123456789abcdef");
+    let (mut enc, dec) = create_cipher(b"0123456789abcdef");
 
     let mut packet = [0u8; 65536];
     for i in 0..packet.len() {

azalea-protocol/src/connect.rs

@@ -12,41 +12,74 @@ use azalea_crypto::{Aes128CfbDec, Aes128CfbEnc};
 use std::fmt::Debug;
 use std::marker::PhantomData;
 use thiserror::Error;
+use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::TcpStream;
 
-pub struct Connection<R: ProtocolPacket, W: ProtocolPacket> {
-    /// The buffered writer
-    pub stream: TcpStream,
+pub struct ReadConnection<R: ProtocolPacket> {
+    pub read_stream: OwnedReadHalf,
     pub compression_threshold: Option<u32>,
-    pub enc_cipher: Option<Aes128CfbEnc>,
     pub dec_cipher: Option<Aes128CfbDec>,
     _reading: PhantomData<R>,
+}
+
+pub struct WriteConnection<W: ProtocolPacket> {
+    pub write_stream: OwnedWriteHalf,
+    pub compression_threshold: Option<u32>,
+    pub enc_cipher: Option<Aes128CfbEnc>,
     _writing: PhantomData<W>,
 }
 
+pub struct Connection<R: ProtocolPacket, W: ProtocolPacket> {
+    pub reader: ReadConnection<R>,
+    pub writer: WriteConnection<W>,
+}
+
+impl<R> ReadConnection<R>
+where
+    R: ProtocolPacket + Debug,
+{
+    pub async fn read(&mut self) -> Result<R, ReadPacketError> {
+        read_packet::<R, _>(
+            &mut self.read_stream,
+            self.compression_threshold,
+            &mut self.dec_cipher,
+        )
+        .await
+    }
+}
+impl<W> WriteConnection<W>
+where
+    W: ProtocolPacket + Debug,
+{
+    /// Write a packet to the server
+    pub async fn write(&mut self, packet: W) -> std::io::Result<()> {
+        write_packet(
+            packet,
+            &mut self.write_stream,
+            self.compression_threshold,
+            &mut self.enc_cipher,
+        )
+        .await
+    }
+}
+
 impl<R, W> Connection<R, W>
 where
     R: ProtocolPacket + Debug,
     W: ProtocolPacket + Debug,
 {
     pub async fn read(&mut self) -> Result<R, ReadPacketError> {
-        read_packet::<R, _>(
-            &mut self.stream,
-            self.compression_threshold,
-            &mut self.dec_cipher,
-        )
-        .await
+        self.reader.read().await
     }
 
     /// Write a packet to the server
     pub async fn write(&mut self, packet: W) -> std::io::Result<()> {
-        write_packet(
-            packet,
-            &mut self.stream,
-            self.compression_threshold,
-            &mut self.enc_cipher,
-        )
-        .await
+        self.writer.write(packet).await
+    }
+
+    /// Split the reader and writer into two objects. This doesn't allocate.
+    pub fn into_split(self) -> (ReadConnection<R>, WriteConnection<W>) {
+        (self.reader, self.writer)
     }
 }
 
@@ -66,13 +99,21 @@ impl Connection<ClientboundHandshakePacket, ServerboundHandshakePacket> {
         // enable tcp_nodelay
         stream.set_nodelay(true)?;
 
+        let (read_stream, write_stream) = stream.into_split();
+
         Ok(Connection {
-            stream,
-            compression_threshold: None,
-            enc_cipher: None,
-            dec_cipher: None,
-            _reading: PhantomData,
-            _writing: PhantomData,
+            reader: ReadConnection {
+                read_stream,
+                compression_threshold: None,
+                dec_cipher: None,
+                _reading: PhantomData,
+            },
+            writer: WriteConnection {
+                write_stream,
+                compression_threshold: None,
+                enc_cipher: None,
+                _writing: PhantomData,
+            },
         })
     }
 
@@ -89,17 +130,19 @@ impl Connection<ClientboundLoginPacket, ServerboundLoginPacket> {
     pub fn set_compression_threshold(&mut self, threshold: i32) {
         // if you pass a threshold of less than 0, compression is disabled
         if threshold >= 0 {
-            self.compression_threshold = Some(threshold as u32);
+            self.reader.compression_threshold = Some(threshold as u32);
+            self.writer.compression_threshold = Some(threshold as u32);
         } else {
-            self.compression_threshold = None;
+            self.reader.compression_threshold = None;
+            self.writer.compression_threshold = None;
         }
     }
 
     pub fn set_encryption_key(&mut self, key: [u8; 16]) {
         // minecraft has a cipher decoder and encoder, i don't think it matters though?
         let (enc_cipher, dec_cipher) = azalea_crypto::create_cipher(&key);
-        self.enc_cipher = Some(enc_cipher);
-        self.dec_cipher = Some(dec_cipher);
+        self.writer.enc_cipher = Some(enc_cipher);
+        self.reader.dec_cipher = Some(dec_cipher);
     }
 
     pub fn game(self) -> Connection<ClientboundGamePacket, ServerboundGamePacket> {
@@ -120,12 +163,18 @@ where
         W2: ProtocolPacket + Debug,
     {
         Connection {
-            stream: connection.stream,
-            compression_threshold: connection.compression_threshold,
-            enc_cipher: connection.enc_cipher,
-            dec_cipher: connection.dec_cipher,
-            _reading: PhantomData,
-            _writing: PhantomData,
+            reader: ReadConnection {
+                read_stream: connection.reader.read_stream,
+                compression_threshold: connection.reader.compression_threshold,
+                dec_cipher: connection.reader.dec_cipher,
+                _reading: PhantomData,
+            },
+            writer: WriteConnection {
+                compression_threshold: connection.writer.compression_threshold,
+                write_stream: connection.writer.write_stream,
+                enc_cipher: connection.writer.enc_cipher,
+                _writing: PhantomData,
+            },
         }
     }
 }

azalea-protocol/src/read.rs

@@ -221,10 +221,7 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::packets::{
-        game::{clientbound_player_chat_packet::ChatType, ClientboundGamePacket},
-        handshake::ClientboundHandshakePacket,
-    };
+    use crate::packets::game::{clientbound_player_chat_packet::ChatType, ClientboundGamePacket};
     use std::io::Cursor;
 
     #[tokio::test]