AsyncRead and AsyncWrite

Every byte stream in Tokio (a TCP connection, a file, a pipe, a child process's output) is used through two traits: AsyncRead for reading bytes and AsyncWrite for writing bytes. This lesson covers both traits, the helper methods built on them, what a partial read is, how end of file (EOF) is reported, and how to test I/O code without a network by using in-memory readers and tokio::io::duplex.

Item	Purpose
`AsyncRead`, `AsyncWrite`	the traits every async byte stream implements
`AsyncReadExt`	`read`, `read_exact`, `read_to_end`, `read_to_string`, `read_u32` and other integer readers, `take`, `chain`
`AsyncWriteExt`	`write`, `write_all`, `write_u32` and other integer writers, `flush`, `shutdown`
`BufReader`, `AsyncBufReadExt`	buffered reading, `read_line`, `lines`
`BufWriter`	buffered writing
`io::copy`, `io::split`	move all bytes from a reader to a writer, split a stream into two halves
`io::duplex`	an in-memory pair of connected streams
`&[u8]`, `Cursor`, `Vec<u8>`	in-memory readers and writers
`io::empty`, `io::sink`, `io::repeat`	readers and writers with fixed behavior

All examples need the io-util feature (included in full). Each example imports the extension traits it uses, because the helper methods exist only when the trait is in scope.

Example 1: Read from a byte slice

AsyncRead is implemented for &[u8], so a byte string can stand in for a network stream. read copies bytes into a buffer you provide and returns how many bytes it copied.

Rust

use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut source: &[u8] = b"hello world";
    let mut buf = [0u8; 32];

    let n = source.read(&mut buf).await?;
    println!("read {n} bytes: {:?}", std::str::from_utf8(&buf[..n]).unwrap());
    Ok(())
}

Output

read 11 bytes: "hello world"

read returns a usize inside an io::Result. The buffer is 32 bytes, the source holds 11, so one call read everything the source had.

Example 2: The two traits

AsyncRead and AsyncWrite each have poll methods, in the style of the Future trait from Futures and async/await. This block shows the core method of each; it is a reading aid, not a program to run.

Rust

pub trait AsyncRead {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>>;
}

pub trait AsyncWrite {
    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8])
        -> Poll<io::Result<usize>>;
    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
}

ReadBuf is Tokio's wrapper around a byte buffer that tracks which part has been filled. The result of poll_read follows three cases from the trait documentation:

Result	Meaning
`Ready(Ok(()))`	bytes were placed in the buffer. No new bytes means EOF, or a buffer with no free space.
`Pending`	nothing to read now. The waker is registered and the task is woken when the stream becomes readable.
`Ready(Err(e))`	an I/O error

Application code rarely calls poll_* directly. It calls the methods of AsyncReadExt and AsyncWriteExt, which return futures. Writing your own implementation is covered in Writing Futures and Streams by Hand.

AsyncWrite::poll_write returns how many bytes it accepted, which can be fewer than offered. poll_flush pushes buffered data toward the destination. poll_shutdown signals that no more data will be written.

Example 3: Partial reads

A single read call returns whatever is available, up to the buffer size. It does not fill the buffer. The call below uses a 4-byte buffer on 11 bytes of data, so it needs three calls.

Rust

use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut source: &[u8] = b"hello world";
    let mut buf = [0u8; 4];

    for call in 1..=3 {
        let n = source.read(&mut buf).await?;
        println!("call {call}: {n} bytes {:?}", std::str::from_utf8(&buf[..n]).unwrap());
    }
    Ok(())
}

Output

call 1: 4 bytes "hell"
call 2: 4 bytes "o wo"
call 3: 3 bytes "rld"

Here the limit is the buffer. On a network connection the limit is often the data that has arrived so far: a sender that writes 1,000 bytes can be read as 400 and then 600 bytes, or in any other split. Code that assumes one read returns one whole message is wrong even if it passes on a fast local machine.

Example 4: EOF is a read of zero bytes

When the source has no more data and will never have more, read returns Ok(0). This is EOF (end of file, or end of stream for a socket). A loop that reads until it sees 0 handles streams of any size.

Rust

use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut source: &[u8] = b"abcdefghij";
    let mut buf = [0u8; 4];
    let mut total = 0;

    loop {
        let n = source.read(&mut buf).await?;
        if n == 0 {
            println!("EOF");
            break;
        }
        total += n;
        println!("chunk {:?}", std::str::from_utf8(&buf[..n]).unwrap());
    }
    println!("total {total}");
    Ok(())
}

Output

chunk "abcd"
chunk "efgh"
chunk "ij"
EOF
total 10

The chunks are 4, 4, and 2 bytes. The final chunk is shorter than the buffer, which is normal and does not signal EOF. Only the next call, returning 0, does.

Example 5: A zero result also comes from an empty buffer

Ok(0) has two causes: EOF, and a buffer with no room. A loop that passes an empty buffer would mistake it for EOF.

Rust

use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut source: &[u8] = b"data";

    let n = source.read(&mut []).await?;
    println!("empty buffer: {n}");

    let mut buf = [0u8; 8];
    let n = source.read(&mut buf).await?;
    println!("real buffer: {n}");
    Ok(())
}

Output

empty buffer: 0
real buffer: 4

The first call returned 0 even though 4 bytes were waiting, and it consumed nothing. Keep read buffers non-empty.

Example 6: read_exact and unexpected EOF

read_exact fills the whole buffer, calling read as many times as needed. If EOF arrives first, it returns an error of kind UnexpectedEof. This is the tool for fixed-size fields such as a 4-byte header.

Rust

use std::io::ErrorKind;
use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() {
    let mut source: &[u8] = b"0123456789";
    let mut header = [0u8; 4];

    source.read_exact(&mut header).await.unwrap();
    println!("header {:?}", std::str::from_utf8(&header).unwrap());

    let mut big = [0u8; 8];
    let err = source.read_exact(&mut big).await.unwrap_err();
    println!("kind {:?}", err.kind());
    println!("is UnexpectedEof: {}", err.kind() == ErrorKind::UnexpectedEof);
    println!("message: {err}");
}

Output

header "0123"
kind UnexpectedEof
is UnexpectedEof: true
message: early eof

Only 6 bytes remained for an 8-byte buffer. The error does not report how many bytes were read, and the contents of the buffer after an error are unspecified. Treat the bytes as lost and stop.

Example 7: read_to_end and read_to_string

read_to_end appends everything up to EOF to a Vec<u8> and returns the count. read_to_string does the same into a String and returns an InvalidData error if the bytes are not valid UTF-8.

Rust

use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut source: &[u8] = b"first part, ";
    let mut bytes = Vec::new();
    let n = source.read_to_end(&mut bytes).await?;
    println!("read_to_end: {n} bytes, vec len {}", bytes.len());

    let mut text_source: &[u8] = b"line one\nline two\n";
    let mut text = String::new();
    let n = text_source.read_to_string(&mut text).await?;
    println!("read_to_string: {n} bytes");
    print!("{text}");

    let mut bad: &[u8] = &[0xff, 0xfe, 0xfd];
    let mut sink = String::new();
    let err = bad.read_to_string(&mut sink).await.unwrap_err();
    println!("invalid utf-8: {:?}", err.kind());
    Ok(())
}

Output

read_to_end: 12 bytes, vec len 12
read_to_string: 18 bytes
line one
line two
invalid utf-8: InvalidData

Both methods wait for EOF, so on a socket they return only after the peer closes its sending side. They also grow the buffer without a limit. The take adapter in Example 16 bounds the size.

Example 8: Reading and writing integers

AsyncReadExt has read_u8, read_u16, read_u32, read_u64, and signed and float versions, each reading a fixed number of bytes in big-endian order (most significant byte first, the usual order for network protocols). Each has a _le variant for little-endian. AsyncWriteExt has matching write_u32 and similar methods. These call read_exact internally, so a short stream gives UnexpectedEof.

Rust

use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut out: Vec<u8> = Vec::new();
    out.write_u8(7).await?;
    out.write_u16(0x0102).await?;
    out.write_u32(0xDEADBEEF).await?;
    out.write_u16_le(0x0102).await?;
    println!("{:02x?}", out);

    let mut input: &[u8] = &out;
    let a = input.read_u8().await?;
    let b = input.read_u16().await?;
    let c = input.read_u32().await?;
    let d = input.read_u16_le().await?;
    println!("{a} {b:#06x} {c:#x} {d:#06x}");

    let err = input.read_u32().await.unwrap_err();
    println!("after the end: {:?}", err.kind());
    Ok(())
}

Output

[07, 01, 02, de, ad, be, ef, 02, 01]
7 0x0102 0xdeadbeef 0x0102
after the end: UnexpectedEof

Vec<u8> implements AsyncWrite, so it works as a writer that collects bytes. The 16-bit value 0x0102 is written as 01 02 in big-endian and 02 01 in little-endian.

Example 9: write, write_all, and partial writes

write returns how many bytes the destination accepted, which can be fewer than offered. write_all loops until every byte is accepted. tokio::io::duplex(n) creates two connected in-memory streams; bytes written to one end are read from the other, and each direction holds at most n bytes. With n = 4, a write of 10 bytes is accepted only in part.

Rust

use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut a, mut b) = duplex(4);

    let n = a.write(b"0123456789").await?;
    println!("write accepted {n} of 10 bytes");

    let mut buf = [0u8; 16];
    let n = b.read(&mut buf).await?;
    println!("peer read {:?}", std::str::from_utf8(&buf[..n]).unwrap());

    let n = a.write(b"456789").await?;
    println!("second write accepted {n} of 6 bytes");
    Ok(())
}

Output

write accepted 4 of 10 bytes
peer read "0123"
second write accepted 4 of 6 bytes

The first write accepted 4 bytes, the capacity of the pipe. After the peer drained the pipe, space was free again. Code that calls write and ignores the returned count silently drops data. write_all exists to prevent that.

Example 10: Backpressure with write_all on a full pipe

When the pipe is full, write_all waits until the reader frees space. This is backpressure: a slow reader slows the writer instead of letting memory grow. A 4-byte pipe carrying 26 bytes needs the reader to run concurrently, so the reader runs in a spawned task (see Spawning Tasks).

Rust

use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut tx, mut rx) = duplex(4);

    let reader = tokio::spawn(async move {
        let mut received = Vec::new();
        rx.read_to_end(&mut received).await.unwrap();
        received
    });

    tx.write_all(b"abcdefghijklmnopqrstuvwxyz").await?;
    tx.shutdown().await?;

    let received = reader.await.unwrap();
    println!("{}", String::from_utf8(received).unwrap());
    Ok(())
}

Output

abcdefghijklmnopqrstuvwxyz

write_all returned only after the reader took all 26 bytes through a 4-byte window. Without the spawned reader, the writer would wait forever on the full pipe. shutdown (Example 12) is what lets read_to_end see EOF.

Example 11: Dropping one end of a pipe

If one end of a duplex pair is dropped, the other end can still read what was already written, then sees EOF. Writes to the surviving end fail with BrokenPipe. Sockets behave in a similar way (see TCP Clients and Servers).

Rust

use std::io::ErrorKind;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut a, mut b) = duplex(64);

    a.write_all(b"last words").await?;
    drop(a);

    let mut buf = [0u8; 32];
    let n = b.read(&mut buf).await?;
    println!("data before close: {:?}", std::str::from_utf8(&buf[..n]).unwrap());
    let n = b.read(&mut buf).await?;
    println!("then EOF: {n}");

    let err = b.write_all(b"anyone there?").await.unwrap_err();
    println!("write to closed peer: {:?}", err.kind());
    println!("is BrokenPipe: {}", err.kind() == ErrorKind::BrokenPipe);
    Ok(())
}

Output

data before close: "last words"
then EOF: 0
write to closed peer: BrokenPipe
is BrokenPipe: true

Buffered bytes are not lost when the writer drops. The reader drains them and then gets 0.

Example 12: flush and shutdown

flush asks the writer to push any buffered bytes to the destination. shutdown flushes and then closes the write direction, which the peer sees as EOF. After shutdown, the other direction still works: this is the half-close used by protocols where one side finishes sending and then waits for a reply.

Rust

use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut client, mut server) = duplex(64);

    client.write_all(b"request").await?;
    client.flush().await?;
    client.shutdown().await?;

    let mut request = Vec::new();
    server.read_to_end(&mut request).await?;
    println!("server got {:?}", String::from_utf8(request).unwrap());

    server.write_all(b"response").await?;
    server.shutdown().await?;

    let mut response = Vec::new();
    client.read_to_end(&mut response).await?;
    println!("client got {:?}", String::from_utf8(response).unwrap());
    Ok(())
}

Output

server got "request"
client got "response"

The server's read_to_end could finish because the client called shutdown. The client could still read the response afterwards because shutdown closes only the client's sending direction.

Example 13: BufReader and read_line

Reading a byte at a time from a slow source means many small reads. BufReader wraps a reader with an internal buffer (8 KiB by default), fills it with large reads, and serves small requests from memory. It also adds AsyncBufReadExt, whose read_line reads up to and including the next \n.

Rust

use tokio::io::{AsyncBufReadExt, BufReader};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let data: &[u8] = b"GET /index\nHost: local\n\nbody without newline";
    let mut reader = BufReader::new(data);

    let mut line = String::new();
    loop {
        line.clear();
        let n = reader.read_line(&mut line).await?;
        if n == 0 {
            println!("EOF");
            break;
        }
        println!("{n} bytes: {:?}", line);
    }
    Ok(())
}

Output

11 bytes: "GET /index\n"
12 bytes: "Host: local\n"
1 bytes: "\n"
20 bytes: "body without newline"
EOF

read_line appends to the string, so the loop clears it each time. The line keeps its \n. The empty line returns 1 byte, not 0. The last line has no \n and is still returned, because EOF ends it. Only a return of 0 means EOF.

Example 14: lines

lines() turns a BufRead source into an async iterator: next_line().await returns Ok(Some(String)) per line, with the line ending removed, and Ok(None) at EOF.

Rust

use tokio::io::{AsyncBufReadExt, BufReader};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let data: &[u8] = b"alpha\nbeta\r\n\ngamma";
    let mut lines = BufReader::new(data).lines();

    while let Some(line) = lines.next_line().await? {
        println!("{:?}", line);
    }
    println!("done");
    Ok(())
}

Output

"alpha"
"beta"
""
"gamma"
done

Both \n and \r\n are removed. The empty line comes back as an empty string, and the unterminated final line is still returned. lines has no maximum line length, so a peer that never sends \n makes the string grow until memory runs out. Bytes and Framing covers limits.

Example 15: BufWriter

BufWriter collects small writes in an 8 KiB buffer and passes them on in larger writes. The data reaches the inner writer only when the buffer fills, when flush is called, or when shutdown is called. Dropping a BufWriter does not flush it, because a drop cannot wait.

Rust

use tokio::io::{AsyncWriteExt, BufWriter};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut writer = BufWriter::new(Vec::<u8>::new());

    writer.write_all(b"one ").await?;
    writer.write_all(b"two ").await?;
    println!("inner after two writes: {}", writer.get_ref().len());

    writer.flush().await?;
    println!("inner after flush: {}", writer.get_ref().len());

    writer.write_all(b"three").await?;
    let inner = writer.into_inner();
    println!("inner after into_inner without flush: {}", inner.len());
    Ok(())
}

Output

inner after two writes: 0
inner after flush: 8
inner after into_inner without flush: 8

The inner Vec stayed empty until flush. into_inner returns the inner writer without flushing, so the 5 bytes of three were discarded. Always call flush or shutdown before finishing with a BufWriter. Many small writes to a socket are costly because each is a system call; BufWriter reduces the count.

Example 16: take and chain

take(n) limits a reader to at most n bytes and then reports EOF. chain(other) reads the first reader to EOF and then the second. Together they bound input and glue pieces.

Rust

use tokio::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let source: &[u8] = b"0123456789";
    let mut limited = source.take(4);
    let mut out = String::new();
    limited.read_to_string(&mut out).await?;
    println!("take(4): {out}");
    println!("limit left: {}", limited.limit());

    let head: &[u8] = b"HEAD|";
    let tail: &[u8] = b"tail";
    let mut both = head.chain(tail);
    let mut joined = String::new();
    both.read_to_string(&mut joined).await?;
    println!("chain: {joined}");
    Ok(())
}

Output

take(4): 0123
limit left: 0
chain: HEAD|tail

take is the standard guard against a peer that sends too much: stream.take(MAX).read_to_end(&mut buf) never stores more than MAX bytes.

Example 17: io::copy

tokio::io::copy(&mut reader, &mut writer) reads from the reader and writes to the writer until the reader reaches EOF, and returns the number of bytes copied. At EOF it flushes the writer, but it does not shut the writer down.

Rust

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut source: &[u8] = b"copy these bytes";
    let mut dest: Vec<u8> = Vec::new();

    let copied = io::copy(&mut source, &mut dest).await?;
    println!("copied {copied} bytes: {:?}", String::from_utf8(dest.clone()).unwrap());

    let (mut a, mut b) = io::duplex(8);
    let writer = tokio::spawn(async move {
        let mut data: &[u8] = b"through a small pipe";
        let n = io::copy(&mut data, &mut a).await.unwrap();
        a.shutdown().await.unwrap();
        n
    });
    let mut received = String::new();
    b.read_to_string(&mut received).await?;
    println!("received {received:?}, writer copied {}", writer.await.unwrap());
    Ok(())
}

Output

copied 16 bytes: "copy these bytes"
received "through a small pipe", writer copied 20

The second half copies 20 bytes through an 8-byte pipe, with the reader running at the same time. The writer task calls shutdown after copy because copy only flushes, and without the shutdown read_to_string would wait forever.

Example 18: io::split

A stream that is both AsyncRead and AsyncWrite can be split into a ReadHalf and a WriteHalf with tokio::io::split. Each half can move into its own task, so one task reads while another writes. The halves synchronize internally, which costs a lock; TCP streams have a cheaper split, covered in TCP Clients and Servers.

Rust

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> io::Result<()> {
    let (client, server) = io::duplex(64);
    let (mut server_read, mut server_write) = io::split(server);

    // The server echoes every byte back, converted to upper case.
    let echo = tokio::spawn(async move {
        let mut buf = [0u8; 16];
        loop {
            let n = server_read.read(&mut buf).await.unwrap();
            if n == 0 {
                break;
            }
            buf[..n].make_ascii_uppercase();
            server_write.write_all(&buf[..n]).await.unwrap();
        }
        server_write.shutdown().await.unwrap();
    });

    let (mut client_read, mut client_write) = io::split(client);
    client_write.write_all(b"hello halves").await?;
    client_write.shutdown().await?;

    let mut reply = String::new();
    client_read.read_to_string(&mut reply).await?;
    println!("{reply}");

    echo.await.unwrap();
    Ok(())
}

Output

HELLO HALVES

This is the shape of an echo server: read until 0, write back, then close. The client shut down its write half so the server saw EOF, and the server shut down its write half so the client saw EOF.

Example 19: empty, sink, and repeat

Three fixed streams help in tests and tools:

Function	Behavior
`io::empty()`	reads return EOF at once, writes discard the data and report success
`io::sink()`	writes discard the data and report success
`io::repeat(byte)`	reads an endless supply of one byte

Rust

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut empty = io::empty();
    let mut buf = Vec::new();
    println!("empty: {} bytes", empty.read_to_end(&mut buf).await?);

    let mut sink = io::sink();
    sink.write_all(b"vanishes").await?;
    println!("sink accepted the write");

    let mut zeros = io::repeat(0).take(5);
    let mut five = Vec::new();
    zeros.read_to_end(&mut five).await?;
    println!("repeat.take(5): {:?}", five);

    let copied = io::copy(&mut io::repeat(b'x').take(1000), &mut io::sink()).await?;
    println!("copied {copied} bytes into a sink");
    Ok(())
}

Output

empty: 0 bytes
sink accepted the write
repeat.take(5): [0, 0, 0, 0, 0]
copied 1000 bytes into a sink

repeat never reaches EOF, so it is always paired with take. Copying into a sink measures how many bytes a reader produces.

Example 20: Cursor as a reader and writer

std::io::Cursor wraps a buffer with a position. Tokio implements AsyncRead for Cursor<T: AsRef<[u8]>> and AsyncWrite for Cursor<Vec<u8>> and Cursor<&mut [u8]>. Unlike &[u8], a cursor keeps the original data and lets you read its position.

Rust

use std::io::Cursor;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut reader = Cursor::new(b"abcdef".to_vec());
    let mut two = [0u8; 2];
    reader.read_exact(&mut two).await?;
    println!("after 2 bytes: position {}", reader.position());
    reader.set_position(4);
    let mut rest = String::new();
    reader.read_to_string(&mut rest).await?;
    println!("from position 4: {rest}");
    println!("data still there: {:?}", reader.get_ref());

    let mut writer = Cursor::new(Vec::new());
    writer.write_all(b"written").await?;
    println!("wrote {} bytes: {:?}", writer.position(), writer.get_ref());
    Ok(())
}

Output

after 2 bytes: position 2
from position 4: ef
data still there: [97, 98, 99, 100, 101, 102]
wrote 7 bytes: [119, 114, 105, 116, 116, 101, 110]

&[u8] advances by shrinking the slice, so the consumed bytes are gone from the variable. Cursor moves an index instead.

Example 21: Functions generic over the traits

A function that accepts impl AsyncRead + Unpin works with a socket, a file, a duplex end, or a byte slice. Unpin is the marker that lets the helper methods take &mut self without pinning; sockets, files, &[u8], Cursor, and duplex ends are all Unpin. This is what makes I/O code testable: the test passes bytes, the service passes a socket.

Rust

use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, AsyncBufReadExt};

// Reads lines until EOF. Returns how many lines it saw and the longest one.
async fn summarize<R: AsyncRead + Unpin>(input: R) -> std::io::Result<(usize, usize)> {
    let mut lines = BufReader::new(input).lines();
    let (mut count, mut longest) = (0, 0);
    while let Some(line) = lines.next_line().await? {
        count += 1;
        longest = longest.max(line.len());
    }
    Ok((count, longest))
}

// Writes a length-prefixed message: a u32 length, then the payload.
async fn write_message<W: AsyncWrite + Unpin>(out: &mut W, payload: &[u8]) -> std::io::Result<()> {
    out.write_u32(payload.len() as u32).await?;
    out.write_all(payload).await?;
    out.flush().await
}

// Reads one length-prefixed message.
async fn read_message<R: AsyncRead + Unpin>(input: &mut R) -> std::io::Result<Vec<u8>> {
    let len = input.read_u32().await? as usize;
    let mut payload = vec![0u8; len];
    input.read_exact(&mut payload).await?;
    Ok(payload)
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let stats = summarize(&b"one\ntwo words\nthree"[..]).await?;
    println!("lines={} longest={}", stats.0, stats.1);

    let (mut a, mut b) = tokio::io::duplex(1024);
    write_message(&mut a, b"first").await?;
    write_message(&mut a, b"second message").await?;
    drop(a);

    println!("{:?}", String::from_utf8(read_message(&mut b).await?).unwrap());
    println!("{:?}", String::from_utf8(read_message(&mut b).await?).unwrap());
    let end = read_message(&mut b).await.unwrap_err();
    println!("third read: {:?}", end.kind());
    Ok(())
}

Output

lines=3 longest=9
"first"
"second message"
third read: UnexpectedEof

read_message uses read_u32 and read_exact because both wait for the full field, however many partial reads that takes. At a clean end of stream, read_u32 fails with UnexpectedEof. A real protocol reader treats that error at a message boundary as "the peer finished" and the same error in the middle of a message as a truncated message.

Example 22: A reader that returns data in small pieces

A spawned task writes three pieces into a duplex pipe, one after another. The reader sees the data split by the writer's writes and by its own buffer size. A parser that works only when the whole message arrives in one read would fail here.

Rust

use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::task::yield_now;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut tx, mut rx) = duplex(64);

    let writer = tokio::spawn(async move {
        for piece in ["HEL", "LO, ", "WORLD"] {
            tx.write_all(piece.as_bytes()).await.unwrap();
            yield_now().await;
        }
    });

    let mut message = Vec::new();
    let mut buf = [0u8; 64];
    loop {
        let n = rx.read(&mut buf).await?;
        if n == 0 {
            break;
        }
        message.extend_from_slice(&buf[..n]);
    }
    writer.await.unwrap();

    println!("message {:?}", String::from_utf8(message).unwrap());
    Ok(())
}

Output

message "HELLO, WORLD"

The number of reads depends on scheduling, so the example prints only the assembled message. This is how to write tests for I/O code. Assert on the reassembled bytes, never on how many read calls it took.

Example 23: A line-based protocol handler, tested in memory

The handler reads commands line by line and writes a reply per command. Because it takes AsyncRead + AsyncWrite, the same function serves a socket in TCP Clients and Servers and a duplex end here.

Rust

use tokio::io::{
    duplex, split, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
};

async fn serve<S: AsyncRead + AsyncWrite + Unpin>(stream: S) -> std::io::Result<()> {
    let (read_half, mut write_half) = split(stream);
    let mut lines = BufReader::new(read_half).lines();
    let mut total: i64 = 0;

    while let Some(line) = lines.next_line().await? {
        let reply = match line.split_once(' ') {
            Some(("ADD", n)) => match n.parse::<i64>() {
                Ok(n) => {
                    total += n;
                    format!("OK {total}")
                }
                Err(_) => "ERR not a number".to_string(),
            },
            None if line == "TOTAL" => format!("TOTAL {total}"),
            _ => "ERR unknown command".to_string(),
        };
        write_half.write_all(reply.as_bytes()).await?;
        write_half.write_all(b"\n").await?;
    }
    write_half.shutdown().await
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut client, server) = duplex(256);
    let handler = tokio::spawn(serve(server));

    client.write_all(b"ADD 5\nADD 7\nADD x\nHELLO\nTOTAL\n").await?;
    client.shutdown().await?;

    let mut replies = String::new();
    client.read_to_string(&mut replies).await?;
    print!("{replies}");

    handler.await.unwrap()?;
    Ok(())
}

Output

OK 5
OK 12
ERR not a number
ERR unknown command
TOTAL 12

The client wrote all five commands at once and closed its sending side. The handler processed them in order and, after EOF, closed its own side, which ended read_to_string. Replies are in command order because a single task handles the stream.

Interview points

read returns Ok(n) with 0 < n <= buf.len(). It does not fill the buffer, and it does not preserve message boundaries. Ok(0) means EOF unless the buffer was empty.
read_exact returns UnexpectedEof if the stream ends first. read_to_end and read_to_string wait for EOF and have no size limit. take(n) adds one.
write may accept fewer bytes than offered. write_all loops. Neither flushes a buffering writer.
flush pushes buffered bytes onward. shutdown flushes and closes the write direction, which the peer sees as EOF. Dropping a BufWriter does not flush.
BufReader reduces the number of reads on the inner source, and lines and read_line need it (or another AsyncBufRead). read_line keeps the newline and lines removes it. Neither limits the line length.
Integer helpers such as read_u32 and write_u32 are big-endian. _le variants use little-endian.
io::copy runs until the reader reaches EOF, flushes the writer, and does not shut it down. It uses an 8 KiB buffer.
Cancel safety (from the method documentation): read, write, and flush are cancel safe. read_exact, write_all, and the multi-byte integer readers are not, because part of the data may already be consumed or written when the future is dropped. select!: Racing Futures defines cancel safety.
io::split works for any AsyncRead + AsyncWrite and uses an internal lock. duplex(n) buffers n bytes per direction; a full pipe makes the writer wait; a dropped end gives the peer EOF after the buffered data, and gives writes BrokenPipe.
Generic functions over AsyncRead + Unpin and AsyncWrite + Unpin can be tested with &[u8], Cursor, and duplex without sockets.

Summary

AsyncRead and AsyncWrite are the poll-based traits for byte streams. AsyncReadExt and AsyncWriteExt supply the methods used in application code.
A read can return fewer bytes than requested at any time. EOF is a read that returns 0 into a non-empty buffer.
Use read_exact for fixed-size fields and write_all for complete writes.
Wrap slow or small-grained sources in BufReader, and small writes in BufWriter with an explicit flush.
tokio::io::duplex, &[u8], and Cursor make I/O code deterministic to test; assert on the reassembled bytes, not on the number of reads.