LTW: tokio::io::DuplexStream

tokio::io::DuplexStream is a very useful type if you need an in-memory AsyncRead and AsyncWrite type. As an example, suppose you have some code that implements a custom RPC protocol. The client is expected to write a single byte opcode followed by two single byte arguments. The server will respond with a single byte error indicator and a single byte return value.

async fn rpc_client(mut io: impl AsyncRead + AsyncWrite + Unpin) -> Result<i8> {
    io.write(&[b'+', 1, 2]).await?;

    let mut buffer = [0, 0];
    io.read(buffer.as_mut_slice()).await?;

    if buffer[0] == 0 {
        Ok(buffer[1] as i8)
    } else {
        Err(Error)
    }
}

async fn rpc_server(mut io: impl AsyncRead + AsyncWrite + Unpin) -> Result<()> {
    let mut buffer = [0, 0, 0];
    io.read(buffer.as_mut_slice()).await?;

    match buffer[0] {
        b'+' => {
            io.write(&[0, buffer[1] + buffer[2]]).await?;
        }
        _ => todo!(),
    }

    Ok(())
}

This code is written generically using AsyncRead and AsyncWrite, so we can test this using tokio::io::DuplexStream.

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn it_works() {
        let (client, server) = tokio::io::duplex(8);

        tokio::spawn(rpc_server(server));

        assert_eq!(rpc_client(client).await.unwrap(), 3);
    }
}

The argument to tokio::io::duplex() is the internal buffer size. It is the maximum amount that the writer can write before it will block.