rust 55 lines · 8 steps

A length-prefixed frame codec in Tokio

How a length-prefixed codec splits a byte stream into discrete frames and serializes them back.

Explained by highlit
Tokio Intermediate framing codecs buffers error-handling networking
1use bytes::{Buf, BufMut, BytesMut};
2use tokio_util::codec::{Decoder, Encoder};
3 
4const MAX_FRAME: usize = 8 * 1024 * 1024;
5const HEADER_LEN: usize = 4;
6 
7#[derive(Debug, thiserror::Error)]
8pub enum FrameError {
9 #[error("frame of {0} bytes exceeds maximum of {MAX_FRAME}")]
10 TooLarge(usize),
11 #[error(transparent)]
12 Io(#[from] std::io::Error),
13}
14 
15#[derive(Default)]
16pub struct FrameCodec;
17 
18impl Decoder for FrameCodec {
19 type Item = BytesMut;
20 type Error = FrameError;
21 
22 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
23 if src.len() < HEADER_LEN {
24 return Ok(None);
25 }
26 
27 let len = u32::from_be_bytes(src[..HEADER_LEN].try_into().unwrap()) as usize;
28 if len > MAX_FRAME {
29 return Err(FrameError::TooLarge(len));
30 }
31 
32 let total = HEADER_LEN + len;
33 if src.len() < total {
34 src.reserve(total - src.len());
35 return Ok(None);
36 }
37 
38 src.advance(HEADER_LEN);
39 Ok(Some(src.split_to(len)))
40 }
41}
42 
43impl Encoder<&[u8]> for FrameCodec {
44 type Error = FrameError;
45 
46 fn encode(&mut self, item: &[u8], dst: &mut BytesMut) -> Result<(), Self::Error> {
47 if item.len() > MAX_FRAME {
48 return Err(FrameError::TooLarge(item.len()));
49 }
50 dst.reserve(HEADER_LEN + item.len());
51 dst.put_u32(item.len() as u32);
52 dst.put_slice(item);
53 Ok(())
54 }
55}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A 4-byte length prefix lets a codec turn an arbitrary byte stream into cleanly delimited messages.
  2. 2Returning Ok(None) is how a Tokio decoder signals it needs more bytes before it can produce a frame.
  3. 3Enforcing a maximum frame size on both decode and encode guards against unbounded memory allocation from untrusted input.

Related explainers

rust 
use axum::{
    extract::State,
    http::StatusCode,
    Json,

Idempotent webhook handling in Axum

deduplication idempotency shared-state
Intermediate 8 steps
ruby 
class DatasetImportJob < ApplicationJob
  queue_as :default
 
  def perform(import)

Streaming import progress with a Rails job

background-jobs turbo-streams progress-tracking
Intermediate 9 steps
rust 
use axum::{
    body::{Body, Bytes},
    extract::Request,
    http::StatusCode,

Logging request bodies in Axum middleware

middleware streaming-body logging
Intermediate 8 steps
rust 
use std::time::Duration;
 
use axum::{
    body::Body,

Turning Axum timeouts into 504 JSON

middleware timeouts error-handling
Intermediate 7 steps
rust 
use axum::{
    extract::State,
    response::sse::{Event, KeepAlive, Sse},
};

Broadcasting live prices over SSE in Axum

server-sent-events broadcast-channel streams
Advanced 9 steps
go 
package tsvio
 
import (
	"bufio"

Reading and writing TSV records in Go

parsing io-streams error-handling
Intermediate 10 steps

Share this explainer

Here's the card — post it anywhere.

A length-prefixed frame codec in Tokio — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code