remowt-link-shared 0.1.4

use std::io;

use bifrostlink::Port;
use bytes::{Bytes, BytesMut};
use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};

/// Wire a length-prefixed duplex byte stream (e.g. a child process's
/// stdout/stdin) into a bifrost [`Port`]. Each frame is a big-endian `u32`
/// length followed by that many payload bytes.
pub fn child_port<R, W>(mut reader: R, mut writer: W) -> Port
where
	R: AsyncRead + Unpin + Send + 'static,
	W: AsyncWrite + Unpin + Send + 'static,
{
	Port::new(|mut rx, tx| async move {
		let read_task = async move {
			loop {
				let len = match reader.read_u32().await {
					Ok(len) => len,
					Err(e) => {
						tracing::error!("child read failed: {e}");
						break;
					}
				};
				let mut buf = BytesMut::zeroed(len as usize);
				if let Err(e) = reader.read_exact(&mut buf).await {
					tracing::error!("child read failed: {e}");
					break;
				}
				if tx.send(buf.freeze()).is_err() {
					break;
				}
			}
		};
		let write_task = async move {
			while let Some(msg) = rx.recv().await {
				if let Err(e) = write_frame(&mut writer, msg).await {
					tracing::error!("child write failed: {e}");
					break;
				}
			}
		};
		tokio::join!(read_task, write_task);
	})
}

async fn write_frame<W: AsyncWrite + Unpin>(writer: &mut W, msg: Bytes) -> io::Result<()> {
	let len = u32::try_from(msg.len())
		.map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "message larger than 4GB"))?;
	writer.write_u32(len).await?;
	writer.write_all(&msg).await?;
	writer.flush().await?;
	Ok(())
}