You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: packet_io.rs
    path of file: ./repos/valence/tools/packet_inspector/src
    
    the code of the file until where you have to start completion: use std::io;
use std::io::ErrorKind;

use anyhow::ensure;
use bytes::{BufMut, BytesMut};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use valence_protocol::decode::{PacketDecoder, PacketFrame};
use valence_protocol::encode::PacketEncoder;
use valence_protocol::{CompressionThreshold, Encode, VarInt, MAX_PACKET_SIZE};

pub(crate) struct PacketIoReader {
    reader: tokio::io::ReadHalf<tokio::net::TcpStream>,
    dec: PacketDecoder,
    threshold: CompressionThreshold,
}

impl PacketIoReader {
    pub(crate) async fn recv_packet_raw(&mut self) -> anyhow::Result<PacketFrame> {
        loop {
            if let Some(frame) = self.dec.try_next_packet()? {
                // self.logger
                //     .log("Unknown".to_string(), self.direction.clone(), frame.clone());

                return Ok(frame);
            }

            self.dec.reserve(READ_BUF_SIZE);
            let mut buf = self.dec.take_capacity();

            if self.reader.read_buf(&mut buf).await? == 0 {
                return Err(io::Error::from(ErrorKind::UnexpectedEof).into());
            }

            // This should always be an O(1) unsplit because we reserved space earlier and
            // the call to `read_buf` shouldn't have grown the allocation.
            self.dec.queue_bytes(buf);
        }
    }

    #[allow(dead_code)]
    pub(crate) fn set_compression(&mut self, threshold: CompressionThreshold) {
        self.threshold = threshold;
        self.dec.set_compression(threshold);
    }
}

pub(crate) struct PacketIoWriter {
    writer: tokio::io::WriteHalf<tokio::net::TcpStream>,
    enc: PacketEncoder,
    threshold: CompressionThreshold,
}

impl PacketIoWriter {
    /*
      No  | Packet Length |  VarInt     | Length of (Data Length) + Compressed length of (Packet ID + Data)
      No  | Data Length   |  VarInt     | Length of uncompressed (Packet ID + Data) or 0
      Yes | Packet ID	  |  VarInt     | zlib compressed packet ID (see the sections below)
      Yes | Data          |  Byte Array | zlib compressed packet data (see the sections below)
    */
    pub(crate) async fn send_packet_raw(&mut self, frame: &PacketFrame) -> anyhow::Result<()> {
        let id_varint = VarInt(frame.id);
        let id_buf = varint_to_bytes(id_varint);

        let mut uncompressed_packet = BytesMut::new();
        uncompressed_packet.extend_from_slice(&id_buf);
        uncompressed_packet.extend_from_slice(&frame.body);
        let uncompressed_packet_length = uncompressed_packet.len();
        let uncompressed_packet_length_varint = VarInt(uncompressed_packet_length as i32);

        if self.threshold.0 >= 0 {
    
    pub(crate) async fn send_packet_raw(&mut self, frame: &PacketFrame) -> anyhow::Result<()> {
        let id_varint = VarInt(frame.id);
        let id_buf = varint_to_bytes(id_varint);

        let mut uncompressed_packet = BytesMut::new();
        uncompressed_packet.extend_from_slice(&id_buf);
        uncompressed_packet.extend_from_slice(&frame.body);
        let uncompressed_packet_length = uncompressed_packet.len();
        let uncompressed_packet_length_varint = VarInt(uncompressed_packet_length as i32);

        if self.threshold.0 >= 0 {
            if uncompressed_packet_length > self.threshold.0 as usize {
                use std::io::Read;

                use flate2::bufread::ZlibEncoder;
                use flate2::Compression;

                let mut z = ZlibEncoder::new(&uncompressed_packet[..], Compression::new(4));
                let mut compressed = Vec::new();

                let data_len_size = uncompressed_packet_length_varint.written_size();

                let packet_len = data_len_size + z.read_to_end(&mut compressed)?;

                ensure!(
                    packet_len <= MAX_PACKET_SIZE as usize,
                    "packet exceeds maximum length"
                );

                drop(z);

                self.enc
                    .append_bytes(&varint_to_bytes(VarInt(packet_len as i32)));

                self.enc
                    .append_bytes(&varint_to_bytes(uncompressed_packet_length_varint));

                self.enc.append_bytes(&compressed);

                let bytes = self.enc.take();

                self.writer.write_all(&bytes).await?;
                self.writer.flush().await?;

                // now we need to compress the packet.
            } else {
                // no need to compress, but we do need to inject a zero
                let empty = VarInt(0);

                let data_len_size = empty.written_size();
                let packet_len = data_len_size + uncompressed_packet_length;

                self.enc
                    .append_bytes(&varint_to_bytes(VarInt(packet_len as i32)));
                self.enc.append_bytes(&varint_to_bytes(empty));
                self.enc.append_bytes(&uncompressed_packet);
                let bytes = self.enc.take();
                self.writer.write_all(&bytes).await?;
                self.writer.flush().await?;
            }

            return Ok(());
        }

        let length = varint_to_bytes(VarInt(uncompressed_packet_length as i32));

        // the frame should be uncompressed at this point.
        self.enc.append_bytes(&length);
        self.enc.append_bytes(&uncompressed_packet);

        let bytes = self.enc.take();

        self.writer.write_all(&bytes).await?;

        Ok(())
    }