thfmr-protocol 0.2.2

use {
    crate::{error::Result, message},
    log::{info, warn},
    tokio::{
        io,
        io::{AsyncBufReadExt, AsyncWriteExt},
        net,
        sync::mpsc,
    },
};

pub(crate) async fn run_read(
    socket: io::ReadHalf<net::TcpStream>,
    producer: &mpsc::Sender<message::Message>,
) -> Result<()> {
    let mut buf_reader = io::BufReader::new(socket);
    let mut buf = vec![];

    loop {
        let num_bytes = buf_reader.read_until(0, &mut buf).await?;

        let num_bytes = if buf.ends_with(&[0]) {
            num_bytes - 1
        } else {
            num_bytes
        };

        if num_bytes == 0 {
            return Ok(());
        }

        let decoded_message: serde_json::error::Result<message::Message> =
            serde_json::from_slice(&buf[0..num_bytes]);

        if let Ok(message) = decoded_message {
            info!("Received message: {:?}", message);

            producer.send(message).await?;
        } else {
            warn!(
                "Received un-decodable message: {:?}",
                String::from_utf8_lossy(&buf[0..num_bytes])
            )
        }

        buf.clear();
    }
}

pub(crate) async fn run_write(
    socket: io::WriteHalf<net::TcpStream>,
    producer: &mut mpsc::Receiver<message::Message>,
) -> Result<()> {
    let mut buf_writer = io::BufWriter::new(socket);

    loop {
        let message = producer.recv().await;

        if let None = message {
            return Ok(());
        }

        let mut packet = serde_json::to_vec(&message)?;
        packet.push(0);
        buf_writer.write_all(&packet).await?;
        buf_writer.flush().await?;
    }
}