Bytes and Framing

A byte stream such as TCP delivers bytes in order but does not keep the boundaries between the writes that produced them. Framing is the rule a protocol uses to mark where one message ends and the next begins. After this lesson you can hold data in Bytes and BytesMut, read and write integers with Buf and BufMut, frame messages by delimiter or by length prefix, decode frames that arrive in pieces, and reject oversized frames before allocating memory for them.

Item	Role
`Bytes`	immutable, reference-counted byte buffer; clones and slices share memory
`BytesMut`	growable byte buffer; splits into `Bytes` without copying
`Buf`	read cursor trait: `get_u8`, `get_u32`, `advance`, `remaining`
`BufMut`	write cursor trait: `put_u8`, `put_u32`, `put_slice`
`AsyncReadExt::read_buf`	read from a socket into a `BufMut` such as `BytesMut`
delimiter framing	a marker byte (for example `\n`) ends each message
length-prefix framing	a fixed-size number in front of each message gives its length
maximum frame size	a limit checked before buffering a frame

The examples use these crates:

TOMLCargo.toml

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

Example 1: TCP-style streams have no message boundaries

tokio::io::duplex(n) creates two connected in-memory ends of a byte stream, like a socket pair, with a buffer of n bytes in each direction (AsyncRead and AsyncWrite introduces it). Here the writer sends two separate messages. The reader cannot tell where the first ends.

Rust

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

#[tokio::main]
async fn main() {
    let (mut writer, mut reader) = duplex(64);
    writer.write_all(b"hello").await.unwrap();
    writer.write_all(b"world").await.unwrap();

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

    writer.write_all(b"hello").await.unwrap();
    writer.write_all(b"world").await.unwrap();

    let mut small = [0u8; 4];
    for _ in 0..3 {
        let n = reader.read(&mut small).await.unwrap();
        println!("small read returned {n}: {:?}", std::str::from_utf8(&small[..n]).unwrap());
    }
}

Output

Two writes of 5 bytes became one read of 10 bytes. With a 4-byte buffer, the same 10 bytes came out in reads of 4, 4, and 2 bytes. The number of write calls on one side and read calls on the other are unrelated. A protocol has to carry its own boundaries. The rest of the lesson builds the tools for that.

Example 2: Bytes, an immutable shared buffer

Bytes holds a slice of bytes behind a reference count. It is immutable. clone copies a pointer and increments the count. It does not copy the data. Bytes::from_static wraps a &'static [u8] with no allocation. Bytes::copy_from_slice and Bytes::from(Vec<u8>) create owned data.

Rust

use bytes::Bytes;

fn main() {
    let a = Bytes::from_static(b"hello world");
    let b = Bytes::copy_from_slice(b"copied");
    let c = Bytes::from(vec![1u8, 2, 3]);
    println!("{a:?} len={}", a.len());
    println!("{b:?} {c:?}");

    let big = Bytes::from(vec![7u8; 1024]);
    let clone = big.clone();
    println!("same memory: {}", big.as_ptr() == clone.as_ptr());
    println!("equal: {}", big == clone);
}

Output

Bytes is Send and Sync, so a clone can move into another task cheaply. This is why buffers passed between tasks, or fanned out to many connections, are Bytes and not Vec<u8>.

Example 3: Slicing without copying

slice(range) returns a new Bytes that views part of the same allocation. split_to(n) removes the first n bytes from the original and returns them. split_off(n) removes everything from n on and returns it. All three share the original memory.

Rust

use bytes::Bytes;

fn main() {
    let data = Bytes::from_static(b"GET /index.html HTTP/1.1");
    let method = data.slice(0..3);
    let path = data.slice(4..15);
    println!("method={method:?} path={path:?}");
    println!("path starts inside data: {}", path.as_ptr() as usize - data.as_ptr() as usize);

    let mut rest = data.clone();
    let head = rest.split_to(4);
    println!("head={head:?} rest={rest:?}");

    let tail = rest.split_off(11);
    println!("rest={rest:?} tail={tail:?}");
}

Output

The path slice starts 4 bytes after the start of data, at the same address plus 4, so no bytes were copied. A parser can hand out many small Bytes values that all point into one received buffer. The allocation is freed when the last view is dropped.

Example 4: BytesMut, a growable buffer

BytesMut is the writable counterpart. It can grow, and freeze() turns it into Bytes without copying. split() takes all current contents as a new BytesMut and leaves self empty but keeping its remaining capacity. extend_from_slice appends.

Rust

use bytes::{BufMut, BytesMut};

fn main() {
    let mut buf = BytesMut::with_capacity(64);
    println!("len={} capacity>=64: {}", buf.len(), buf.capacity() >= 64);

    buf.extend_from_slice(b"hello ");
    buf.put_slice(b"world");
    println!("{buf:?} len={}", buf.len());

    let first = buf.split_to(6);
    println!("first={first:?} buf={buf:?}");

    let all = buf.split();
    println!("all={all:?} buf is empty: {}", buf.is_empty());

    let frozen = all.freeze();
    println!("frozen={frozen:?}");
}

Output

split_to(n) on a BytesMut is the operation a decoder uses to take one frame off the front of a receive buffer. It moves an index and does not copy. The frame and the buffer share one allocation until the frame is dropped.

Example 5: Reading with Buf

Buf is a read cursor. It is implemented for &[u8], Bytes, BytesMut, and others. Every get_* method reads a value and advances the cursor. remaining() gives the bytes left. chunk() is the current contiguous slice. advance(n) skips n bytes. copy_to_bytes(n) takes n bytes as Bytes.

Rust

use bytes::{Buf, Bytes};

fn main() {
    let mut buf = Bytes::from_static(&[0x01, 0x00, 0x02, 0x00, 0x00, 0x00, 0x03, b'a', b'b', b'c']);

    println!("remaining: {}", buf.remaining());
    println!("u8:  {}", buf.get_u8());
    println!("u16: {}", buf.get_u16());
    println!("u32: {}", buf.get_u32());
    println!("remaining: {}", buf.remaining());

    let text = buf.copy_to_bytes(2);
    println!("text={text:?} chunk={:?}", buf.chunk());
    buf.advance(1);
    println!("remaining after advance: {}", buf.remaining());

    let mut slice: &[u8] = &[9, 8, 7];
    println!("from &[u8]: {} then {} left", slice.get_u8(), slice.remaining());
}

Output

get_u16 and get_u32 read big-endian by default (Example 7). A plain &[u8] is also a Buf: reading from it moves the slice forward.

Example 6: Writing with BufMut

BufMut is a write cursor. It is implemented for BytesMut and Vec<u8>. Every put_* method appends a value. put_slice appends a byte slice. Writing to a BytesMut or Vec<u8> grows the buffer as needed.

Rust

use bytes::{BufMut, BytesMut};

fn main() {
    let mut out = BytesMut::new();
    out.put_u8(0xAB);
    out.put_u16(0x0102);
    out.put_u32(0xDEADBEEF);
    out.put_i16(-2);
    out.put_slice(b"ok");
    println!("{:02x?}", &out[..]);

    let mut v: Vec<u8> = Vec::new();
    v.put_u32(1);
    v.put_u8(b'!');
    println!("{v:?}");
}

Output

Example 7: Endianness

A number wider than one byte needs an order for its bytes. Big-endian puts the most significant byte first. Little-endian puts the least significant byte first. Network protocols conventionally use big-endian, called network byte order. put_u32 and get_u32 use big-endian. The _le variants (put_u32_le, get_u32_le) use little-endian. Sender and receiver must agree, and a protocol specification names the order for every multi-byte field.

Method	`0x01020304` on the wire
`put_u32` (big-endian)	`01 02 03 04`
`put_u32_le` (little-endian)	`04 03 02 01`

Rust

use bytes::{Buf, BufMut, BytesMut};

fn main() {
    let mut buf = BytesMut::new();
    buf.put_u32(0x0102_0304);
    buf.put_u32_le(0x0102_0304);
    println!("{:02x?}", &buf[..]);

    let mut read = buf.freeze();
    let right = read.get_u32();
    let wrong = read.get_u32();
    println!("big-endian read of first: {right:#x}");
    println!("big-endian read of second: {wrong:#x}");

    println!("{:02x?}", 0x0102_0304u32.to_be_bytes());
    println!("{:02x?}", 0x0102_0304u32.to_le_bytes());
}

Output

Reading the second value with the wrong order produces a different number, with no error. Mixed-up byte order is a silent bug.

Example 8: Reading past the end

get_* methods panic when fewer bytes remain than the value needs. try_get_* methods return Err(TryGetError) instead and leave the cursor unchanged. Data from a network is untrusted, so decoding code uses try_get_* or checks remaining() first.

Rust

use bytes::Buf;

fn main() {
    let mut buf: &[u8] = &[0x00, 0x01, 0x02];
    match buf.try_get_u32() {
        Ok(v) => println!("value {v}"),
        Err(e) => println!("error: {e}"),
    }
    println!("still {} bytes remaining", buf.remaining());
    println!("u16 works: {:?}", buf.try_get_u16());
}

Output

The panicking form:

Rustpanics at runtime

use bytes::Buf;

fn main() {
    let mut buf: &[u8] = &[0x00, 0x01, 0x02];
    let _ = buf.get_u32();
}

Output

Example 9: Delimiter framing

Delimiter framing ends each message with a marker byte. Text protocols such as SMTP, Redis inline commands, and line-based chat use \n. The decoder searches the receive buffer for the delimiter. If it finds one, it splits off everything up to and including it. If it does not, it returns None and waits for more bytes.

Rust

use bytes::{Buf, Bytes, BytesMut};

fn decode_line(buf: &mut BytesMut) -> Option<Bytes> {
    let newline = buf.iter().position(|b| *b == b'\n')?;
    let mut line = buf.split_to(newline + 1);
    line.truncate(newline);
    Some(line.freeze())
}

fn main() {
    let mut buf = BytesMut::new();
    for piece in [&b"PING\nPO"[..], b"NG\nHEL", b"LO", b"\n"] {
        buf.extend_from_slice(piece);
        println!("received {:?}, buffered {}", std::str::from_utf8(piece).unwrap(), buf.len());
        while let Some(line) = decode_line(&mut buf) {
            println!("  line {:?}", line);
        }
    }
    println!("left over: {}", buf.remaining());
}

Output

The first piece contains one whole line and the start of another. decode_line returns the whole one and leaves PO in the buffer. The pieces HEL and LO produce no line until the final \n arrives. A single network read can contain zero, one, or many frames, so the decoder is always called in a loop until it returns None.

Delimiter framing has two costs. The payload cannot contain the delimiter unless it is escaped, so binary data does not fit. And the decoder scans for the delimiter, so a large frame is scanned again on each new piece unless the search position is remembered.

Example 10: Length-prefix framing

Length-prefix framing puts the payload size in front of each message as a fixed-width number. The decoder reads the number, then waits until that many bytes are buffered. There is no scanning, and the payload can contain any byte. This lesson uses a 4-byte big-endian length. The encoder writes the header and the payload into a BytesMut.

Rust

use bytes::{BufMut, BytesMut};

fn encode_frame(payload: &[u8], dst: &mut BytesMut) {
    dst.reserve(4 + payload.len());
    dst.put_u32(payload.len() as u32);
    dst.put_slice(payload);
}

fn main() {
    let mut wire = BytesMut::new();
    encode_frame(b"hello", &mut wire);
    encode_frame(b"", &mut wire);
    encode_frame(&[0, 255, 10], &mut wire);
    println!("{:02x?}", &wire[..]);
    println!("total bytes: {}", wire.len());
}

Output

Each frame costs 4 bytes of overhead. An empty payload is legal and appears as 00 00 00 00. The binary payload 0, 255, 10 (which includes the newline byte 10) needs no escaping.

Example 11: Incremental decoding of partial frames

A decoder is a function of the buffer. It takes &mut BytesMut and does one of three things:

Buffer state	Result	Buffer after
fewer than 4 bytes	`Ok(None)`	unchanged
header present, payload incomplete	`Ok(None)`	unchanged
header and full payload present	`Ok(Some(frame))`	frame removed

The decoder reads the length with u32::from_be_bytes on a slice of the buffer. That reads the header without consuming it, so an incomplete frame can be retried later from the same starting position.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};

fn encode_frame(payload: &[u8], dst: &mut BytesMut) {
    dst.put_u32(payload.len() as u32);
    dst.put_slice(payload);
}

fn decode_frame(buf: &mut BytesMut) -> Option<Bytes> {
    if buf.len() < 4 {
        return None;
    }
    let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
    if buf.len() < 4 + len {
        return None;
    }
    buf.advance(4);
    Some(buf.split_to(len).freeze())
}

fn main() {
    let mut wire = BytesMut::new();
    encode_frame(b"first", &mut wire);
    encode_frame(b"second frame", &mut wire);
    let wire = wire.freeze();

    let mut buf = BytesMut::new();
    for (i, piece) in wire.chunks(5).enumerate() {
        buf.extend_from_slice(piece);
        print!("piece {i} ({} bytes), buffered {}:", piece.len(), buf.len());
        let mut found = 0;
        while let Some(frame) = decode_frame(&mut buf) {
            print!(" frame {:?}", std::str::from_utf8(&frame).unwrap());
            found += 1;
        }
        if found == 0 {
            print!(" incomplete");
        }
        println!();
    }
    println!("left over: {}", buf.len());
}

Output

The wire is 25 bytes, delivered in pieces of 5. The header of the first frame arrives in piece 0 with one payload byte, and the frame completes in piece 1. The second frame's header and payload straddle several pieces, and the decoder returns None on each until the last byte arrives. Splitting the input at any position gives the same frames. That property is the test for a decoder.

Example 12: Reading frames from an async stream with read_buf

A real receive loop reads from the socket into a BytesMut, then decodes every complete frame. AsyncReadExt::read_buf(&mut buf) reads into the spare capacity of any BufMut, grows a BytesMut when it is full, and appends what it read. It returns 0 at end of stream (or when the buffer has no room, which does not happen for BytesMut). It is cancel safe.

The writer here sends the wire bytes 3 at a time with a pause between writes. The paused clock makes each pause pass instantly, and the reader always receives exactly one write at a time. Time: sleep, timeout, and interval explains the clock.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt, duplex};
use tokio::time::sleep;

fn encode_frame(payload: &[u8], dst: &mut BytesMut) {
    dst.put_u32(payload.len() as u32);
    dst.put_slice(payload);
}

fn decode_frame(buf: &mut BytesMut) -> Option<Bytes> {
    if buf.len() < 4 {
        return None;
    }
    let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
    if buf.len() < 4 + len {
        return None;
    }
    buf.advance(4);
    Some(buf.split_to(len).freeze())
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (mut client, mut server) = duplex(1024);

    let mut wire = BytesMut::new();
    encode_frame(b"hi", &mut wire);
    encode_frame(b"framing", &mut wire);

    tokio::spawn(async move {
        for chunk in wire.chunks(3) {
            client.write_all(chunk).await.unwrap();
            sleep(Duration::from_millis(10)).await;
        }
    });

    let mut buf = BytesMut::new();
    loop {
        let n = server.read_buf(&mut buf).await.unwrap();
        println!("read {n} bytes, buffered {}", buf.len());
        while let Some(frame) = decode_frame(&mut buf) {
            println!("  frame {:?}", std::str::from_utf8(&frame).unwrap());
        }
        if n == 0 {
            break;
        }
    }
    println!("left over: {}", buf.len());
}

Output

Every read returned 3 bytes, so frames became visible only when their last byte arrived. The final read returned 0 because the writer task finished and dropped its end, which closes the stream.

Example 13: The maximum frame size

The length field is chosen by the sender. A 4-byte field can claim up to 4,294,967,295 bytes. A decoder that trusts it will reserve that much memory, or wait forever for it, and a client can open many connections that each send only a header. That is a denial-of-service attack that costs the attacker four bytes per connection.

The defence is a maximum frame size: check the length as soon as the header is readable, before reserving or buffering any payload, and fail the connection when the length exceeds the limit.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};

const MAX_FRAME: usize = 1024;

#[derive(Debug)]
enum FrameError {
    TooLarge { len: usize, max: usize },
}

fn decode_frame(buf: &mut BytesMut) -> Result<Option<Bytes>, FrameError> {
    if buf.len() < 4 {
        return Ok(None);
    }
    let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
    if len > MAX_FRAME {
        return Err(FrameError::TooLarge { len, max: MAX_FRAME });
    }
    if buf.len() < 4 + len {
        return Ok(None);
    }
    buf.advance(4);
    Ok(Some(buf.split_to(len).freeze()))
}

fn main() {
    let mut buf = BytesMut::new();
    buf.put_u32(5);
    buf.put_slice(b"fine!");
    println!("{:?}", decode_frame(&mut buf));

    buf.put_u32(u32::MAX);
    println!("{:?}", decode_frame(&mut buf));

    println!("buffered after error: {} bytes", buf.len());
    println!("a naive decoder would have reserved {} bytes", u32::MAX);
}

Output

The check uses only the 4 header bytes, so the attacker cannot make the server buffer more than MAX_FRAME + 4 bytes per connection. After a TooLarge error the byte stream is out of sync, so the caller closes the connection and does not try to skip the frame. The limit should be the largest legitimate message plus some margin. A server with C connections and limit M needs about C * (M + 4) bytes of receive buffer in the worst case.

Example 14: The delimiter version of the limit

Delimiter framing needs the same defence. A client can send bytes forever without a newline, and the buffer grows with every byte. The decoder rejects the buffer when it holds more than the limit and still contains no delimiter.

Rust

use bytes::{Bytes, BytesMut};

const MAX_LINE: usize = 8;

#[derive(Debug)]
enum LineError {
    TooLong,
}

fn decode_line(buf: &mut BytesMut) -> Result<Option<Bytes>, LineError> {
    let window = buf.len().min(MAX_LINE + 1);
    match buf[..window].iter().position(|b| *b == b'\n') {
        Some(newline) => {
            let mut line = buf.split_to(newline + 1);
            line.truncate(newline);
            Ok(Some(line.freeze()))
        }
        None if buf.len() > MAX_LINE => Err(LineError::TooLong),
        None => Ok(None),
    }
}

fn main() {
    let mut buf = BytesMut::from(&b"short\nthis line is far too long"[..]);
    println!("{:?}", decode_line(&mut buf));
    println!("{:?}", decode_line(&mut buf));

    let mut exact = BytesMut::from(&b"12345678\n"[..]);
    println!("{:?}", decode_line(&mut exact));
}

Output

The search covers at most MAX_LINE + 1 bytes, so the scan cost per call is bounded too. A line of exactly MAX_LINE bytes plus its newline is accepted.

Example 15: End of stream in the middle of a frame

When the stream ends, the buffer should be empty. If bytes remain, the peer disconnected in the middle of a frame. That is an error and is different from a clean close between frames.

Rust

use bytes::{Buf, BufMut, BytesMut};
use std::io;
use tokio::io::{AsyncReadExt, AsyncWriteExt, duplex};

async fn read_all_frames(
    reader: &mut (impl AsyncReadExt + Unpin),
) -> io::Result<Vec<Vec<u8>>> {
    let mut buf = BytesMut::new();
    let mut frames = Vec::new();
    loop {
        while buf.len() >= 4 {
            let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
            if buf.len() < 4 + len {
                break;
            }
            buf.advance(4);
            frames.push(buf.split_to(len).to_vec());
        }
        if reader.read_buf(&mut buf).await? == 0 {
            return if buf.is_empty() {
                Ok(frames)
            } else {
                Err(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    format!("stream ended with {} unfinished bytes", buf.len()),
                ))
            };
        }
    }
}

#[tokio::main]
async fn main() {
    let (mut w, mut r) = duplex(64);
    let mut wire = BytesMut::new();
    wire.put_u32(2);
    wire.put_slice(b"ok");
    w.write_all(&wire).await.unwrap();
    drop(w);
    println!("clean close: {:?}", read_all_frames(&mut r).await);

    let (mut w, mut r) = duplex(64);
    let mut wire = BytesMut::new();
    wire.put_u32(10);
    wire.put_slice(b"only5");
    w.write_all(&wire).await.unwrap();
    drop(w);
    println!("cut off: {:?}", read_all_frames(&mut r).await);
}

Output

UnexpectedEof is the standard io::ErrorKind for a stream that ends before a complete item.

Example 16: Zero-copy frames from one buffer

Because split_to and freeze share the allocation, decoding many frames from one large buffer copies no payload bytes. Every frame is a view into the original memory, at a computable offset. The frames remain valid after the buffer that produced them is dropped, because each Bytes holds a reference count.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};

fn main() {
    let mut buf = BytesMut::new();
    for word in ["alpha", "beta", "gamma"] {
        buf.put_u32(word.len() as u32);
        buf.put_slice(word.as_bytes());
    }
    let base = buf.as_ptr() as usize;

    let mut frames: Vec<Bytes> = Vec::new();
    while buf.len() >= 4 {
        let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
        buf.advance(4);
        frames.push(buf.split_to(len).freeze());
    }
    drop(buf);

    for frame in &frames {
        let offset = frame.as_ptr() as usize - base;
        println!("{:?} at offset {offset}", frame);
    }
}

Output

Offsets 4, 13, and 21 match the layout: a 4-byte header, then 5 payload bytes, then the next header at 9, its payload at 13, and so on. All three frames point into the same allocation, and the buffer was dropped before they were printed.

Example 17: A binary message format

A frame carries an opaque payload. The payload has its own structure. This message type has a 1-byte tag followed by tag-specific fields, all big-endian. The decoder reads the fields with try_get_*, so a payload that is shorter than its tag requires becomes an error and not a panic.

Rust

use bytes::{Buf, BufMut, Bytes, BytesMut};

#[derive(Debug, PartialEq)]
enum Message {
    Ping(u32),
    Text { from: u16, body: String },
}

#[derive(Debug, PartialEq)]
enum DecodeError {
    Truncated,
    UnknownTag(u8),
    BadUtf8,
}

impl Message {
    fn encode(&self, dst: &mut BytesMut) {
        match self {
            Message::Ping(id) => {
                dst.put_u8(1);
                dst.put_u32(*id);
            }
            Message::Text { from, body } => {
                dst.put_u8(2);
                dst.put_u16(*from);
                dst.put_u16(body.len() as u16);
                dst.put_slice(body.as_bytes());
            }
        }
    }

    fn decode(mut payload: Bytes) -> Result<Message, DecodeError> {
        let tag = payload.try_get_u8().map_err(|_| DecodeError::Truncated)?;
        match tag {
            1 => {
                let id = payload.try_get_u32().map_err(|_| DecodeError::Truncated)?;
                Ok(Message::Ping(id))
            }
            2 => {
                let from = payload.try_get_u16().map_err(|_| DecodeError::Truncated)?;
                let len = payload.try_get_u16().map_err(|_| DecodeError::Truncated)? as usize;
                if payload.remaining() < len {
                    return Err(DecodeError::Truncated);
                }
                let body = String::from_utf8(payload.copy_to_bytes(len).to_vec())
                    .map_err(|_| DecodeError::BadUtf8)?;
                Ok(Message::Text { from, body })
            }
            other => Err(DecodeError::UnknownTag(other)),
        }
    }
}

fn main() {
    let mut buf = BytesMut::new();
    Message::Ping(7).encode(&mut buf);
    Message::Text { from: 3, body: "hi".into() }.encode(&mut buf);
    println!("{:02x?}", &buf[..]);

    let ping = Message::decode(buf.split_to(5).freeze());
    let text = Message::decode(buf.split().freeze());
    println!("{ping:?}");
    println!("{text:?}");

    println!("{:?}", Message::decode(Bytes::from_static(&[1, 0, 0])));
    println!("{:?}", Message::decode(Bytes::from_static(&[9])));
    println!("{:?}", Message::decode(Bytes::from_static(&[2, 0, 1, 0, 2, 0xFF, 0xFE])));
}

Output

The frame layer guarantees a whole payload. The message layer decides what the bytes mean and rejects malformed content. Keeping the two layers separate makes each one testable alone.

Example 18: Writing frames with write_all_buf

AsyncWriteExt::write_all_buf(&mut buf) writes every byte of any Buf to a writer and advances the buffer to empty. It suits an encode-then-write pattern: encode several frames into one BytesMut, then send them with one call.

Rust

use bytes::{BufMut, BytesMut};
use tokio::io::{AsyncReadExt, AsyncWriteExt, duplex};

#[tokio::main]
async fn main() {
    let (mut w, mut r) = duplex(64);

    let mut out = BytesMut::new();
    for word in ["one", "two", "three"] {
        out.put_u32(word.len() as u32);
        out.put_slice(word.as_bytes());
    }
    println!("encoded {} bytes", out.len());

    w.write_all_buf(&mut out).await.unwrap();
    println!("after write_all_buf: {} bytes left in buffer", out.len());
    drop(w);

    let mut received = Vec::new();
    r.read_to_end(&mut received).await.unwrap();
    println!("received {} bytes", received.len());
}

Output

Batching several frames into one write reduces the number of system calls and the number of small TCP segments.

Example 19: Choosing a framing method

Property	Delimiter	Length prefix
Binary payloads	need escaping	any bytes
Decode cost	scan for the delimiter	read a fixed header
Human-readable on the wire	yes	no
Size known before the payload arrives	no	yes
Limit check	on buffered length without a delimiter	on the header, before buffering
Resynchronize after garbage	skip to the next delimiter	not possible; close the connection
Typical use	text protocols (SMTP, IRC, Redis inline)	binary RPC (gRPC messages, Kafka, Postgres)

Some protocols combine both. HTTP/1.1 uses a delimiter (\r\n\r\n) for headers, then either a Content-Length (length prefix) or chunked encoding (a length line per chunk) for the body.

Interview points

TCP is a byte stream. A single read can return part of a message, one message, or several. The number of write calls does not map to the number of read calls.
Framing options: delimiter, length prefix, fixed-size records, or a self-describing format. Length prefix is the default for binary protocols.
A decoder returns None when the buffer does not hold a complete frame, consumes nothing in that case, and is called in a loop after each read.
Split-invariance is the correctness test: the same bytes cut at any positions must produce the same frames.
Validate the length field before allocating. Without a maximum frame size, a 4-byte header can force a multi-gigabyte reservation. Close the connection on violation.
Delimiter framing also needs a maximum length, because a peer can send bytes without ever sending the delimiter.
Bytes clones and slices share one reference-counted allocation. BytesMut::split_to(n).freeze() yields a frame with no copy.
Buf::get_* panics on short input. try_get_* returns an error. Decode untrusted input with the checked forms.
get_u32 is big-endian. get_u32_le is little-endian. Network protocols use big-endian unless the specification says otherwise.
read_buf appends to a BytesMut, grows it, and is cancel safe. It returns 0 at end of stream. Bytes left in the buffer at end of stream mean a truncated frame.

Summary

A byte stream has no message boundaries. Protocols add them with delimiters or length prefixes.
Bytes is immutable and reference-counted; clone and slice are constant-time. BytesMut is growable; split_to, split, and freeze move data between the two without copying.
Buf reads and advances; BufMut writes and grows. Both are implemented for common types including &[u8], Vec<u8>, Bytes, and BytesMut.
Big-endian is the default for get_* and put_*; _le methods use little-endian.
A frame decoder takes &mut BytesMut and returns Ok(None) for incomplete input, Ok(Some(frame)) for a complete frame, and Err for a protocol violation.
The maximum frame size is checked when the header becomes readable. It bounds per-connection memory to about the limit.
read_buf reads a socket into a BytesMut. End of stream with a non-empty buffer is an unexpected-EOF error.
write_all_buf writes a whole Buf, so several encoded frames can go out in one call.