1use serde::{Serialize, de::DeserializeOwned};2use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};34use crate::{ProtocolError, Result};56pub(crate) async fn write_json<T, W>(writer: &mut W, value: &T, maximum: usize) -> Result<()>7where8 T: Serialize,9 W: AsyncWrite + Unpin,10{11 let payload = serde_json::to_vec(value)?;12 validate_message_length(payload.len(), maximum)?;1314 writer15 .write_all(&(payload.len() as u32).to_be_bytes())16 .await?;17 writer.write_all(&payload).await?;18 writer.flush().await?;19 Ok(())20}2122pub(crate) async fn read_json<T, R>(reader: &mut R, maximum: usize) -> Result<Option<T>>23where24 T: DeserializeOwned,25 R: AsyncRead + Unpin,26{27 let Some(length_bytes) = read_prefix_or_eof(reader).await? else {28 return Ok(None);29 };30 let length = u32::from_be_bytes(length_bytes) as usize;31 validate_message_length(length, maximum)?;3233 let mut payload = vec![0; length];34 reader.read_exact(&mut payload).await?;35 Ok(Some(serde_json::from_slice(&payload)?))36}3738pub(crate) async fn read_prefix_or_eof<R>(reader: &mut R) -> Result<Option<[u8; 4]>>39where40 R: AsyncRead + Unpin,41{42 let mut prefix = [0; 4];43 let read = reader.read(&mut prefix[..1]).await?;44 if read == 0 {45 return Ok(None);46 }47 reader.read_exact(&mut prefix[1..]).await?;48 Ok(Some(prefix))49}5051fn validate_message_length(actual: usize, maximum: usize) -> Result<()> {52 if actual == 0 || actual > maximum || actual > u32::MAX as usize {53 return Err(ProtocolError::InvalidMessageLength { actual, maximum });54 }55 Ok(())56}