refactor(proto): remove lifetime from read, 1.8 typed decoding

Author: andybrwr

Cargo.lock

@@ -17,7 +17,7 @@
  * <https://www.gnu.org/licenses/>.
  */
 
-use proc_macro::{Span, TokenStream};
+use proc_macro::TokenStream;
 use quote::quote;
 use syn::{parse_macro_input, parse_quote, DeriveInput, Fields, Ident, Index, Lit, Path};
 
@@ -41,10 +41,9 @@ pub fn derive_packet(input: TokenStream) -> TokenStream {
         attr.parse_nested_meta(|meta| {
             if meta.path.is_ident("id") {
                 let lit = meta.value()?.parse()?;
-                match lit {
-                    Lit::Str(i) => {
-                        id = Some(i);
-                    }
+                id = match lit {
+                    Lit::Str(i) => Some(quote!(crawlspace_proto::PacketId::String(#i))),
+                    Lit::Int(i) => Some(quote!(crawlspace_proto::PacketId::Numeric(#i))),
                     _ => panic!("attribute value `id` must be a string"),
                 }
             } else if meta.path.is_ident("state") {
@@ -88,28 +87,20 @@ pub fn derive_packet(input: TokenStream) -> TokenStream {
 
     let id = id.expect("id must be provided for packet");
     let state = state.expect("state must be provided for packet");
-    let direction = Ident::new(
-        direction.expect("direction must be provided for packet"),
-        Span::call_site().into(),
-    );
 
     let name = input.ident;
     let where_clause = input.generics.where_clause.clone();
     let generics = input.generics;
 
     quote! {
-    impl #generics Packet for #name #generics #where_clause {
-        fn id() -> &'static str {
+    impl #generics crawlspace_proto::Packet for #name #generics #where_clause {
+        fn packet_id() -> crawlspace_proto::PacketId {
             #id
         }
 
-        fn state() -> PacketState {
+        fn packet_state() -> crawlspace_proto::ConnectionState {
             #state
         }
-
-        fn direction() -> PacketDirection {
-            PacketDirection::#direction
-        }
     }
     }
     .into()
@@ -118,12 +109,12 @@ pub fn derive_packet(input: TokenStream) -> TokenStream {
 /// Automatically implements "straight-across" encoding for the given struct, i.e. fields are
 /// serialized in order as is. Supports #[varint] and #[varlong] attributes on integer types to
 /// serialize as those formats instead.
-#[proc_macro_derive(Encode, attributes(varint, varlong))]
-pub fn derive_encode(input: TokenStream) -> TokenStream {
+#[proc_macro_derive(Write, attributes(varint, varlong))]
+pub fn derive_write(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
 
     let syn::Data::Struct(data) = input.data else {
-        panic!("Can only derive Encode on a struct");
+        panic!("Can only derive Write on a struct");
     };
 
     let name = input.ident;
@@ -143,19 +134,19 @@ pub fn derive_encode(input: TokenStream) -> TokenStream {
                     .any(|attr| attr.meta.path().is_ident("varint"))
                 {
                     fields_encoded.extend(quote! {
-                        VarInt(self.#field_name as i32).encode(&mut w)?;
+                        VarInt(self.#field_name as i32).write(w)?;
                     });
                 } else if field
                     .attrs
                     .iter()
                     .any(|attr| attr.meta.path().is_ident("varlong"))
                 {
                     fields_encoded.extend(quote! {
-                        VarLong(self.#field_name as i64).encode(&mut w)?;
+                        VarLong(self.#field_name as i64).write(w)?;
                     });
                 } else {
                     fields_encoded.extend(quote! {
-                        self.#field_name.encode(&mut w)?;
+                        self.#field_name.write(w)?;
                     });
                 }
             }
@@ -170,19 +161,19 @@ pub fn derive_encode(input: TokenStream) -> TokenStream {
                     .any(|attr| attr.meta.path().is_ident("varint"))
                 {
                     fields_encoded.extend(quote! {
-                        VarInt(self.#i as i32).encode(&mut w)?;
+                        VarInt(self.#i as i32).write(w)?;
                     });
                 } else if field
                     .attrs
                     .iter()
                     .any(|attr| attr.meta.path().is_ident("varlong"))
                 {
                     fields_encoded.extend(quote! {
-                        VarLong(self.#i as i64).encode(&mut w)?;
+                        VarLong(self.#i as i64).write(w)?;
                     });
                 } else {
                     fields_encoded.extend(quote! {
-                        self.#i.encode(&mut w)?;
+                        self.#i.write(w)?;
                     });
                 }
             }
@@ -191,8 +182,8 @@ pub fn derive_encode(input: TokenStream) -> TokenStream {
     }
 
     quote! {
-        impl #generics Encode for #name #generics #where_clause {
-            fn encode(&self, mut w: impl std::io::Write) -> color_eyre::Result<()> {
+        impl #generics Write for #name #generics #where_clause {
+            fn encode(&self, w: &mut impl std::io::Write) -> crawlspace_proto::Result<()> {
                 #fields_encoded
 
                 Ok(())
@@ -205,12 +196,12 @@ pub fn derive_encode(input: TokenStream) -> TokenStream {
 /// Automatically implements "straight-across" decoding for the given struct, i.e. fields are
 /// deserialized in order as is. Supports #[decode_as(type)] to deserialize according to a different type.
 /// uses TryInto to convert to the expected type where necessary.
-#[proc_macro_derive(Decode, attributes(decode_as))]
-pub fn derive_decode(input: TokenStream) -> TokenStream {
+#[proc_macro_derive(Read, attributes(decode_as))]
+pub fn derive_read(input: TokenStream) -> TokenStream {
     let input = parse_macro_input!(input as DeriveInput);
 
     let syn::Data::Struct(data) = input.data else {
-        panic!("Can only derive Decode on a struct");
+        panic!("Can only derive Read on a struct");
     };
 
     let name = input.ident;
@@ -223,8 +214,6 @@ pub fn derive_decode(input: TokenStream) -> TokenStream {
                 let field_name = field.ident.expect("couldn't get ident for named field");
                 let ty = field.ty;
 
-                let wrapped = format!("for field {field_name} in {name}");
-
                 if let Some(attr) = field
                     .attrs
                     .iter()
@@ -235,14 +224,11 @@ pub fn derive_decode(input: TokenStream) -> TokenStream {
                         .expect("decode_as value must be a Path");
 
                     field_tokens.extend(quote! {
-                        #field_name: <#ty as Decode>::decode(r)
-                            .wrap_err(#wrapped)?
-                            .try_into()?,
+                        #field_name: <#ty as Read>::read(r)?.try_into()?,
                     });
                 } else {
                     field_tokens.extend(quote! {
-                        #field_name: <#ty as Decode>::decode(r)
-                            .wrap_err(#wrapped)?,
+                        #field_name: <#ty as Read>::read(r)?,
                     });
                 }
             }
@@ -254,11 +240,9 @@ pub fn derive_decode(input: TokenStream) -> TokenStream {
         }
         Fields::Unnamed(fields) => {
             let mut field_tokens = proc_macro2::TokenStream::new();
-            for (i, field) in fields.unnamed.into_iter().enumerate() {
+            for field in fields.unnamed.into_iter() {
                 let ty = field.ty;
 
-                let wrapped = format!("for field {i} in {name}");
-
                 if let Some(attr) = field
                     .attrs
                     .iter()
@@ -269,13 +253,12 @@ pub fn derive_decode(input: TokenStream) -> TokenStream {
                         .expect("decode_as value must be a Path");
 
                     field_tokens.extend(quote! {
-                        <#ty as Decode>::decode(r)
-                            .wrap_err(#wrapped)?
+                        <#ty as Read>::read(r)?
                             .try_into()?,
                     });
                 } else {
                     field_tokens.extend(quote! {
-                        <#ty as Decode>::decode(r).wrap_err(#wrapped)?,
+                        <#ty as Read>::read(r)?,
                     });
                 }
             }
@@ -289,18 +272,12 @@ pub fn derive_decode(input: TokenStream) -> TokenStream {
     let struct_generics = input.generics;
     let where_clause = struct_generics.where_clause.clone();
 
-    let mut impl_generics = struct_generics.clone();
-    if impl_generics.lifetimes().count() == 0 {
-        impl_generics.params.push(parse_quote!('a));
-    }
-
     quote! {
-        impl #impl_generics Decode #impl_generics for #name #struct_generics #where_clause {
-            fn decode(r: &mut &'a [u8]) -> color_eyre::Result<Self>
+        impl #struct_generics crawlspace_proto::Read for #name #struct_generics #where_clause {
+            fn read(r: &mut impl std::io::Read) -> crawlspace_proto::Result<Self>
             where
                 Self: Sized,
             {
-                use color_eyre::eyre::WrapErr;
                 Ok(#struct_tokens)
             }
         }

crawlspace-macro/src/lib.rs

@@ -4,5 +4,9 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-bytes = "1.10.1"
+crawlspace-macro = { path = "../crawlspace-macro" }
 crawlspace-proto = { path = "../crawlspace-proto" }
+
+bytes = "1.10.1"
+tokio = { version = "1.47.1", features = ["io-util", "net", "time"] }
+tracing = "0.1.41"

crawlspace-proto-1_8/Cargo.toml

@@ -0,0 +1,10 @@
+use crawlspace_macro::{Packet, Read};
+use crawlspace_proto::{ConnectionState, PacketId::Numeric};
+
+#[derive(Packet, Read)]
+#[packet(
+    id = "Numeric(0x00)",
+    state = "ConnectionState::Handshake",
+    serverbound
+)]
+pub struct HandshakeS {}

crawlspace-proto-1_8/src/handshake.rs

@@ -17,36 +17,122 @@
  * <https://www.gnu.org/licenses/>.
  */
 
-use bytes::BytesMut;
+use std::time::Duration;
+
+use bytes::{Buf, BytesMut};
 use crawlspace_proto::{
-    Packet, Read, ServerboundPacket,
+    ConnectionState, ErrorKind, Packet, Read, ServerboundPacket,
     datatypes::{VarInt, VariableNumber},
 };
+use handshake::HandshakeS;
+use tokio::{
+    io::AsyncReadExt,
+    net::tcp::{OwnedReadHalf, OwnedWriteHalf},
+};
+use tracing::{debug, warn};
+
+mod handshake;
+
+const BUF_SIZE: usize = 4096;
+const MAX_PACKET_SIZE: i32 = 2097152;
+
+type Frame = (i32, BytesMut);
 
 /// Minecraft versions 1.8-1.8.9
 /// Protocol version 47
-pub struct Protocol47<R, W> {
-    reader: R,
-    writer: W,
-    bytebuf: BytesMut,
+pub struct Protocol47 {
+    connection_state: ConnectionState,
+    read_half: OwnedReadHalf,
+    write_half: OwnedWriteHalf,
+    read_buf: BytesMut,
 }
 
-impl<R: std::io::Read, W: std::io::Write> Protocol47<R, W> {
-    pub fn new(reader: R, writer: W) -> Self {
+impl Protocol47 {
+    pub fn new(read_half: OwnedReadHalf, write_half: OwnedWriteHalf) -> Self {
         Self {
-            reader,
-            writer,
-            bytebuf: BytesMut::new(),
+            connection_state: ConnectionState::Handshake,
+            read_half,
+            write_half,
+            read_buf: BytesMut::new(),
         }
     }
 
-    fn read_packet(&mut self) -> Result<Box<dyn ServerboundPacket>, crawlspace_proto::ErrorKind> {
-        let len = VarInt::read(&mut self.reader)?;
+    pub async fn await_packet<T: ServerboundPacket>(&mut self) -> crawlspace_proto::Result<T> {
+        // TODO: skip decoding for frames we're not looking for
+        loop {
+            let frame = self.read_frame().await?;
+
+            if T::packet_id() == frame.0 {
+                return Ok(T::read(&mut &frame.1[..])?);
+            }
 
-        todo!();
+            debug!(
+                "discarding packet with id {} while awaiting {:?}",
+                frame.0,
+                T::packet_id()
+            );
+        }
     }
-}
 
-impl<R: std::io::Read, W: std::io::Write> crawlspace_proto::Protocol for Protocol47<R, W> {
-    fn handshake_player(&mut self) {}
+    pub async fn read_frame(&mut self) -> crawlspace_proto::Result<Frame> {
+        // TODO: maybe move this somewhere else? i don't know if a global timeout of 5 seconds per
+        // packet is realistic but for testing it's chill i suppose
+        tokio::time::timeout(Duration::from_secs(5), async move {
+            loop {
+                if let Some(frame) = self.try_read_next()? {
+                    return Ok(frame);
+                };
+
+                // otherwise, keep reading the rest of the packet
+                // (commented for my own sanity - these method names are awful)
+
+                // reserve more space
+                self.read_buf.reserve(BUF_SIZE);
+                // split into two parts - self.read_buf containing already read data
+                // and a new buf to fill with new data (.len returns number of bytes already held,
+                // not capacity)
+                let mut buf = self.read_buf.split_off(self.read_buf.len());
+
+                // fills the remainder of the buf (just newly allocated space)
+                if self.read_half.read_buf(&mut buf).await? == 0 {
+                    return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof).into());
+                }
+
+                // joins "bufs" back together (just moves end pointer)
+                self.read_buf.unsplit(buf);
+            }
+        })
+        .await
+        .map_err(|_| ErrorKind::Timeout)?
+    }
+
+    /// Ok(None) represents an incomplete but correctly formed packet
+    fn try_read_next(&mut self) -> crawlspace_proto::Result<Option<Frame>> {
+        let mut buf = &self.read_buf[..];
+
+        let len = VarInt::read(&mut buf)?;
+
+        if len.0 < 0 || len.0 > MAX_PACKET_SIZE {
+            return Err(ErrorKind::InvalidData(format!(
+                "Packet length {len} is out of bounds (min 0, max {MAX_PACKET_SIZE})"
+            )));
+        };
+
+        if buf.len() < len.0 as usize {
+            // packet is incomplete, keep waiting
+            return Ok(None);
+        }
+
+        // TODO: use compression here
+        self.read_buf.advance(len.len());
+        let mut data = self.read_buf.split_to(len.0 as usize);
+        buf = &data[..];
+
+        let packet_id = VarInt::read(&mut buf)?.0;
+
+        // advance to end of packet id
+        data.advance(data.len() - buf.len());
+
+        Ok(Some((packet_id, data)))
+    }
 }