implement more of reader

Cargo.lock

@@ -559,9 +559,21 @@ dependencies = [
  "mio",
  "num_cpus",
  "pin-project-lite",
+ "tokio-macros",
  "winapi",
 ]
 
+[[package]]
+name = "tokio-macros"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c9efc1aba077437943f7515666aa2b882dfabfbfdf89c819ea75a8d6e9eaba5e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.6.9"

minecraft-protocol/Cargo.toml

@@ -10,6 +10,6 @@ async-recursion = "^0.3.2"
 byteorder = "^1.4.3"
 bytes = "^1.1.0"
 thiserror = "^1.0.30"
-tokio = {version = "^1.14.0", features = ["io-util", "net"]}
+tokio = {version = "^1.14.0", features = ["io-util", "net", "macros"]}
 tokio-util = "^0.6.9"
 trust-dns-resolver = "^0.20.3"

minecraft-protocol/src/connection.rs

@@ -1,7 +1,9 @@
 use crate::{mc_buf, packets::Packet, ServerIpAddress};
 use bytes::BytesMut;
+use std::io::{Cursor, Read, Seek, SeekFrom, Write};
+use tokio::io::AsyncWriteExt;
 use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
+    io::{AsyncReadExt, BufReader, BufWriter},
     net::TcpStream,
 };
 
@@ -36,11 +38,31 @@ impl Connection {
         })
     }
 
-    pub async fn read_packet(&mut self) {
-        // the first thing minecraft sends us is the length as a varint, which can be up to 5 bytes
-        let mut buf = Vec::new();
-        self.stream.read_buf(&mut buf).await;
-        mc_buf::read_varint(buf)
+    pub async fn read_packet(&mut self) -> Result<(), String> {
+        // what this does:
+        // 1. reads the first 5 bytes, probably only some of this will be used to get the packet length
+        // 2. how much we should read = packet length - 5
+        // 3. read the rest of the packet and add it to the cursor
+
+        // 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(5 * 1024, &mut self.stream);
+
+        let packet_size = mc_buf::read_varint(&mut buf).await?;
+
+        println!("packet size from varint: {}", packet_size);
+
+        let packet_id = mc_buf::read_byte(&mut buf).await?;
+
+        // read the rest of the packet
+        let mut packet_data = Vec::with_capacity(packet_size as usize);
+        buf.read_buf(&mut packet_data).await.unwrap();
+        println!(
+            "packet id {}: {}",
+            packet_id,
+            String::from_utf8(packet_data.clone()).unwrap()
+        );
+
+        Ok(())
     }
 
     /// Write a packet to the server

minecraft-protocol/src/mc_buf.rs

@@ -1,9 +1,9 @@
 //! Utilities for reading and writing for the Minecraft protocol
 
-use std::io::Cursor;
+use std::io::{Cursor, Write};
 
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
-use tokio::io::AsyncReadExt;
+use tokio::io::{AsyncRead, AsyncReadExt, BufReader};
 
 // const MAX_VARINT_SIZE: u32 = 5;
 // const MAX_VARLONG_SIZE: u32 = 10;
@@ -11,7 +11,9 @@ use tokio::io::AsyncReadExt;
 const MAX_STRING_LENGTH: u16 = 32767;
 // const MAX_COMPONENT_STRING_LENGTH: u32 = 262144;
 
-pub async fn read_byte(buf: &mut Cursor<Vec<u8>>) -> Result<u8, String> {
+pub async fn read_byte<T: AsyncRead + std::marker::Unpin>(
+    buf: &mut BufReader<T>,
+) -> Result<u8, String> {
     match AsyncReadExt::read_u8(buf).await {
         Ok(r) => Ok(r),
         Err(_) => Err("Error reading byte".to_string()),
@@ -19,41 +21,63 @@ pub async fn read_byte(buf: &mut Cursor<Vec<u8>>) -> Result<u8, String> {
 }
 
 pub fn write_byte(buf: &mut Vec<u8>, n: u8) {
-    buf.write_u8(n).unwrap();
+    WriteBytesExt::write_u8(buf, n).unwrap();
 }
 
 pub fn write_bytes(buf: &mut Vec<u8>, bytes: &[u8]) {
     buf.extend_from_slice(bytes);
 }
 
-pub async fn read_varint(buf: &mut Cursor<Vec<u8>>) -> Result<u32, String> {
-    let mut value: u32 = 0;
-    let mut length: u32 = 0;
-    let mut current_byte: u8;
-
-    loop {
-        current_byte = read_byte(buf).await?;
-        value |= ((current_byte & 0x7F) as u32) << (length * 7);
-
-        length += 1;
-        if length > 5 {
-            return Err("VarInt too big".to_string());
+// fast varints stolen from https://github.com/luojia65/mc-varint/blob/master/src/lib.rs#L67
+pub async fn read_varint<T: AsyncRead + std::marker::Unpin>(
+    buf: &mut BufReader<T>,
+) -> Result<u32, String> {
+    let mut buffer = [0];
+    let mut ans = 0;
+    for i in 0..4 {
+        buf.read_exact(&mut buffer)
+            .await
+            .or_else(|_| Err("Invalid VarInt".to_string()))?;
+        ans |= ((buffer[0] & 0b0111_1111) as u32) << 7 * i;
+        if buffer[0] & 0b1000_0000 == 0 {
+            break;
         }
+    }
+    Ok(ans)
+}
 
-        if (value & 0x80) != 0x80 {
-            return Ok(value);
+pub fn write_varint(buf: &mut Vec<u8>, mut value: u32) {
+    let mut buffer = [0];
+    while value != 0 {
+        buffer[0] = (value & 0b0111_1111) as u8;
+        value = (value >> 7) & (u32::max_value() >> 6);
+        if value != 0 {
+            buffer[0] |= 0b1000_0000;
         }
+        buf.write(&buffer).unwrap();
     }
 }
 
-pub fn write_varint(buf: &mut Vec<u8>, mut n: u32) {
-    loop {
-        if (n & 0xFFFFFF80) == 0 {
-            write_byte(buf, n as u8);
-            return ();
-        }
-        write_byte(buf, (n & 0x7F | 0x80) as u8);
-        n >>= 7;
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn test_write_varint() {
+        let mut buf = Vec::new();
+        write_varint(&mut buf, 123456);
+        assert_eq!(buf, vec![192, 196, 7]);
+    }
+
+    #[tokio::test]
+    async fn test_read_varint() {
+        let mut buf = BufReader::new(Cursor::new(vec![192, 196, 7]));
+        assert_eq!(read_varint(&mut buf).await.unwrap(), 123456);
+    }
+
+    #[tokio::test]
+    async fn test_read_varint_longer() {
+        let mut buf = BufReader::new(Cursor::new(vec![138, 56, 0, 135, 56, 123]));
+        assert_eq!(read_varint(&mut buf).await.unwrap(), 7178);
     }
 }
 
@@ -74,5 +98,5 @@ pub fn write_utf(buf: &mut Vec<u8>, string: &String) {
 }
 
 pub fn write_short(buf: &mut Vec<u8>, n: u16) {
-    buf.write_u16::<BigEndian>(n).unwrap();
+    WriteBytesExt::write_u16::<BigEndian>(buf, n).unwrap();
 }

minecraft-protocol/src/server_status_pinger.rs

@@ -4,7 +4,6 @@ use crate::{
     packets::{ClientIntentionPacket, ConnectionProtocol, ServerboundStatusRequestPacket},
     resolver, ServerAddress,
 };
-use tokio::io::AsyncReadExt;
 
 pub async fn ping_server(address: &ServerAddress) -> Result<(), String> {
     let resolved_address = resolver::resolve_address(&address).await?;
@@ -23,25 +22,29 @@ pub async fn ping_server(address: &ServerAddress) -> Result<(), String> {
     .await;
     conn.send_packet(&ServerboundStatusRequestPacket {}).await;
 
-    let data = mc_buf::read_varint(conn.stream);
-    println!("data {}", data);
+    conn.read_packet().await.unwrap();
 
-    // log what the server sends back
-    loop {
-        if 0 == conn.stream.read_buf(&mut conn.buffer).await.unwrap() {
-            // The remote closed the connection. For this to be a clean
-            // shutdown, there should be no data in the read buffer. If
-            // there is, this means that the peer closed the socket while
-            // sending a frame.
+    Ok(())
 
-            // log conn.buffer
-            println!("{:?}", conn.buffer);
-            if conn.buffer.is_empty() {
-                println!("buffer is empty ok");
-                return Ok(());
-            } else {
-                return Err("connection reset by peer".into());
-            }
-        }
-    }
+    // let data = mc_buf::read_varint(conn.stream);
+    // println!("data {}", data);
+
+    // // log what the server sends back
+    // loop {
+    //     if 0 == conn.stream.read_buf(&mut conn.buffer).await.unwrap() {
+    //         // The remote closed the connection. For this to be a clean
+    //         // shutdown, there should be no data in the read buffer. If
+    //         // there is, this means that the peer closed the socket while
+    //         // sending a frame.
+
+    //         // log conn.buffer
+    //         println!("{:?}", conn.buffer);
+    //         if conn.buffer.is_empty() {
+    //             println!("buffer is empty ok");
+    //             return Ok(());
+    //         } else {
+    //             return Err("connection reset by peer".into());
+    //         }
+    //     }
+    // }
 }