try to implement compression

Cargo.lock

@@ -2,6 +2,25 @@
 # It is not intended for manual editing.
 version = 3
 
+[[package]]
+name = "adler"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
+
+[[package]]
+name = "async-compression"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5443ccbb270374a2b1055fc72da40e1f237809cd6bb0e97e66d264cd138473a6"
+dependencies = [
+ "flate2",
+ "futures-core",
+ "memchr",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "async-recursion"
 version = "0.3.2"
@@ -64,6 +83,7 @@ dependencies = [
 name = "azalea-protocol"
 version = "0.1.0"
 dependencies = [
+ "async-compression",
  "async-recursion",
  "async-trait",
  "azalea-auth",
@@ -113,6 +133,15 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
+[[package]]
+name = "crc32fast"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "738c290dfaea84fc1ca15ad9c168d083b05a714e1efddd8edaab678dc28d2836"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "data-encoding"
 version = "2.3.2"
@@ -131,6 +160,18 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "flate2"
+version = "1.0.22"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e6988e897c1c9c485f43b47a529cef42fde0547f9d8d41a7062518f1d8fc53f"
+dependencies = [
+ "cfg-if",
+ "crc32fast",
+ "libc",
+ "miniz_oxide",
+]
+
 [[package]]
 name = "form_urlencoded"
 version = "1.0.1"
@@ -334,6 +375,16 @@ version = "2.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "308cc39be01b73d0d18f82a0e7b2a3df85245f84af96fdddc5d202d27e47b86a"
 
+[[package]]
+name = "miniz_oxide"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a92518e98c078586bc6c934028adcca4c92a53d6a958196de835170a01d84e4b"
+dependencies = [
+ "adler",
+ "autocfg",
+]
+
 [[package]]
 name = "mio"
 version = "0.7.14"

azalea-client/src/connect.rs

@@ -29,25 +29,28 @@ pub async fn join_server(address: &ServerAddress) -> Result<(), String> {
     .await;
     let mut conn = conn.login();
 
-    // login start
+    // login
     conn.write(ServerboundHelloPacket { username }.get()).await;
 
-    // encryption request
+    let mut conn = loop {
-    loop {
         match conn.read().await.unwrap() {
-            LoginPacket::ClientboundHelloPacket(encryption_request_packet) => {
+            LoginPacket::ClientboundHelloPacket(p) => {
-                println!(
+                println!("Got encryption request {:?} {:?}", p.nonce, p.public_key);
-                    "Got encryption request {:?} {:?}",
-                    encryption_request_packet.nonce, encryption_request_packet.public_key
-                );
             }
-            _ => (),
+            LoginPacket::ClientboundLoginCompressionPacket(p) => {
+                println!("Got compression request {:?}", p.compression_threshold);
+                conn.set_compression_threshold(p.compression_threshold);
+            }
+            LoginPacket::ClientboundGameProfilePacket(p) => {
+                println!("Got profile {:?}", p.game_profile);
+                break conn.game();
+            }
+            _ => panic!("unhandled packet"),
         }
-    }
+    };
 
-    // TODO: client auth
+    // game
-
+    panic!("ok i haven't implemented game yet");
-    // TODO: encryption response
 
     Ok(())
 }

azalea-protocol/Cargo.toml

@@ -6,6 +6,7 @@ version = "0.1.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-compression = {version = "^0.3.8", features = ["tokio", "zlib"]}
 async-recursion = "^0.3.2"
 async-trait = "0.1.51"
 azalea-auth = {path = "../azalea-auth"}

azalea-protocol/src/connect.rs

@@ -24,6 +24,7 @@ pub struct GameConnection {
     pub flow: PacketFlow,
     /// The buffered writer
     pub stream: TcpStream,
+    pub compression_threshold: Option<u32>,
 }
 
 pub struct StatusConnection {
@@ -36,6 +37,7 @@ pub struct LoginConnection {
     pub flow: PacketFlow,
     /// The buffered writer
     pub stream: TcpStream,
+    pub compression_threshold: Option<u32>,
 }
 
 impl HandshakeConnection {
@@ -62,6 +64,7 @@ impl HandshakeConnection {
         LoginConnection {
             flow: self.flow,
             stream: self.stream,
+            compression_threshold: None,
         }
     }
 
@@ -73,7 +76,7 @@ impl HandshakeConnection {
     }
 
     pub async fn read(&mut self) -> Result<HandshakePacket, String> {
-        read_packet::<HandshakePacket>(&self.flow, &mut self.stream).await
+        read_packet::<HandshakePacket>(&self.flow, &mut self.stream, None).await
     }
 
     /// Write a packet to the server
@@ -84,7 +87,7 @@ impl HandshakeConnection {
 
 impl GameConnection {
     pub async fn read(&mut self) -> Result<GamePacket, String> {
-        read_packet::<GamePacket>(&self.flow, &mut self.stream).await
+        read_packet::<GamePacket>(&self.flow, &mut self.stream, self.compression_threshold).await
     }
 
     /// Write a packet to the server
@@ -95,7 +98,7 @@ impl GameConnection {
 
 impl StatusConnection {
     pub async fn read(&mut self) -> Result<StatusPacket, String> {
-        read_packet::<StatusPacket>(&self.flow, &mut self.stream).await
+        read_packet::<StatusPacket>(&self.flow, &mut self.stream, None).await
     }
 
     /// Write a packet to the server
@@ -106,11 +109,28 @@ impl StatusConnection {
 
 impl LoginConnection {
     pub async fn read(&mut self) -> Result<LoginPacket, String> {
-        read_packet::<LoginPacket>(&self.flow, &mut self.stream).await
+        read_packet::<LoginPacket>(&self.flow, &mut self.stream, self.compression_threshold).await
     }
 
     /// Write a packet to the server
     pub async fn write(&mut self, packet: LoginPacket) {
         write_packet(packet, &mut self.stream).await;
     }
+
+    pub fn set_compression_threshold(&mut self, threshold: i32) {
+        // if you pass a threshold of 0 or less, compression is disabled
+        if threshold > 0 {
+            self.compression_threshold = Some(threshold as u32);
+        } else {
+            self.compression_threshold = None;
+        }
+    }
+
+    pub fn game(self) -> GameConnection {
+        GameConnection {
+            flow: self.flow,
+            stream: self.stream,
+            compression_threshold: self.compression_threshold,
+        }
+    }
 }

azalea-protocol/src/mc_buf.rs

@@ -134,6 +134,8 @@ impl Writable for Vec<u8> {
 pub trait Readable {
     async fn read_int_id_list(&mut self) -> Result<Vec<i32>, String>;
     async fn read_varint(&mut self) -> Result<i32, String>;
+    fn get_varint_size(&mut self, value: i32) -> u8;
+    fn get_varlong_size(&mut self, value: i32) -> u8;
     async fn read_byte_array(&mut self) -> Result<Vec<u8>, String>;
     async fn read_bytes(&mut self, n: usize) -> Result<Vec<u8>, String>;
     async fn read_utf(&mut self) -> Result<String, String>;
@@ -173,6 +175,26 @@ where
         Ok(ans)
     }
 
+    fn get_varint_size(&mut self, value: i32) -> u8 {
+        for i in 1..5 {
+            if (value & -1 << i * 7) != 0 {
+                continue;
+            }
+            return i;
+        }
+        return 5;
+    }
+
+    fn get_varlong_size(&mut self, value: i32) -> u8 {
+        for i in 1..10 {
+            if (value & -1 << i * 7) != 0 {
+                continue;
+            }
+            return i;
+        }
+        return 10;
+    }
+
     async fn read_byte_array(&mut self) -> Result<Vec<u8>, String> {
         let length = self.read_varint().await? as usize;
         Ok(self.read_bytes(length).await?)

azalea-protocol/src/packets/game/ClientboundLoginPacket.rs

@@ -0,0 +1,53 @@
+use super::GamePacket;
+use crate::mc_buf::{Readable, Writable};
+use azalea_core::resource_location::ResourceLocation;
+use std::hash::Hash;
+use tokio::io::BufReader;
+
+#[derive(Hash, Clone, Debug)]
+pub struct ClientboundLoginPacket {
+	// private final int playerId;
+    // private final boolean hardcore;
+    // private final GameType gameType;
+    // @Nullable
+    // private final GameType previousGameType;
+    // private final Set<ResourceKey<Level>> levels;
+    // private final RegistryAccess.RegistryHolder registryHolder;
+    // private final DimensionType dimensionType;
+    // private final ResourceKey<Level> dimension;
+    // private final long seed;
+    // private final int maxPlayers;
+    // private final int chunkRadius;
+    // private final int simulationDistance;
+    // private final boolean reducedDebugInfo;
+    // private final boolean showDeathScreen;
+    // private final boolean isDebug;
+    // private final boolean isFlat;
+	
+}
+
+impl ClientboundLoginPacket {
+    pub fn get(self) -> GamePacket {
+        GamePacket::ClientboundLoginPacket(self)
+    }
+
+    pub fn write(&self, buf: &mut Vec<u8>) {
+        buf.write_varint(self.transaction_id as i32).unwrap();
+        buf.write_utf(self.identifier.to_string().as_str()).unwrap();
+        buf.write_bytes(&self.data).unwrap();
+    }
+
+    pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
+        buf: &mut T,
+    ) -> Result<GamePacket, String> {
+        let transaction_id = buf.read_varint().await? as u32;
+        let identifier = ResourceLocation::new(&buf.read_utf().await?)?;
+        let data = buf.read_bytes(1048576).await?;
+        Ok(ClientboundLoginPacket {
+            transaction_id,
+            identifier,
+            data,
+        }
+        .get())
+    }
+}

azalea-protocol/src/packets/game/mod.rs

@@ -22,7 +22,7 @@ impl ProtocolPacket for GamePacket {
     async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
         _id: u32,
         flow: &PacketFlow,
-        _buf: &mut BufReader<T>,
+        _buf: &mut T,
     ) -> Result<GamePacket, String>
     where
         Self: Sized,

azalea-protocol/src/packets/handshake/client_intention_packet.rs

@@ -28,7 +28,7 @@ impl ClientIntentionPacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        _buf: &mut BufReader<T>,
+        _buf: &mut T,
     ) -> Result<HandshakePacket, String> {
         Err("ClientIntentionPacket::parse not implemented".to_string())
         // Ok(ClientIntentionPacket {}.get())

azalea-protocol/src/packets/handshake/mod.rs

@@ -33,7 +33,7 @@ impl ProtocolPacket for HandshakePacket {
     async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
         id: u32,
         flow: &PacketFlow,
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<HandshakePacket, String>
     where
         Self: Sized,

azalea-protocol/src/packets/login/clientbound_custom_query_packet.rs

@@ -23,7 +23,7 @@ impl ClientboundCustomQueryPacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<LoginPacket, String> {
         let transaction_id = buf.read_varint().await? as u32;
         let identifier = ResourceLocation::new(&buf.read_utf().await?)?;

azalea-protocol/src/packets/login/clientbound_game_profile_packet.rs

@@ -23,7 +23,7 @@ impl ClientboundGameProfilePacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<LoginPacket, String> {
         let uuid = Uuid::from_int_array([
             buf.read_int().await? as u32,

azalea-protocol/src/packets/login/clientbound_hello_packet.rs

@@ -22,7 +22,7 @@ impl ClientboundHelloPacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<LoginPacket, String> {
         let server_id = buf.read_utf_with_len(20).await?;
         let public_key = buf.read_byte_array().await?;

azalea-protocol/src/packets/login/clientbound_login_compression_packet.rs

@@ -20,7 +20,7 @@ impl ClientboundLoginCompressionPacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<LoginPacket, String> {
         let compression_threshold = buf.read_varint().await?;
 

azalea-protocol/src/packets/login/mod.rs

@@ -51,7 +51,7 @@ impl ProtocolPacket for LoginPacket {
     async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
         id: u32,
         flow: &PacketFlow,
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<LoginPacket, String>
     where
         Self: Sized,

azalea-protocol/src/packets/login/serverbound_hello_packet.rs

@@ -20,7 +20,7 @@ impl ServerboundHelloPacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        _buf: &mut BufReader<T>,
+        _buf: &mut T,
     ) -> Result<LoginPacket, String> {
         Err("ServerboundHelloPacket::read not implemented".to_string())
     }

azalea-protocol/src/packets/mod.rs

@@ -38,109 +38,10 @@ where
     async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
         id: u32,
         flow: &PacketFlow,
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<Self, String>
     where
         Self: Sized;
 
     fn write(&self, buf: &mut Vec<u8>);
 }
-
-// impl Packet {
-//     fn get_inner_packet(&self) -> &dyn PacketTrait {
-//         match self {
-//             Packet::ClientIntentionPacket(packet) => packet,
-//             Packet::ServerboundStatusRequestPacket(packet) => packet,
-//             Packet::ClientboundStatusResponsePacket(packet) => packet,
-//             Packet::ServerboundHelloPacket(packet) => packet,
-//             Packet::ClientboundHelloPacket(packet) => packet,
-//         }
-//     }
-
-//     pub fn id(&self) -> u32 {
-//         match self {
-//             Packet::ClientIntentionPacket(_packet) => 0x00,
-//             Packet::ServerboundStatusRequestPacket(_packet) => 0x00,
-//             Packet::ClientboundStatusResponsePacket(_packet) => 0x00,
-//             Packet::ServerboundHelloPacket(_packet) => 0x00,
-//             Packet::ClientboundHelloPacket(_packet) => 0x01,
-//         }
-//     }
-
-//     /// Read a packet by its id, ConnectionProtocol, and flow
-//     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-//         id: u32,
-//         protocol: &ConnectionProtocol,
-//         flow: &PacketFlow,
-//         buf: &mut BufReader<T>,
-//     ) -> Result<Packet, String> {
-//         match protocol {
-//             ConnectionProtocol::Handshake => match flow {
-//                 PacketFlow::ClientToServer => match id {
-//                     0x00 => Ok(
-//                         handshake::client_intention_packet::ClientIntentionPacket::read(buf).await?,
-//                     ),
-//                     _ => Err(format!("Unknown ClientToServer handshake packet id: {}", id)),
-//                 }
-//                 PacketFlow::ServerToClient => Err("ServerToClient handshake packets not implemented".to_string()),
-//             },
-
-//             ConnectionProtocol::Game => Err("Game protocol not implemented yet".to_string()),
-
-//             ConnectionProtocol::Status => match flow {
-//                 PacketFlow::ServerToClient => match id {
-//                     0x00 => Ok(
-//                         status::clientbound_status_response_packet::ClientboundStatusResponsePacket
-//                             ::read(buf)
-//                             .await?,
-//                     ),
-//                     _ => Err(format!("Unknown ServerToClient status packet id: {}", id)),
-//                 },
-//                 PacketFlow::ClientToServer => match id {
-//                     0x00 => Ok(
-//                         status::serverbound_status_request_packet::ServerboundStatusRequestPacket
-//                             ::read(buf)
-//                             .await?,
-//                     ),
-//                     _ => Err(format!("Unknown ClientToServer status packet id: {}", id)),
-//                 },
-//             },
-
-//             ConnectionProtocol::Login => match flow {
-//                 PacketFlow::ServerToClient => match id {
-//                     0x01 => Ok(
-//                         login::clientbound_hello_packet::ClientboundHelloPacket::read(buf).await?,
-//                     ),
-//                     _ => Err(format!("Unknown ServerToClient login packet id: {}", id)),
-//                 },
-//                 PacketFlow::ClientToServer => match id {
-//                     0x00 => Ok(
-//                         login::serverbound_hello_packet::ServerboundHelloPacket::read(buf).await?,
-//                     ),
-//                     _ => Err(format!("Unknown ClientToServer login packet id: {}", id)),
-//                 },
-//             },
-//         }
-//     }
-
-//     pub fn write(&self, buf: &mut Vec<u8>) {
-//         self.get_inner_packet().write(buf);
-//     }
-// }
-
-// #[async_trait]
-// pub trait PacketTrait
-// where
-//     Self: Sized,
-// {
-//     /// Return a version of the packet that you can actually use for stuff
-//     fn get(self) -> dyn ProtocolPacket;
-
-//     fn write(&self, buf: &mut Vec<u8>);
-
-//     async fn read<T: AsyncRead + std::marker::Unpin + std::marker::Send, P: ProtocolPacket>(
-//         buf: &mut BufReader<T>,
-//     ) -> Result<P, String>
-//     where
-//         Self: Sized;
-// }

azalea-protocol/src/packets/status/clientbound_status_response_packet.rs

@@ -43,7 +43,7 @@ impl ClientboundStatusResponsePacket {
     pub fn write(&self, _buf: &mut Vec<u8>) {}
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<StatusPacket, String> {
         let status_string = buf.read_utf().await?;
         let status_json: Value =

azalea-protocol/src/packets/status/mod.rs

@@ -41,7 +41,7 @@ impl ProtocolPacket for StatusPacket {
     async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
         id: u32,
         flow: &PacketFlow,
-        buf: &mut BufReader<T>,
+        buf: &mut T,
     ) -> Result<StatusPacket, String>
     where
         Self: Sized,

azalea-protocol/src/packets/status/serverbound_status_request_packet.rs

@@ -16,7 +16,7 @@ impl ServerboundStatusRequestPacket {
     }
 
     pub async fn read<T: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send>(
-        _buf: &mut BufReader<T>,
+        _buf: &mut T,
     ) -> Result<StatusPacket, String> {
         Err("ServerboundStatusRequestPacket::read not implemented".to_string())
     }

azalea-protocol/src/read.rs

@@ -1,10 +1,14 @@
-use tokio::{io::BufReader, net::TcpStream};
-
 use crate::{connect::PacketFlow, mc_buf::Readable, packets::ProtocolPacket};
+use async_compression::tokio::bufread::ZlibDecoder;
+use tokio::{
+    io::{AsyncReadExt, BufReader},
+    net::TcpStream,
+};
 
 pub async fn read_packet<P: ProtocolPacket>(
     flow: &PacketFlow,
     stream: &mut TcpStream,
+    compression_threshold: Option<u32>,
 ) -> Result<P, String> {
     // what this does:
     // 1. reads the first 5 bytes, probably only some of this will be used to get the packet length
@@ -15,14 +19,64 @@ pub async fn read_packet<P: ProtocolPacket>(
     // the first thing minecraft sends us is the length as a varint, which can be up to 5 bytes long
     let mut buf = BufReader::with_capacity(4 * 1024 * 1024, stream);
 
-    let _packet_size = buf.read_varint().await?;
+    // Packet Length
+    let packet_size = buf.read_varint().await?;
 
-    // then, minecraft tells us the packet id as a varint
+    // if there's no compression, we can just read the rest of the packet normally
-    let packet_id = buf.read_varint().await?;
+    if compression_threshold.is_none() {
+        // then, minecraft tells us the packet id as a varint
+        let packet_id = buf.read_varint().await?;
 
-    // if we recognize the packet id, parse it
+        // if we recognize the packet id, parse it
 
-    let packet = P::read(packet_id.try_into().unwrap(), flow, &mut buf).await?;
+        println!("reading uncompressed packet id: {}", packet_id);
+        let packet = P::read(packet_id.try_into().unwrap(), flow, &mut buf).await?;
 
-    Ok(packet)
+        return Ok(packet);
+    }
+
+    println!("compressed packet size: {}", packet_size);
+
+    // there's compression
+    // Data Length
+    let data_size = buf.read_varint().await?;
+    println!("data size: {}", data_size);
+
+    // this packet has no compression
+    if data_size == 0 {
+        // Packet ID
+        let packet_id = buf.read_varint().await?;
+        println!(
+            "reading compressed packet without compression packet id: {}",
+            packet_id
+        );
+        let packet = P::read(packet_id.try_into().unwrap(), flow, &mut buf).await?;
+        return Ok(packet);
+    }
+
+    // this packet has compression
+    let packet_size_varint_size = buf.get_varint_size(packet_size);
+
+    let mut compressed_data = vec![0; packet_size as usize - packet_size_varint_size as usize];
+    buf.read_exact(compressed_data.as_mut_slice())
+        .await
+        .expect("Not enough compressed data");
+
+    let mut z = ZlibDecoder::new(compressed_data.as_slice());
+
+    // Packet ID
+    let packet_id = z.read_varint().await.unwrap();
+    println!("reading compressed packet id: {}", packet_id);
+
+    if let Ok(packet) = P::read(packet_id as u32, flow, &mut z).await {
+        Ok(packet)
+    } else {
+        // read the rest of the bytes
+        let packet_id_varint_size = z.get_varint_size(packet_id);
+        let mut buf = vec![0; packet_size as usize - packet_id_varint_size as usize];
+        z.read_exact(buf.as_mut_slice()).await.unwrap();
+        println!("{:?}", buf);
+
+        Err(format!("Error on packet id: {}", packet_id))
+    }
 }

bot/src/main.rs

@@ -5,6 +5,7 @@ async fn main() {
     println!("Hello, world!");
 
     let address = "95.111.249.143:10000";
+    // let address = "localhost:63482";
     let _response = join_server(&address.try_into().unwrap()).await.unwrap();
     // println!("{}", response.description.to_ansi(None));
     println!("connected");