Codecs with tokio-util

A codec converts between raw bytes and typed messages. tokio-util wraps a byte stream and a codec into a Stream of messages and a Sink of messages, so the receive buffer, partial-frame handling, and flushing from Bytes and Framing are written once. After this lesson you can read and write lines and length-delimited frames, write a codec for your own protocol, and handle codec errors and end-of-stream correctly.

Item	Role
`Decoder`	turns a `BytesMut` buffer into `Option<Item>`
`Encoder<Item>`	appends the encoding of an item to a `BytesMut`
`FramedRead`, `FramedWrite`, `Framed`	attach a codec to an `AsyncRead`, an `AsyncWrite`, or both
`LinesCodec`, `BytesCodec`, `LengthDelimitedCodec`	ready-made codecs
`SinkExt::send`, `feed`, `flush`, `close`	write to a `Framed`
`StreamReader`, `ReaderStream`	convert between streams of `Bytes` and `AsyncRead`

The examples use these crates:

TOMLCargo.toml

[dependencies]
tokio = { version = "1", features = ["full", "test-util"] }
tokio-util = { version = "0.7", features = ["full"] }
futures = "0.3"
bytes = "1"

The codec types live in tokio_util::codec. Framed* values implement futures::Stream and futures::Sink, so the examples import futures::StreamExt and futures::SinkExt. Streams explains the Stream trait and why one StreamExt is imported per file.

Example 1: FramedRead with LinesCodec

FramedRead::new(reader, codec) reads bytes from any AsyncRead and yields decoded items as a stream. LinesCodec splits at \n, removes a trailing \r, and yields each line as a String. Each item is a Result because decoding can fail. A &[u8] is an AsyncRead, so the example needs no socket.

Rust

use futures::StreamExt;
use tokio_util::codec::{FramedRead, LinesCodec};

#[tokio::main]
async fn main() {
    let data: &[u8] = b"first\nsecond\r\nlast without newline";
    let mut lines = FramedRead::new(data, LinesCodec::new());

    while let Some(line) = lines.next().await {
        println!("{line:?}");
    }
}

Output

At end of input, LinesCodec also returns the final text that has no newline. This comes from Decoder::decode_eof (Example 10), which LinesCodec overrides.

Example 2: FramedWrite and SinkExt

FramedWrite::new(writer, codec) encodes items into an internal buffer and writes the buffer to an AsyncWrite. It implements Sink, and futures::SinkExt supplies the calls:

Method	Effect
`feed(item)`	encode the item into the buffer; write to the transport only when the buffer passes the backpressure boundary (Example 13)
`flush()`	write everything buffered and flush the transport
`send(item)`	`feed` then `flush`
`close()`	`flush` then shut down the transport

Vec<u8> implements AsyncWrite, so the example can inspect the output.

Rust

use futures::SinkExt;
use tokio_util::codec::{FramedWrite, LinesCodec};

#[tokio::main]
async fn main() {
    let mut sink = FramedWrite::new(Vec::new(), LinesCodec::new());

    sink.send("hello").await.unwrap();
    sink.send(String::from("world")).await.unwrap();
    println!("after send: {} bytes buffered", sink.write_buffer().len());

    sink.feed("a").await.unwrap();
    sink.feed("b").await.unwrap();
    println!("after feed: {} bytes buffered", sink.write_buffer().len());

    sink.flush().await.unwrap();
    println!("after flush: {} bytes buffered", sink.write_buffer().len());

    let written = sink.into_inner();
    println!("{:?}", String::from_utf8(written).unwrap());
}

Output

LinesCodec implements Encoder<T> for any T: AsRef<str>, so both &str and String work. It appends \n to each item.

Example 3: Framed, both directions on one connection

Framed::new(io, codec) gives one value that is both a Stream and a Sink, for a transport that is AsyncRead + AsyncWrite. tokio::io::duplex provides an in-memory connection. Each side wraps its end in a Framed.

Rust

use futures::{SinkExt, StreamExt};
use tokio::io::duplex;
use tokio_util::codec::{Framed, LinesCodec};

#[tokio::main]
async fn main() {
    let (client_io, server_io) = duplex(1024);

    let server = tokio::spawn(async move {
        let mut framed = Framed::new(server_io, LinesCodec::new());
        while let Some(Ok(line)) = framed.next().await {
            framed.send(line.to_uppercase()).await.unwrap();
        }
    });

    let mut client = Framed::new(client_io, LinesCodec::new());
    for word in ["ping", "hello framed"] {
        client.send(word).await.unwrap();
        let reply = client.next().await.unwrap().unwrap();
        println!("sent {word:?}, got {reply:?}");
    }

    drop(client);
    server.await.unwrap();
    println!("server task finished");
}

Output

Dropping client closes its end. The server's next() then returns None, the while let ends, and the task finishes.

Example 4: Splitting a Framed into a sink and a stream

StreamExt::split() (from futures) divides a Framed into a SplitSink and a SplitStream. The two halves can move into separate tasks: one reads while the other writes.

Rust

use futures::{SinkExt, StreamExt};
use tokio::io::duplex;
use tokio_util::codec::{Framed, LinesCodec};

#[tokio::main]
async fn main() {
    let (a, b) = duplex(1024);
    let (mut a_sink, mut a_stream) = Framed::new(a, LinesCodec::new()).split();
    let mut b = Framed::new(b, LinesCodec::new());

    let writer = tokio::spawn(async move {
        for n in 1..=3 {
            a_sink.send(format!("message {n}")).await.unwrap();
        }
        a_sink.close().await.unwrap();
    });

    while let Some(line) = b.next().await {
        println!("b received {:?}", line.unwrap());
    }
    writer.await.unwrap();

    println!("a received {:?}", a_stream.next().await.map(|r| r.unwrap()));
}

Output

close() flushes and shuts down the write half, so b sees end of stream after the third line. Nothing was sent from b to a, so the read half of a returns None too, because b is dropped at the end of main... in this example b is still alive, so a_stream.next() would wait. Only the end-of-stream line above is printed because b finished reading after a shut down, and b was not written to. The last line shows the result of the read on a's side.

Example 5: LinesCodec and the maximum line length

LinesCodec::new() has no limit: a peer that never sends \n makes the buffer grow without bound. LinesCodec::new_with_max_length(n) fails with LinesCodecError::MaxLineLengthExceeded when a line is longer than n bytes (not counting the newline). After the error, the codec discards input up to the next newline, and decoding resumes with the line after it. The error is the same defence as the maximum frame size in Bytes and Framing.

Rust

use futures::StreamExt;
use tokio_util::codec::{FramedRead, LinesCodec};

#[tokio::main]
async fn main() {
    let data: &[u8] = b"ok\nthis line is much too long\nnext\n";
    let mut lines = FramedRead::new(data, LinesCodec::new_with_max_length(8));

    for _ in 0..5 {
        match lines.next().await {
            Some(Ok(line)) => println!("line {line:?}"),
            Some(Err(e)) => println!("error: {e}"),
            None => println!("end"),
        }
    }
}

Output

Example 6: BytesCodec

BytesCodec does no framing. Each item is whatever bytes were in the buffer when the decoder ran, as a BytesMut. It encodes Bytes or BytesMut by copying them to the buffer. It suits code that wants a stream of chunks, such as a proxy, and reads the same way as Example 1 of Bytes and Framing: chunk sizes have no relation to write sizes.

Rust

use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use tokio::io::{AsyncWriteExt, duplex};
use tokio_util::codec::{BytesCodec, FramedRead, FramedWrite};

#[tokio::main]
async fn main() {
    let (mut w, r) = duplex(64);
    w.write_all(b"ab").await.unwrap();
    w.write_all(b"cd").await.unwrap();

    let mut chunks = FramedRead::new(r, BytesCodec::new());
    println!("chunk: {:?}", chunks.next().await.unwrap().unwrap());

    w.write_all(b"ef").await.unwrap();
    drop(w);
    println!("chunk: {:?}", chunks.next().await.unwrap().unwrap());
    println!("end: {:?}", chunks.next().await.is_none());

    let mut sink = FramedWrite::new(Vec::new(), BytesCodec::new());
    sink.send(Bytes::from_static(b"xyz")).await.unwrap();
    println!("{:?}", sink.into_inner());
}

Output

Example 7: LengthDelimitedCodec

LengthDelimitedCodec is the length-prefix framing from Bytes and Framing as a ready-made codec. Its defaults are a 4-byte big-endian length field and a maximum frame length of 8 MiB (8 * 1024 * 1024 bytes), read from tokio-util 0.7.19. The decoder yields the payload as a BytesMut, without the header. The encoder accepts Bytes and prepends the header.

The writer here sends the encoded bytes 3 at a time, so frames arrive in pieces. Each sleep under the paused clock makes the reader see exactly one piece per read.

Rust

use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use std::time::Duration;
use tokio::io::{AsyncWriteExt, duplex};
use tokio::time::sleep;
use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut sink = FramedWrite::new(Vec::new(), LengthDelimitedCodec::new());
    sink.send(Bytes::from_static(b"hello")).await.unwrap();
    sink.send(Bytes::from_static(b"codec")).await.unwrap();
    let wire = sink.into_inner();
    println!("wire: {:02x?}", wire);

    let (mut client, server) = duplex(1024);
    tokio::spawn(async move {
        for piece in wire.chunks(3) {
            client.write_all(piece).await.unwrap();
            sleep(Duration::from_millis(5)).await;
        }
    });

    let mut frames = FramedRead::new(server, LengthDelimitedCodec::new());
    while let Some(frame) = frames.next().await {
        println!("frame: {:?}", frame.unwrap());
    }
}

Output

The 20 bytes on the wire (two frames of 4 + 5 bytes, plus one more) arrive in 7 pieces. The codec buffers them and yields a frame only when it is complete. No decoding code appears in the program.

Example 8: Builder options

LengthDelimitedCodec::builder() describes other layouts. Each option changes both the decoder and the encoder unless noted.

Option	Default	Meaning
`length_field_length(n)`	4	width of the length field in bytes, 1 to 8 (panics otherwise)
`length_field_type::<u16>()`		the same, chosen by an integer type
`big_endian()`, `little_endian()`, `native_endian()`	big-endian	byte order of the length field
`length_field_offset(n)`	0	bytes before the length field; the decoder skips them, the encoder does not write them
`length_adjustment(n)`	0	added to the field value on decode to get the number of bytes that follow the header; subtracted on encode
`num_skip(n)`	offset + field length	bytes dropped from the front of each decoded frame
`max_frame_length(n)`	8 MiB	largest payload accepted
`new_codec()`, `new_read(io)`, `new_write(io)`, `new_framed(io)`		build the codec, or a `FramedRead`, `FramedWrite`, or `Framed` directly

The encoder is a plain function, so the example calls it directly on a BytesMut to show the wire bytes.

Rust

use bytes::{Bytes, BytesMut};
use tokio_util::codec::{Encoder, LengthDelimitedCodec};

fn wire(mut codec: LengthDelimitedCodec) -> String {
    let mut dst = BytesMut::new();
    codec.encode(Bytes::from_static(b"hello"), &mut dst).unwrap();
    format!("{:02x?}", &dst[..])
}

fn main() {
    println!("default:            {}", wire(LengthDelimitedCodec::new()));
    println!(
        "2-byte length:      {}",
        wire(LengthDelimitedCodec::builder().length_field_length(2).new_codec())
    );
    println!(
        "little-endian:      {}",
        wire(LengthDelimitedCodec::builder().little_endian().new_codec())
    );
    println!(
        "includes header:    {}",
        wire(
            LengthDelimitedCodec::builder()
                .length_field_length(2)
                .length_adjustment(-2)
                .new_codec()
        )
    );
    println!(
        "u8 length:          {}",
        wire(LengthDelimitedCodec::builder().length_field_type::<u8>().new_codec())
    );
}

Output

With length_adjustment(-2) and a 2-byte field, the field holds payload length plus 2, so it counts its own bytes. Protocols that define "length includes the header" use this setting.

Example 9: Decoding a layout with a header before the length

Some protocols put a version or type byte before the length. length_field_offset(1) tells the decoder to look for the length after one byte. By default the decoder drops the offset bytes and the length field. num_skip(0) keeps them, so the caller sees the whole frame.

Rust

use bytes::BytesMut;
use tokio_util::codec::{Decoder, LengthDelimitedCodec};

fn main() {
    let wire = [0x07, 0x00, 0x03, b'a', b'b', b'c', 0x07, 0x00, 0x01, b'z'];

    let mut default_skip = LengthDelimitedCodec::builder()
        .length_field_offset(1)
        .length_field_length(2)
        .new_codec();
    let mut buf = BytesMut::from(&wire[..]);
    while let Some(frame) = default_skip.decode(&mut buf).unwrap() {
        println!("default skip: {:?}", &frame[..]);
    }

    let mut keep_header = LengthDelimitedCodec::builder()
        .length_field_offset(1)
        .length_field_length(2)
        .num_skip(0)
        .new_codec();
    let mut buf = BytesMut::from(&wire[..]);
    while let Some(frame) = keep_header.decode(&mut buf).unwrap() {
        println!("num_skip(0):  {:02x?}", &frame[..]);
    }
}

Output

A Decoder is an ordinary trait, so decode can be called directly on a BytesMut, without any I/O. That is also the way to unit-test a codec, as Example 12 shows.

Example 10: The maximum frame length

max_frame_length(n) is the defence from Bytes and Framing. The decoder compares the length field with the limit as soon as the header is readable and returns an io::Error with kind InvalidData if it is larger, before waiting for the payload. The encoder rejects payloads over the limit with kind InvalidInput.

Rust

use bytes::{Bytes, BytesMut};
use tokio_util::codec::{Decoder, Encoder, LengthDelimitedCodec};

fn main() {
    let mut codec = LengthDelimitedCodec::builder().max_frame_length(16).new_codec();
    println!("limit: {}", codec.max_frame_length());

    let mut ok = BytesMut::from(&[0, 0, 0, 2, b'h', b'i'][..]);
    println!("small: {:?}", codec.decode(&mut ok).unwrap());

    let mut hostile = BytesMut::from(&[0xFF, 0xFF, 0xFF, 0xFF][..]);
    let err = codec.decode(&mut hostile).unwrap_err();
    println!("hostile header: kind={:?} message={err}", err.kind());

    let mut dst = BytesMut::new();
    let err = codec.encode(Bytes::from(vec![0u8; 17]), &mut dst).unwrap_err();
    println!("oversized encode: kind={:?} message={err}", err.kind());
    println!("bytes written: {}", dst.len());
}

Output

After the decoder accepts a header, it reserves room for the payload in the buffer. With the default limit, one connection can therefore make the server reserve up to 8 MiB. Servers with many connections lower the limit to the largest legitimate message.

Example 11: Writing a custom codec

A custom codec implements Decoder and Encoder<Item>. Decoder needs an Item type, an Error type that implements From<io::Error>, and decode(&mut self, src: &mut BytesMut) -> Result<Option<Item>, Error>. The contract:

Situation	Return	Buffer
not enough bytes for an item	`Ok(None)`	leave unchanged (or call `reserve`)
a complete item	`Ok(Some(item))`	remove exactly the item's bytes
invalid data	`Err(error)`	anything

Framed calls decode repeatedly after each read until it returns Ok(None). Encoder<Item> needs an Error type that implements From<io::Error> and encode(&mut self, item: Item, dst: &mut BytesMut).

The protocol here has three commands. Each starts with a tag byte. Get (tag 1) has a 1-byte key length and the key. Set (tag 2) has a key the same way, then a 2-byte big-endian value length and the value. Quit (tag 3) has no fields. The value length is checked against a limit as soon as it is readable.

Rust

use bytes::{Buf, BufMut, BytesMut};
use std::io;
use tokio_util::codec::{Decoder, Encoder};

const MAX_VALUE: usize = 1024;

#[derive(Debug, PartialEq)]
enum Command {
    Get(String),
    Set(String, String),
    Quit,
}

#[derive(Debug)]
enum CodecError {
    Io(io::Error),
    UnknownTag(u8),
    ValueTooLarge(usize),
    KeyTooLong(usize),
    BadUtf8,
}

impl From<io::Error> for CodecError {
    fn from(e: io::Error) -> Self {
        CodecError::Io(e)
    }
}

fn utf8(bytes: &[u8]) -> Result<String, CodecError> {
    std::str::from_utf8(bytes)
        .map(str::to_owned)
        .map_err(|_| CodecError::BadUtf8)
}

struct CommandCodec;

impl Decoder for CommandCodec {
    type Item = Command;
    type Error = CodecError;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Command>, CodecError> {
        let Some(&tag) = src.first() else {
            return Ok(None);
        };
        match tag {
            3 => {
                src.advance(1);
                Ok(Some(Command::Quit))
            }
            1 => {
                if src.len() < 2 {
                    return Ok(None);
                }
                let key_len = src[1] as usize;
                if src.len() < 2 + key_len {
                    return Ok(None);
                }
                let key = utf8(&src[2..2 + key_len])?;
                src.advance(2 + key_len);
                Ok(Some(Command::Get(key)))
            }
            2 => {
                if src.len() < 2 {
                    return Ok(None);
                }
                let key_len = src[1] as usize;
                let value_head = 2 + key_len;
                if src.len() < value_head + 2 {
                    return Ok(None);
                }
                let value_len = u16::from_be_bytes([src[value_head], src[value_head + 1]]) as usize;
                if value_len > MAX_VALUE {
                    return Err(CodecError::ValueTooLarge(value_len));
                }
                let total = value_head + 2 + value_len;
                if src.len() < total {
                    src.reserve(total - src.len());
                    return Ok(None);
                }
                let key = utf8(&src[2..value_head])?;
                let value = utf8(&src[value_head + 2..total])?;
                src.advance(total);
                Ok(Some(Command::Set(key, value)))
            }
            other => Err(CodecError::UnknownTag(other)),
        }
    }
}

impl Encoder<Command> for CommandCodec {
    type Error = CodecError;

    fn encode(&mut self, item: Command, dst: &mut BytesMut) -> Result<(), CodecError> {
        match item {
            Command::Quit => dst.put_u8(3),
            Command::Get(key) => {
                let key_len = u8::try_from(key.len()).map_err(|_| CodecError::KeyTooLong(key.len()))?;
                dst.put_u8(1);
                dst.put_u8(key_len);
                dst.put_slice(key.as_bytes());
            }
            Command::Set(key, value) => {
                let key_len = u8::try_from(key.len()).map_err(|_| CodecError::KeyTooLong(key.len()))?;
                if value.len() > MAX_VALUE {
                    return Err(CodecError::ValueTooLarge(value.len()));
                }
                dst.put_u8(2);
                dst.put_u8(key_len);
                dst.put_slice(key.as_bytes());
                dst.put_u16(value.len() as u16);
                dst.put_slice(value.as_bytes());
            }
        }
        Ok(())
    }
}

fn main() {
    let mut codec = CommandCodec;
    let mut wire = BytesMut::new();
    codec.encode(Command::Set("color".into(), "blue".into()), &mut wire).unwrap();
    codec.encode(Command::Get("color".into()), &mut wire).unwrap();
    codec.encode(Command::Quit, &mut wire).unwrap();
    println!("{} bytes: {:02x?}", wire.len(), &wire[..]);

    while let Some(cmd) = codec.decode(&mut wire).unwrap() {
        println!("{cmd:?}");
    }
    println!("left over: {}", wire.len());

    let long_key = "k".repeat(300);
    let mut dst = BytesMut::new();
    println!("{:?}", codec.encode(Command::Get(long_key), &mut dst));
}

Output

decode never consumes bytes until a whole command is present, so the same Set can be retried after more bytes arrive. Checking the value length before waiting for the value keeps a hostile length from reserving memory.

Example 12: Testing a codec against split input

Because decode takes a plain buffer, a test can feed the same bytes cut at every possible position and check that the decoded commands are always the same. This is the split-invariance property from Bytes and Framing. The codec from Example 11 appears again in shortened form so the program is complete.

Rust

use bytes::{Buf, BufMut, BytesMut};
use std::io;
use tokio_util::codec::Decoder;

#[derive(Debug, PartialEq, Clone)]
struct Pair {
    key: String,
    value: String,
}

struct PairCodec;

impl Decoder for PairCodec {
    type Item = Pair;
    type Error = io::Error;

    // Wire: key_len u8, key, value_len u8, value.
    fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<Pair>> {
        let Some(&key_len) = src.first() else {
            return Ok(None);
        };
        let key_len = key_len as usize;
        let Some(&value_len) = src.get(1 + key_len) else {
            return Ok(None);
        };
        let total = 2 + key_len + value_len as usize;
        if src.len() < total {
            return Ok(None);
        }
        let key = String::from_utf8_lossy(&src[1..1 + key_len]).into_owned();
        let value = String::from_utf8_lossy(&src[2 + key_len..total]).into_owned();
        src.advance(total);
        Ok(Some(Pair { key, value }))
    }
}

fn main() {
    let mut wire = BytesMut::new();
    for (k, v) in [("a", "1"), ("name", "tokio"), ("", "empty key")] {
        wire.put_u8(k.len() as u8);
        wire.put_slice(k.as_bytes());
        wire.put_u8(v.len() as u8);
        wire.put_slice(v.as_bytes());
    }

    let decode_all = |pieces: Vec<&[u8]>| -> Vec<Pair> {
        let mut codec = PairCodec;
        let mut buf = BytesMut::new();
        let mut out = Vec::new();
        for piece in pieces {
            buf.extend_from_slice(piece);
            while let Some(pair) = codec.decode(&mut buf).unwrap() {
                out.push(pair);
            }
        }
        assert!(buf.is_empty());
        out
    };

    let expected = decode_all(vec![&wire[..]]);
    println!("whole buffer: {} pairs", expected.len());

    let mut splits_checked = 0;
    for at in 0..=wire.len() {
        let (left, right) = wire.split_at(at);
        assert_eq!(decode_all(vec![left, right]), expected);
        splits_checked += 1;
    }
    for size in 1..=wire.len() {
        assert_eq!(decode_all(wire.chunks(size).collect()), expected);
        splits_checked += 1;
    }
    println!("{splits_checked} different splits gave identical results");
    println!("{:?}", expected[2]);
}

Output

The two loops cover every two-piece split and every fixed chunk size from 1 byte up to the whole message. A decoder that consumes bytes before a frame is complete, or misjudges a boundary, fails one of these assertions.

Example 13: Errors end the stream, and EOF in the middle of a frame

When decode returns Err, FramedRead yields that error as an item and then returns None on the next poll. Code that loops while let Some(item) = framed.next().await sees the error item first, so it must handle the Err and stop. When the transport ends with bytes still in the buffer, decode_eof runs. Its default calls decode and, if bytes are left over, returns an io::Error with the message bytes remaining on stream.

Rust

use futures::StreamExt;
use tokio::io::{AsyncWriteExt, duplex};
use tokio_util::codec::{FramedRead, LengthDelimitedCodec};

#[tokio::main]
async fn main() {
    let (mut w, r) = duplex(64);
    w.write_all(&[0, 0, 0, 2, b'o', b'k', 0, 0, 0, 9, b'x']).await.unwrap();
    drop(w);

    let mut frames = FramedRead::new(r, LengthDelimitedCodec::new());
    println!("1: {:?}", frames.next().await);
    let second = frames.next().await.unwrap();
    println!("2: kind={:?} message={}", second.as_ref().unwrap_err().kind(), second.unwrap_err());
    println!("3: {:?}", frames.next().await.is_none());

    let (mut w, r) = duplex(64);
    w.write_all(&[0xFF, 0xFF, 0xFF, 0xFF]).await.unwrap();
    let mut frames = FramedRead::new(r, LengthDelimitedCodec::new());
    let err = frames.next().await.unwrap().unwrap_err();
    println!("oversized header: {err}");
    println!("after error: {:?}", frames.next().await.is_none());
}

Output

The first connection ends after a complete frame and a partial one. The partial one turns into the bytes remaining on stream error at end of input. In the second, the header claims 4,294,967,295 bytes. The default limit is 8 MiB, so the decoder errors on the header alone. An error from the decoder is not recoverable in general, because the position of the next frame is unknown, so a server closes the connection after the first error.

Example 14: A TCP server with Framed

Everything combines in a server. Framed<TcpStream, LengthDelimitedCodec> gives frames in and frames out. The server binds port 0 and the client reads the address from the listener. Only the frame contents are printed, so the output does not depend on the port.

Rust

use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use tokio::net::{TcpListener, TcpStream};
use tokio_util::codec::{Framed, LengthDelimitedCodec};

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    let server = tokio::spawn(async move {
        let (socket, _) = listener.accept().await.unwrap();
        let mut framed = Framed::new(socket, LengthDelimitedCodec::new());
        while let Some(frame) = framed.next().await {
            let frame = frame.unwrap();
            let reply = format!("echo {} bytes: {}", frame.len(), String::from_utf8_lossy(&frame));
            framed.send(Bytes::from(reply)).await.unwrap();
        }
    });

    let socket = TcpStream::connect(addr).await.unwrap();
    let mut framed = Framed::new(socket, LengthDelimitedCodec::new());
    for message in ["one", "two words", ""] {
        framed.send(Bytes::from(message)).await.unwrap();
        let reply = framed.next().await.unwrap().unwrap();
        println!("{}", String::from_utf8_lossy(&reply));
    }

    drop(framed);
    server.await.unwrap();
    println!("server saw the client disconnect");
}

Output

Each send writes a frame and each next waits for a whole reply frame. The empty message is a valid frame with a zero-length payload. When the client drops its Framed, the socket closes and the server's next() returns None.

Example 15: Backpressure in the write buffer

FramedWrite and Framed keep an internal write buffer. feed adds an encoded item to it and does not touch the transport until the buffer reaches the backpressure boundary, which is 8192 bytes by default (INITIAL_CAPACITY in the tokio-util source). At that point poll_ready flushes first, so a caller that feeds faster than the transport accepts is made to wait. set_backpressure_boundary changes the boundary.

The transport in the example is a duplex whose reader does not read, so it stops accepting data after its own 16-byte buffer is full.

Rust

use bytes::Bytes;
use futures::SinkExt;
use std::time::Duration;
use tokio::io::duplex;
use tokio_util::codec::{FramedWrite, LengthDelimitedCodec};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (writer, _reader_that_never_reads) = duplex(16);
    let mut sink = FramedWrite::new(writer, LengthDelimitedCodec::new());
    println!("default boundary: {}", sink.backpressure_boundary());
    sink.set_backpressure_boundary(32);

    let mut fed = 0;
    let outcome = tokio::time::timeout(Duration::from_secs(1), async {
        loop {
            sink.feed(Bytes::from_static(b"0123456789")).await.unwrap();
            fed += 1;
        }
    })
    .await;

    println!("feed calls that completed: {fed}");
    println!("stalled: {}", outcome.is_err());
    println!("buffered in sink: {} bytes", sink.write_buffer().len());
}

Output

Each frame is 14 bytes. Feeding continues without touching the transport until the buffer holds 32 bytes or more. Then feed must flush, the flush writes what the 16-byte duplex accepts, and it waits for a reader that never reads. The timeout cancels the stalled feed. A producer awaiting feed therefore slows to the speed of the peer, and memory use stays near the boundary plus one frame.

Example 16: Switching codecs mid-connection

Some protocols begin with a text handshake and continue in binary. Framed::map_codec swaps the codec and keeps the transport and the read buffer, including bytes already received. Here the first line arrives together with a length-prefixed frame in one write.

Rust

use futures::StreamExt;
use tokio::io::{AsyncWriteExt, duplex};
use tokio_util::codec::{Framed, LengthDelimitedCodec, LinesCodec};

#[tokio::main]
async fn main() {
    let (mut peer, io) = duplex(256);
    peer.write_all(b"HELLO v1\n\x00\x00\x00\x03abc").await.unwrap();
    drop(peer);

    let mut framed = Framed::new(io, LinesCodec::new());
    let greeting = framed.next().await.unwrap().unwrap();
    println!("greeting: {greeting:?}");
    println!("already buffered: {} bytes", framed.read_buffer().len());

    let mut framed = framed.map_codec(|_lines| LengthDelimitedCodec::new());
    let frame = framed.next().await.unwrap().unwrap();
    println!("frame: {:?}", frame);
    println!("end: {:?}", framed.next().await.is_none());
}

Output

into_parts() is the lower-level form: it returns the transport, the codec, and both buffers as a FramedParts value, and Framed::from_parts rebuilds a Framed from it. TLS upgrades use this to move a plain connection into an encrypted one without losing buffered bytes.

Example 17: StreamReader and ReaderStream

Two adapters convert between the stream world and the byte-reader world:

Adapter	From	To
`tokio_util::io::StreamReader`	a `Stream` of `Result<impl Buf, io::Error>`	`AsyncRead` (and `AsyncBufRead`)
`tokio_util::io::ReaderStream`	an `AsyncRead`	a `Stream` of `Result<Bytes, io::Error>`

StreamReader lets code that expects an AsyncRead (such as read_to_string or a LinesCodec FramedRead) consume chunks that arrive from a channel or an HTTP body. ReaderStream does the reverse: it reads in chunks of up to its capacity (4096 bytes by default; with_capacity sets it) and yields each chunk as Bytes.

Rust

use bytes::Bytes;
use futures::StreamExt;
use std::io;
use tokio::io::AsyncReadExt;
use tokio_util::io::{ReaderStream, StreamReader};

#[tokio::main]
async fn main() {
    let chunks = futures::stream::iter([
        Ok::<_, io::Error>(Bytes::from_static(b"hel")),
        Ok(Bytes::from_static(b"lo, ")),
        Ok(Bytes::from_static(b"stream")),
    ]);
    let mut reader = StreamReader::new(chunks);
    let mut text = String::new();
    reader.read_to_string(&mut text).await.unwrap();
    println!("StreamReader produced: {text:?}");

    let source: &[u8] = b"0123456789";
    let mut stream = ReaderStream::with_capacity(source, 4);
    while let Some(chunk) = stream.next().await {
        println!("chunk: {:?}", chunk.unwrap());
    }
}

Output

Example 18: Codecs available in tokio-util

Codec	Item	Framing
`LinesCodec`	`String`	`\n`, optional maximum length
`AnyDelimiterCodec`	`Bytes`	any of a set of delimiter bytes; named only here
`BytesCodec`	`BytesMut`	none; whatever is buffered
`LengthDelimitedCodec`	`BytesMut`	length prefix, configurable

SyncIoBridge (in tokio_util::io) wraps an AsyncRead or AsyncWrite so that blocking code, such as a synchronous parser or hasher, can use it from a thread that is allowed to block. It is named here and not used, because it requires spawn_blocking (Blocking and CPU-Bound Work) and is outside this lesson's scope.

For typed messages on the wire (JSON, MessagePack, or a binary format), the usual arrangement is LengthDelimitedCodec for the frames and a serialization step on each payload, as in Example 17 of Bytes and Framing. A custom Decoder that also parses the payload merges the two steps.

Interview points

A Decoder returns Ok(None) when it needs more bytes, Ok(Some(item)) after removing an item's bytes, and Err for invalid data. Framed calls it in a loop after each read.
Decoder::Error and Encoder::Error must implement From<io::Error>, so transport errors and protocol errors share one error type.
Framed is a Stream plus a Sink. FramedRead is only a Stream. FramedWrite is only a Sink. split() divides a Framed for use in two tasks.
send is feed plus flush. feed alone buffers until the backpressure boundary (8192 bytes by default) is reached.
The stream yields the decoder's error and then ends. Servers close the connection after a codec error because the next frame boundary is unknown.
LengthDelimitedCodec defaults: 4-byte big-endian length, 8 MiB maximum frame length. The decoder checks the limit on the header, before the payload arrives.
LinesCodec::new() has no length limit. Use new_with_max_length for input from untrusted peers.
decode_eof handles the end of the transport. Its default reports bytes remaining on stream when a partial frame is left.
A codec is a plain trait, so decode can be unit-tested on a BytesMut with input cut at every position.
StreamReader turns a stream of chunks into AsyncRead. ReaderStream turns AsyncRead into a stream of Bytes. SyncIoBridge lets blocking code read an async source from a blocking thread.

Summary

tokio-util's codec module separates framing (Decoder, Encoder) from I/O (Framed, FramedRead, FramedWrite).
LinesCodec, BytesCodec, and LengthDelimitedCodec cover text lines, raw chunks, and length-prefixed frames.
LengthDelimitedCodec::builder() sets field width, byte order, offset, adjustment, skipped bytes, and the maximum frame length.
A custom codec implements decode (peek, check limits, consume only complete items) and encode (append to the destination buffer).
Codec errors are yielded once and end the stream. End of input with leftover bytes is an error.
The write side buffers and applies backpressure at a configurable boundary; SinkExt::send flushes each item.
map_codec and into_parts change codecs on a live connection without losing buffered bytes.
StreamReader and ReaderStream convert between Stream<Item = Result<Bytes, _>> and AsyncRead.