From 12de93a4bbe40d5c02e78365921b1ea1f3fc0728 Mon Sep 17 00:00:00 2001 From: Alexander Medvedev Date: Mon, 4 May 2026 10:35:04 +0200 Subject: [PATCH] feat(bedrock): add zLib packet compression --- pumpkin-config/src/networking/mod.rs | 6 +- pumpkin-protocol/Cargo.toml | 2 +- .../src/bedrock/client/network_settings.rs | 1 + .../src/bedrock/packet_decoder.rs | 132 +++++++----------- .../src/bedrock/packet_encoder.rs | 99 ++++++++----- pumpkin/src/lib.rs | 44 +++--- pumpkin/src/net/bedrock/connection.rs | 37 +---- pumpkin/src/net/bedrock/login.rs | 25 +++- pumpkin/src/net/bedrock/mod.rs | 108 ++++++++++---- pumpkin/src/net/java/login.rs | 4 +- 10 files changed, 249 insertions(+), 209 deletions(-) diff --git a/pumpkin-config/src/networking/mod.rs b/pumpkin-config/src/networking/mod.rs index c802c91f3..91e906c11 100644 --- a/pumpkin-config/src/networking/mod.rs +++ b/pumpkin-config/src/networking/mod.rs @@ -27,8 +27,10 @@ pub struct NetworkingConfig { pub rcon: RCONConfig, /// Proxy-related networking settings. pub proxy: ProxyConfig, - /// Packet compression settings. - pub packet_compression: CompressionConfig, + /// Java Edition packet compression settings. + pub java_compression: CompressionConfig, + /// Bedrock Edition packet compression settings. + pub bedrock_compression: CompressionConfig, /// LAN broadcast settings. pub lan_broadcast: LANBroadcastConfig, } diff --git a/pumpkin-protocol/Cargo.toml b/pumpkin-protocol/Cargo.toml index 10a7b6f21..0aa8d8267 100644 --- a/pumpkin-protocol/Cargo.toml +++ b/pumpkin-protocol/Cargo.toml @@ -30,7 +30,7 @@ cfb8.workspace = true hybrid-array = "0.4" # compression -async-compression = { workspace = true, features = ["tokio", "zlib"] } +async-compression = { workspace = true, features = ["tokio", "zlib", "deflate"] } flate2.workspace = true take_mut.workspace = true diff --git a/pumpkin-protocol/src/bedrock/client/network_settings.rs b/pumpkin-protocol/src/bedrock/client/network_settings.rs index ce53b42e6..64e51ce1b 100644 --- a/pumpkin-protocol/src/bedrock/client/network_settings.rs +++ b/pumpkin-protocol/src/bedrock/client/network_settings.rs @@ -6,6 +6,7 @@ use crate::serial::PacketWrite; #[packet(143)] pub struct CNetworkSettings { compression_threshold: u16, + /// `ZLib` = 0, Snappy = 1, None = 255 compression_method: u16, client_throttle_enabled: bool, client_throttle_threshold: i8, diff --git a/pumpkin-protocol/src/bedrock/packet_decoder.rs b/pumpkin-protocol/src/bedrock/packet_decoder.rs index 605b987c8..03868fa91 100644 --- a/pumpkin-protocol/src/bedrock/packet_decoder.rs +++ b/pumpkin-protocol/src/bedrock/packet_decoder.rs @@ -4,43 +4,15 @@ use std::{ task::{Context, Poll}, }; -use async_compression::tokio::bufread::ZlibDecoder; -use bytes::Bytes; +use async_compression::tokio::bufread::DeflateDecoder; use tokio::io::{AsyncRead, BufReader, ReadBuf}; use crate::{ - Aes128Cfb8Dec, CompressionThreshold, MAX_PACKET_SIZE, PacketDecodeError, RawPacket, - StreamDecryptor, + Aes128Cfb8Dec, CompressionThreshold, PacketDecodeError, RawPacket, StreamDecryptor, codec::var_uint::VarUInt, ser::{NetworkReadExt, ReadingError}, }; -// decrypt -> decompress -> raw -pub enum DecompressionReader { - Decompress(ZlibDecoder>), - None(R), -} - -impl AsyncRead for DecompressionReader { - #[inline] - fn poll_read( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - match self.get_mut() { - Self::Decompress(reader) => { - let reader = Pin::new(reader); - reader.poll_read(cx, buf) - } - Self::None(reader) => { - let reader = Pin::new(reader); - reader.poll_read(cx, buf) - } - } - } -} - pub enum DecryptionReader { Decrypt(Box>), None(R), @@ -108,71 +80,71 @@ impl UDPNetworkDecoder { // take_mut::take(&mut self.reader, |decoder| decoder.upgrade(cipher)); } - #[expect(clippy::unused_async)] pub async fn get_packet_payload( &mut self, - reader: Cursor>, - ) -> Result { - //let mut payload = Vec::new(); - //reader - // .read_to_end(&mut payload) - // .await - // .map_err(|err| PacketDecodeError::FailedDecompression(err.to_string()))?; + full_packet: Vec, + ) -> Result, PacketDecodeError> { + if full_packet.is_empty() { + return Err(PacketDecodeError::MalformedLength("Empty packet".into())); + } - Ok(reader.into_inner().into()) + // If the first byte isn't 0xfe, it's likely a RakNet control packet or encrypted. + // Ensure your RakNet implementation is providing ONLY the payload here. + if full_packet[0] != 0xfe { + return Err(PacketDecodeError::MalformedLength(format!( + "Missing 0xfe header (found 0x{:02x})", + full_packet[0] + ))); + } + + // If compression is NOT enabled yet, the payload starts at index 1. + if self.compression.is_none() { + return Ok(full_packet[1..].to_vec()); + } + + // If compression IS enabled, Bedrock expects a compression method byte at index 1. + let compression_method = full_packet[1]; + let data_start = 2; + + match compression_method { + 0x00 => { + use tokio::io::AsyncReadExt; + let compressed_payload = &full_packet[data_start..]; + let mut decoder = DeflateDecoder::new(BufReader::new(compressed_payload)); + let mut decompressed = Vec::new(); + decoder + .read_to_end(&mut decompressed) + .await + .map_err(|e| PacketDecodeError::FailedDecompression(e.to_string()))?; + Ok(decompressed) + } + 0xff => { + // None (Compression enabled but this specific packet is raw) + Ok(full_packet[data_start..].to_vec()) + } + _ => Err(PacketDecodeError::FailedDecompression(format!( + "Unsupported compression method: 0x{compression_method:02x}" + ))), + } } pub fn get_game_packet( &mut self, - mut reader: Cursor>, + decompressed_reader: &mut Cursor>, ) -> Result { - if self.compression.is_some() { - let _method = reader.get_u8().unwrap(); - // None Compression - } - - //compression is only included after the network settings packet is sent - // TODO: compression & encryption - let packet_len = VarUInt::decode(&mut reader).map_err(|err| match err { + let packet_len = VarUInt::decode(decompressed_reader).map_err(|err| match err { ReadingError::CleanEOF(_) => PacketDecodeError::ConnectionClosed, err => PacketDecodeError::MalformedLength(err.to_string()), })?; - let packet_len = u64::from(packet_len.0); - if !(0..=MAX_PACKET_SIZE).contains(&packet_len) { - Err(PacketDecodeError::OutOfBounds)?; - } - - let var_header = VarUInt::decode(&mut reader)?; - - // The header is 14 bits. Ensure we only consider these bits. - // A varint u32 could be larger, so we mask to the relevant bits. - let header = var_header.0 & 0x3FFF; // Mask to get the lower 14 bits (2^14 - 1) - - // Extract components from GamePacket Header (14 bits) - // Gamepacket ID (10 bits) - // SubClient Sender ID (2 bits) - // SubClient Target ID (2 bits) - - // SubClient Target ID: Lowest 2 bits - // let _sub_client_target = (header >> 10 & 0b11) as u8; - - // SubClient Sender ID: Next 2 bits (bits 2 and 3) - // let _sub_client_sender = (header >> 12 & 0b11) as u8; - - // Gamepacket ID: Remaining 10 bits (bits 4 to 13) - let gamepacket_id = (header & 0x3FF) as u16; // 0x3FF is 10 bits set to 1 + let var_header = VarUInt::decode(decompressed_reader)?; + let header = var_header.0 & 0x3FFF; + let gamepacket_id = (header & 0x3FF) as u16; let header_size = var_header.written_size(); + let payload_len = packet_len.0 as usize - header_size; - if (packet_len as usize) < header_size { - return Err(PacketDecodeError::MalformedLength( - "Packet length is smaller than the header size".to_string(), - )); - } - let payload_len = packet_len as usize - header_size; - - let payload = reader + let payload = decompressed_reader .read_boxed_slice(payload_len) .map_err(|err| PacketDecodeError::FailedDecompression(err.to_string()))?; diff --git a/pumpkin-protocol/src/bedrock/packet_encoder.rs b/pumpkin-protocol/src/bedrock/packet_encoder.rs index 7438a03e6..be2f86b67 100644 --- a/pumpkin-protocol/src/bedrock/packet_encoder.rs +++ b/pumpkin-protocol/src/bedrock/packet_encoder.rs @@ -5,7 +5,7 @@ use std::{ task::{Context, Poll}, }; -use thiserror::Error; +use flate2::{Compression, write::DeflateEncoder}; use tokio::{io::AsyncWrite, net::UdpSocket}; use crate::{ @@ -119,50 +119,40 @@ impl UDPNetworkEncoder { packet_payload: &[u8], mut writer: impl Write, ) -> Result<(), Error> { - // Game Packet ID - writer.write_u8(0xfe).unwrap(); + let mut inner_buffer = Vec::new(); - if self.compression.is_some() { - // Todo compression - writer.write_u8(u8::MAX).unwrap(); - } - - // TODO: compression & encryption - - // Gamepacket ID (10 bits) << 4 (offset by 2 bits for target + 2 bits for sender) - // SubClient Sender ID (2 bits) << 2 (offset by 2 bits for target) - // SubClient Target ID (2 bits) + // Gamepacket ID Header (14 bits) let header_value: u32 = u32::from(packet_id) | ((sub_client_sender as u32) << 10) | ((sub_client_target as u32) << 12); + let fourteen_bit_header = header_value & 0x3FFF; - // Ensure the combined header doesn't exceed 14 bits (just a sanity check, should be handled by above shifts) - let fourteen_bit_header = header_value & 0x3FFF; // Mask to ensure it fits in 14 bits + let header_varint = VarUInt(fourteen_bit_header); + let total_content_length = (header_varint.written_size() + packet_payload.len()) as u32; - // 2. Calculate total packet_len - // This is where `VarInt::encoded_len` is crucial. - // We need to know the byte length of the header's VarInt *before* we write the packet_len. - let header_byte_len = VarUInt(fourteen_bit_header).written_size(); - - let packet_payload_len = packet_payload.len() as u32; - // total_content_length is the length of the header VarInt bytes + payload bytes. - let total_content_length = header_byte_len as u32 + packet_payload_len; - - // 3. Write packet_len as VarInt - // Note: Your `VarInt` struct takes `i32`, but lengths are typically `u32`. - // Ensure consistency in your actual `VarInt` definition. - // For this example, I'll cast `total_content_length` to `i32`. - writer + inner_buffer .write_var_uint(&VarUInt(total_content_length)) .unwrap(); + inner_buffer.write_var_uint(&header_varint).unwrap(); + inner_buffer.write_all(packet_payload)?; - // 4. Write the combined 14-bit header_value as VarInt - writer - .write_var_uint(&VarUInt(fourteen_bit_header)) - .unwrap(); + // Handle Outer Container + writer.write_u8(0xfe).unwrap(); // Bedrock Game Packet Header - // 5. Write the payload - writer.write_all(packet_payload) + if let Some((_threshold, level)) = self.compression { + // Write Compression Method (0x00 for Zlib) + writer.write_u8(0x00).unwrap(); + + let mut encoder = DeflateEncoder::new(Vec::new(), Compression::new(level)); + encoder.write_all(&inner_buffer).unwrap(); + let compressed_data = encoder.finish().unwrap(); + + writer.write_all(&compressed_data)?; + } else { + writer.write_all(&inner_buffer)?; + } + + Ok(()) } pub async fn write_packet( @@ -175,6 +165,39 @@ impl UDPNetworkEncoder { } } -#[derive(Error, Debug)] -#[error("Invalid compression Level")] -pub struct CompressionLevelError; +#[cfg(test)] +mod tests { + use super::*; + use crate::bedrock::packet_decoder::UDPNetworkDecoder; + use std::io::Cursor; + + #[tokio::test] + async fn bedrock_compression_cycle() { + let mut encoder = UDPNetworkEncoder::new(); + encoder.set_compression((256, 6)); + + let packet_id = 1; + let payload = b"Hello Bedrock Compression!"; + let mut encoded_buf = Vec::new(); + + encoder + .write_game_packet( + packet_id, + SubClient::Main, + SubClient::Main, + payload, + &mut encoded_buf, + ) + .unwrap(); + + let mut decoder = UDPNetworkDecoder::new(); + decoder.set_compression(256); + + let decompressed_payload = decoder.get_packet_payload(encoded_buf).await.unwrap(); + let mut cursor = Cursor::new(decompressed_payload); + let raw_packet = decoder.get_game_packet(&mut cursor).unwrap(); + + assert_eq!(raw_packet.id, packet_id as i32); + assert_eq!(raw_packet.payload.as_ref(), payload); + } +} diff --git a/pumpkin/src/lib.rs b/pumpkin/src/lib.rs index e94de7a04..fd10ec582 100644 --- a/pumpkin/src/lib.rs +++ b/pumpkin/src/lib.rs @@ -418,6 +418,7 @@ impl PumpkinServer { } } + #[expect(clippy::too_many_lines)] pub async fn unified_listener_task( &self, master_client_id_counter: &mut u64, @@ -501,24 +502,33 @@ impl PumpkinServer { let be_clients = bedrock_clients.clone(); let mut clients_guard = bedrock_clients.lock().await; - if let Some(client) = clients_guard.get(&client_addr) { - let client = client.clone(); - let reader = Cursor::new(udp_buf[..len].to_vec()); - let server = self.server.clone(); - tasks.spawn(async move { - client.process_packet(&server, reader).await; - }); - } else if let Some(sock) = self.udp_socket.as_ref() - && let Ok(packet) = BedrockClient::is_connection_request(&mut Cursor::new(&udp_buf[4..len])) { - *master_client_id_counter += 1; - let platform = BedrockClient::new(sock.clone(), client_addr, be_clients); - platform.handle_connection_request(packet).await; - let platform = Arc::new(platform); - platform.start_outgoing_packet_task(); - clients_guard.insert(client_addr, platform); - } + let client = clients_guard.entry(client_addr).or_insert_with(|| { + *master_client_id_counter += 1; + + let new_client = Arc::new(BedrockClient::new( + self.udp_socket.as_ref().unwrap().clone(), + client_addr, + be_clients + )); + + new_client.start_outgoing_packet_task(); + new_client + }).clone(); + + let reader = udp_buf[..len].to_vec(); + let server = self.server.clone(); + tasks.spawn(async move { + client.process_packet(&server, reader).await; + }); + } else if let Some(sock) = self.udp_socket.as_ref() { - let _ = BedrockClient::handle_offline_packet(&self.server, id, &mut Cursor::new(&udp_buf[1..len]), client_addr, sock).await; + let _ = BedrockClient::handle_offline_packet( + &self.server, + id, + &mut Cursor::new(&udp_buf[1..len]), + client_addr, + sock + ).await; } } } diff --git a/pumpkin/src/net/bedrock/connection.rs b/pumpkin/src/net/bedrock/connection.rs index 812cd229c..4c6bcf06d 100644 --- a/pumpkin/src/net/bedrock/connection.rs +++ b/pumpkin/src/net/bedrock/connection.rs @@ -1,45 +1,14 @@ -use std::{ - io::{Cursor, Error}, - net::{Ipv4Addr, SocketAddr, SocketAddrV4}, - time::UNIX_EPOCH, -}; +use std::time::UNIX_EPOCH; use pumpkin_protocol::bedrock::{ RakReliability, - client::raknet::connection::{CConnectedPong, CConnectionRequestAccepted}, - server::raknet::connection::{SConnectedPing, SConnectionRequest, SNewIncomingConnection}, + client::raknet::connection::CConnectedPong, + server::raknet::connection::{SConnectedPing, SNewIncomingConnection}, }; -use pumpkin_protocol::{codec::u24, serial::PacketRead}; use crate::net::bedrock::BedrockClient; impl BedrockClient { - pub fn is_connection_request(reader: &mut Cursor<&[u8]>) -> Result { - //Must be reliable and non split - if u8::read(reader)? == 0x40 { - u16::read_be(reader)?; - //skip reliable seq - u24::read(reader)?; - SConnectionRequest::read(reader) - } else { - Err(Error::other("")) - } - } - - pub async fn handle_connection_request(&self, packet: SConnectionRequest) { - self.send_framed_packet( - &CConnectionRequestAccepted::new( - self.address, - 0, - [SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 19132)); 10], - packet.time, - UNIX_EPOCH.elapsed().unwrap().as_millis() as u64, - ), - RakReliability::Unreliable, - ) - .await; - } - pub const fn handle_new_incoming_connection(&self, _packet: &SNewIncomingConnection) { // self.connection_state.store(ConnectionState::Login); } diff --git a/pumpkin/src/net/bedrock/login.rs b/pumpkin/src/net/bedrock/login.rs index d15bf298e..6b7d2038f 100644 --- a/pumpkin/src/net/bedrock/login.rs +++ b/pumpkin/src/net/bedrock/login.rs @@ -2,7 +2,6 @@ use crate::{ net::{ClientPlatform, DisconnectReason, GameProfile, PlayerConfig, bedrock::BedrockClient}, server::Server, }; -use pumpkin_config::networking::compression::CompressionInfo; use pumpkin_protocol::bedrock::server::{ login::ClientData, resource_pack_response::SResourcePackResponse, }; @@ -84,7 +83,11 @@ fn verify_oidc_token_path( } impl BedrockClient { - pub async fn handle_request_network_settings(&self, packet: SRequestNetworkSettings) { + pub async fn handle_request_network_settings( + &self, + packet: SRequestNetworkSettings, + server: &Server, + ) { if packet.protocol_version < CURRENT_BEDROCK_MC_PROTOCOL as i32 { self.send_game_packet(&CPlayStatus::OutdatedClient).await; return; @@ -92,9 +95,21 @@ impl BedrockClient { self.send_game_packet(&CPlayStatus::OutdatedServer).await; return; } - self.send_game_packet(&CNetworkSettings::new(0, 0, false, 0, 0.0)) - .await; - self.set_compression(CompressionInfo::default()).await; + let compression = server + .advanced_config + .networking + .bedrock_compression + .info + .clone(); + self.send_game_packet(&CNetworkSettings::new( + compression.threshold as u16, + 0, + false, + 0, + 0.0, + )) + .await; + self.set_compression(compression).await; } pub async fn handle_login(self: &Arc, packet: SLogin, server: &Server) -> Option<()> { diff --git a/pumpkin/src/net/bedrock/mod.rs b/pumpkin/src/net/bedrock/mod.rs index a55ae43df..38181fbb3 100644 --- a/pumpkin/src/net/bedrock/mod.rs +++ b/pumpkin/src/net/bedrock/mod.rs @@ -1,11 +1,14 @@ pub mod play; +use crossbeam::atomic::AtomicCell; use std::{ collections::{BTreeMap, HashMap, HashSet}, io::{Cursor, Error, Write}, + net::{Ipv4Addr, SocketAddrV4}, sync::{ Arc, atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering}, }, + time::UNIX_EPOCH, }; use tracing::{debug, error, warn}; @@ -17,7 +20,9 @@ use pumpkin_protocol::{ bedrock::{ MTU, RAKNET_ACK, RAKNET_GAME_PACKET, RAKNET_NACK, RakReliability, SubClient, ack::Acknowledge, - client::disconnect_player::CDisconnectPlayer, + client::{ + disconnect_player::CDisconnectPlayer, raknet::connection::CConnectionRequestAccepted, + }, frame_set::{Frame, FrameSet}, packet_decoder::UDPNetworkDecoder, packet_encoder::UDPNetworkEncoder, @@ -91,6 +96,7 @@ pub struct BedrockClient { output_ordered_index: AtomicU32, /// An notifier that is triggered when this client is closed. close_token: CancellationToken, + last_seen: Arc>, /// Store Fragments until the packet is complete compounds: Arc>>>>, //input_sequence_number: AtomicU32, @@ -129,6 +135,7 @@ impl BedrockClient { output_ordered_index: AtomicU32::new(0), compounds: Arc::new(Mutex::new(HashMap::new())), close_token: CancellationToken::new(), + last_seen: Arc::new(AtomicCell::new(std::time::Instant::now())), received_sequences: Mutex::new(HashSet::new()), pending_acks: Mutex::new(Vec::new()), unacked_outgoing_frames: Mutex::new(HashMap::new()), @@ -153,6 +160,13 @@ impl BedrockClient { while !client.close_token.is_cancelled() { tokio::select! { _ = interval.tick() => { + // Check for timeout (10 seconds) + if client.last_seen.load().elapsed() > std::time::Duration::from_secs(10) { + debug!("Bedrock client {} timed out", client.address); + client.close().await; + break; + } + // Flush ACKs let mut pending = client.pending_acks.lock().await; if !pending.is_empty() { @@ -205,11 +219,9 @@ impl BedrockClient { }); } - pub async fn process_packet(self: &Arc, server: &Arc, packet: Cursor>) { - let packet = self.get_packet_payload(packet).await; - if let Some(packet) = packet - && let Err(error) = self.handle_packet_payload(server, packet).await - { + pub async fn process_packet(self: &Arc, server: &Arc, packet: Vec) { + self.last_seen.store(std::time::Instant::now()); + if let Err(error) = self.handle_packet_payload(server, packet).await { error!( "Failed to handle packet payload for {}: {}", self.address, error @@ -422,10 +434,13 @@ impl BedrockClient { } pub async fn close(&self) { + if self.is_closed() { + return; + } self.close_token.cancel(); + self.be_clients.lock().await.remove(&self.address); self.tasks.close(); self.tasks.wait().await; - self.be_clients.lock().await.remove(&self.address); if let Some(player) = self.player.lock().await.as_ref() { player.remove().await; @@ -457,7 +472,7 @@ impl BedrockClient { pub async fn handle_packet_payload( self: &Arc, server: &Arc, - packet: Bytes, + packet: Vec, ) -> Result<(), Error> { let reader = &mut Cursor::new(packet); @@ -468,7 +483,7 @@ impl BedrockClient { RAKNET_NACK => { self.handle_nack(&Acknowledge::read(reader)?).await; } - 0x80..0x8d => { + 0x80..=0x8d => { self.handle_frame_set(server, FrameSet::read(reader)?) .await?; } @@ -630,10 +645,39 @@ impl BedrockClient { server: &Arc, payload: Vec, ) -> Result<(), Error> { - let mut payload = Cursor::new(payload); - let id = u8::read(&mut payload)?; - self.handle_raknet_packet(server, i32::from(id), payload) - .await + if payload.is_empty() { + return Ok(()); + } + let id = payload[0]; + + if id == RAKNET_GAME_PACKET as u8 { + // Decompress the batch + let decompressed_payload = self + .get_packet_payload(payload) + .await + .ok_or_else(|| Error::other("Failed to decompress game packet batch"))?; + + // Loop through the decompressed buffer to extract ALL batched packets + let mut cursor = Cursor::new(decompressed_payload); + + while (cursor.position() as usize) < cursor.get_ref().len() { + let game_packet = self + .network_reader + .lock() + .await + .get_game_packet(&mut cursor) + .map_err(|e| Error::other(e.to_string()))?; + + self.handle_game_packet(server, game_packet).await?; + } + } else { + // It's an internal RakNet message (like SConnectedPing) + let mut cursor = Cursor::new(payload); + let _id = u8::read(&mut cursor)?; // consume ID byte + self.handle_raknet_packet(i32::from(id), cursor).await?; + } + + Ok(()) } async fn handle_game_packet( @@ -645,8 +689,11 @@ impl BedrockClient { let payload = &mut Cursor::new(&packet.payload); let result = match packet.id { SRequestNetworkSettings::PACKET_ID => { - self.handle_request_network_settings(SRequestNetworkSettings::read(payload)?) - .await; + self.handle_request_network_settings( + SRequestNetworkSettings::read(payload)?, + server, + ) + .await; Ok(()) } SLogin::PACKET_ID => { @@ -735,14 +782,26 @@ impl BedrockClient { async fn handle_raknet_packet( self: &Arc, - server: &Arc, packet_id: i32, mut payload: Cursor>, ) -> Result<(), Error> { let reader = &mut payload; match packet_id { - // The client sends this multiple times and some arrive after we already made the connection - SConnectionRequest::PACKET_ID => (), + SConnectionRequest::PACKET_ID => { + let request = SConnectionRequest::read(reader)?; + + self.send_framed_packet( + &CConnectionRequestAccepted::new( + self.address, + 0, + [SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 19132)); 10], + request.time, + UNIX_EPOCH.elapsed().unwrap().as_millis() as u64, + ), + RakReliability::Unreliable, + ) + .await; + } SNewIncomingConnection::PACKET_ID => { self.handle_new_incoming_connection(&SNewIncomingConnection::read(reader)?); } @@ -753,17 +812,6 @@ impl BedrockClient { SDisconnect::PACKET_ID => { self.close().await; } - - RAKNET_GAME_PACKET => { - let game_packet = self - .network_reader - .lock() - .await - .get_game_packet(payload) - .map_err(|e| Error::other(e.to_string()))?; - - self.handle_game_packet(server, game_packet).await?; - } _ => { warn!("Bedrock: Received Unknown RakNet Online packet: {packet_id}"); } @@ -815,7 +863,7 @@ impl BedrockClient { self.close_token.cancelled().await; } - pub async fn get_packet_payload(&self, packet: Cursor>) -> Option { + pub async fn get_packet_payload(&self, packet: Vec) -> Option> { let mut network_reader = self.network_reader.lock().await; tokio::select! { () = self.await_close_interrupt() => { diff --git a/pumpkin/src/net/java/login.rs b/pumpkin/src/net/java/login.rs index bfae21f57..1bddcc6d6 100644 --- a/pumpkin/src/net/java/login.rs +++ b/pumpkin/src/net/java/login.rs @@ -85,7 +85,7 @@ impl JavaClient { profile_actions: None, }; - if server.advanced_config.networking.packet_compression.enabled { + if server.advanced_config.networking.java_compression.enabled { self.enable_compression(server).await; } @@ -198,7 +198,7 @@ impl JavaClient { let compression = server .advanced_config .networking - .packet_compression + .java_compression .info .clone(); // We want to wait until we have sent the compression packet to the client