Quick start: building an application

A WireframeApp collects route handlers and middleware. Each handler is stored as an Arc pointing to an async function that receives a packet reference and returns (). The builder caches these registrations until handle_connection constructs the middleware chain for an accepted stream.[^2]

use std::sync::Arc;
use wireframe::app::{Envelope, Handler, WireframeApp};

async fn ping(_env: &Envelope) {}

fn build_app() -> wireframe::Result<WireframeApp> {
    let handler: Handler<Envelope> = Arc::new(|env: &Envelope| {
        let _ = env; // inspect payload here
        Box::pin(ping(env))
    });

    WireframeApp::new()?
        .route(1, handler)?
        .wrap(wireframe::middleware::from_fn(|req, next| async move {
            let mut response = next.call(req).await?;
            response.frame_mut().extend_from_slice(b" pong");
            Ok(response)
        }))
}

The snippet below wires the builder into a Tokio runtime, decodes inbound payloads, and emits a serialized response. It showcases the typical main function for a microservice that listens on localhost and responds to a Ping message with a Pong payload.[^2][^10][^15]

use std::{net::SocketAddr, sync::Arc};

use wireframe::{
    app::{Envelope, Handler, WireframeApp},
    middleware,
    message::Message,
    server::{ServerError, WireframeServer},
};

#[derive(bincode::Encode, bincode::BorrowDecode, Debug)]
struct Ping {
    body: String,
}

#[derive(bincode::Encode, bincode::BorrowDecode, Debug, PartialEq)]
struct Pong {
    body: String,
}

async fn ping(env: &Envelope) {
    log::info!("received correlation id: {:?}", env.clone().into_parts().correlation_id());
}

fn build_app() -> wireframe::Result<WireframeApp> {
    let handler: Handler<Envelope> = Arc::new(|env: &Envelope| Box::pin(ping(env)));

    WireframeApp::new()?
        .route(1, handler)?
        .wrap(middleware::from_fn(|req, next| async move {
            let ping = Ping::from_bytes(req.frame()).map(|(msg, _)| msg).ok();
            let mut response = next.call(req).await?;

            if let Some(ping) = ping {
                let payload = Pong {
                    body: format!("pong {}", ping.body),
                }
                .to_bytes()
                .expect("encode Pong message");
                response.frame_mut().clear();
                response.frame_mut().extend_from_slice(&payload);
            }

            Ok(response)
        }))
}

fn app_factory() -> WireframeApp {
    build_app().expect("configure Wireframe application")
}

#[tokio::main]
async fn main() -> Result<(), ServerError> {
    let addr: SocketAddr = "127.0.0.1:4000".parse().expect("valid socket address");
    let server = WireframeServer::new(app_factory).bind(addr)?;
    server.run().await
}

Route identifiers must be unique; the builder returns wireframe::WireframeError::DuplicateRoute when a handler is registered twice, keeping the dispatch table unambiguous.[^2][^5] The crate-level wireframe::Result<T> alias always resolves to this canonical wireframe::WireframeError surface, so setup-time and streaming failures share one error contract.[^5] New applications default to the bundled bincode serializer, a length-delimited codec capped at 1024 bytes per frame, and a 100 ms read timeout. Clamp the length-delimited limit with buffer_capacity (length-delimited only), swap codecs with with_codec, and override the serializer with with_serializer when a different encoding strategy is required.[^3][^4] Use memory_budgets(...) to set explicit per-connection buffering caps for inbound assembly paths. Custom protocols implement FrameCodec to describe their framing rules. Changing frame budgets with buffer_capacity or swapping codecs with with_codec clears fragmentation settings, so call enable_fragmentation() (or fragmentation(Some(cfg))) again when transport fragmentation is required.

Once a stream is accepted—either from a manual accept loop or via WireframeServer—handle_connection(stream) builds (or reuses) the middleware chain, wraps the transport in the configured frame codec (length-delimited by default), enforces per-frame read timeouts, and writes responses. Serialization helpers send_response and send_response_framed (or send_response_framed_with_codec for custom codecs) return typed SendError variants when encoding or I/O fails, and the connection closes after ten consecutive deserialization errors.[^6][^7]

Custom protocols supply a FrameCodec implementation to describe their framing rules. The codec owns the Tokio Decoder and Encoder types, while Wireframe uses the trait surface to map frames to payload bytes and correlation data.

A codec implementation must:

• Define a Frame type and paired decoder/encoder implementations that return std::io::Error on failure.
• Return only the logical payload bytes from frame_payload so metadata parsing and deserialization run against the right buffer.
• Wrap outbound payloads with wrap_payload(&self, Bytes), adding any protocol headers or metadata required by the wire format.
• Provide correlation_id when the protocol stores it outside the payload; Wireframe only uses this hook when the deserialized envelope is missing a correlation identifier.
• Report max_frame_length, which clamps inbound frames and determines the budget used by enable_fragmentation.

Install a custom codec with with_codec. The builder disables fragmentation when codecs or the length-delimited frame budget change, so explicitly call enable_fragmentation() (or fragmentation(Some(cfg))) afterwards when transport fragmentation is required. Wireframe clones the codec per connection, so stateful codecs should ensure Clone produces an independent state (for example, reset sequence counters) when per-connection isolation is required. When a framed stream is already available, use send_response_framed_with_codec, so responses pass through FrameCodec::wrap_payload.

Assume MyCodec implements FrameCodec:

use std::sync::Arc;

use wireframe::app::{Envelope, Handler, WireframeApp};

struct MyCodec;

let handler: Handler<Envelope> = Arc::new(|_: &Envelope| Box::pin(async {}));

let app = WireframeApp::new()?
    .with_codec(MyCodec)
    .route(1, handler)?;

See examples/hotline_codec.rs and examples/mysql_codec.rs for complete implementations.

Codec accessor

Retrieve the configured codec from a WireframeApp instance:

use wireframe::app::WireframeApp;
use wireframe::codec::examples::HotlineFrameCodec;

let codec = HotlineFrameCodec::new(4096);
let app = WireframeApp::new()?.with_codec(codec);
let codec_ref = app.codec(); // &HotlineFrameCodec

Testing custom codecs with wireframe_testing

The wireframe_testing crate provides codec-aware driver functions that handle frame encoding and decoding transparently:

use wireframe::app::WireframeApp;
use wireframe::codec::examples::HotlineFrameCodec;
use wireframe_testing::{drive_with_codec_payloads, drive_with_codec_frames};

let codec = HotlineFrameCodec::new(4096);
let payload: Vec<u8> = vec![0x01, 0x02, 0x03];

// Payload-level: returns decoded response payloads as byte vectors.
let app = WireframeApp::new()?.with_codec(codec.clone());
let payloads =
    drive_with_codec_payloads(app, &codec, vec![payload.clone()]).await?;

// Frame-level: returns decoded codec frames for metadata inspection.
let app = WireframeApp::new()?.with_codec(codec.clone());
let frames =
    drive_with_codec_frames(app, &codec, vec![payload]).await?;

Available codec-aware driver functions:

• drive_with_codec_payloads / drive_with_codec_payloads_with_capacity — owned app, returns payload bytes.
• drive_with_codec_payloads_mut / drive_with_codec_payloads_with_capacity_mut — mutable app reference, returns payload bytes.
• drive_with_codec_frames / drive_with_codec_frames_with_capacity — owned app, returns decoded F::Frame values.

Supporting helpers for composing custom test patterns:

• encode_payloads_with_codec — encode payloads to wire bytes.
• decode_frames_with_codec — decode wire bytes to frames.
• extract_payloads — extract payload bytes from decoded frames.

Codec test fixtures

The wireframe_testing crate provides fixture functions for generating Hotline-framed wire bytes covering common test scenarios — valid frames, invalid frames, incomplete (truncated) frames, and frames with correlation metadata. These fixtures construct raw bytes directly, so they can represent malformed data that the encoder would reject:

use wireframe::codec::examples::HotlineFrameCodec;
use wireframe_testing::{
    valid_hotline_wire, oversized_hotline_wire,
    truncated_hotline_header, correlated_hotline_wire,
    decode_frames_with_codec,
};

let codec = HotlineFrameCodec::new(4096);

// Valid frame — decodes cleanly.
let wire = valid_hotline_wire(b"hello", 7);
let frames = decode_frames_with_codec(&codec, wire).unwrap();

// Oversized frame — rejected with "payload too large".
let wire = oversized_hotline_wire(4096);
assert!(decode_frames_with_codec(&codec, wire).is_err());

// Truncated header — rejected with "bytes remaining on stream".
let wire = truncated_hotline_header();
assert!(decode_frames_with_codec(&codec, wire).is_err());

// Correlated frames — all share the same transaction ID.
let wire = correlated_hotline_wire(42, &[b"a", b"b"]);
let frames = decode_frames_with_codec(&codec, wire).unwrap();

Available fixture functions:

• valid_hotline_wire / valid_hotline_frame — well-formed frames.
• oversized_hotline_wire — payload exceeds max_frame_length.
• mismatched_total_size_wire — header with incorrect total_size.
• truncated_hotline_header / truncated_hotline_payload — incomplete data.
• correlated_hotline_wire — frames sharing a transaction ID.
• sequential_hotline_wire — frames with incrementing transaction IDs.

Feeding partial frames and fragments

Real networks rarely deliver a complete codec frame in a single TCP read. The main crate now exposes these drivers through wireframe::testkit behind the opt-in testkit feature, while wireframe_testing keeps source-compatible re-exports for existing callers.

Chunked-write drivers encode payloads via a codec, concatenate the wire bytes, and write them in configurable chunk sizes (including one byte at a time):

use std::num::NonZeroUsize;
use wireframe::app::WireframeApp;
use wireframe::codec::examples::HotlineFrameCodec;
use wireframe::testkit::drive_with_partial_frames;

let codec = HotlineFrameCodec::new(4096);
let app = WireframeApp::new()?.with_codec(codec.clone());
let chunk = NonZeroUsize::new(1).expect("non-zero");
let payloads =
    drive_with_partial_frames(app, &codec, vec![vec![1, 2, 3]], chunk)
        .await?;

Available chunked-write driver functions:

• drive_with_partial_frames / drive_with_partial_frames_with_capacity — owned app, returns payload bytes.
• drive_with_partial_frames_mut — mutable app reference, returns payload bytes.
• drive_with_partial_codec_frames — owned app, returns decoded F::Frame values.

Fragment-feeding drivers accept a raw payload, fragment it with a Fragmenter, encode each fragment into a codec frame, and feed the frames through the app:

use std::num::NonZeroUsize;
use wireframe::app::WireframeApp;
use wireframe::codec::examples::HotlineFrameCodec;
use wireframe::fragment::Fragmenter;
use wireframe::testkit::drive_with_fragments;

let codec = HotlineFrameCodec::new(4096);
let app = WireframeApp::new()?.with_codec(codec.clone());
let fragmenter = Fragmenter::new(NonZeroUsize::new(20).unwrap());
let payloads =
    drive_with_fragments(app, &codec, &fragmenter, vec![0; 100]).await?;

Available fragment-feeding driver functions:

• drive_with_fragments / drive_with_fragments_with_capacity — owned app, returns payload bytes.
• drive_with_fragments_mut — mutable app reference, returns payload bytes.
• drive_with_fragment_frames — owned app, returns decoded F::Frame values.
• drive_with_partial_fragments — fragment AND feed in chunks, exercising both fragmentation and partial-frame buffering simultaneously.

Asserting reassembly outcomes

The wireframe::testkit::reassembly module provides non-panicking assertion helpers for both transport fragment reassembly and protocol message assembly. The API is built around lightweight snapshot structs, so the same helper can be used from rstest tests and rstest-bdd step definitions. Existing wireframe_testing::reassembly imports remain valid as compatibility re-exports.

use wireframe::message_assembler::MessageKey;
use wireframe::testkit::{
    MessageAssemblySnapshot,
    TestResult,
    assert_message_assembly_completed_for_key,
};

# fn check(snapshot: MessageAssemblySnapshot<'_>) -> TestResult {
assert_message_assembly_completed_for_key(snapshot, MessageKey(7), b"done")?;
# Ok(())
# }

Available message-assembly assertion helpers:

• assert_message_assembly_incomplete — verify the latest assembly is still awaiting more frames.
• assert_message_assembly_completed — verify the latest assembly completed with a specific body.
• assert_message_assembly_completed_for_key — verify the most recent completed message for a specific MessageKey.
• assert_message_assembly_error — verify structured failures such as sequence mismatch, duplicate first frame, and budget errors.
• assert_message_assembly_buffered_count / assert_message_assembly_total_buffered_bytes — verify cleanup and memory accounting side effects.
• assert_message_assembly_evicted — verify timeout-based eviction.

Available fragment-reassembly assertion helpers:

• assert_fragment_reassembly_absent — verify no full message has been reconstructed yet.
• assert_fragment_reassembly_completed_len — verify reconstructed payload length.
• assert_fragment_reassembly_error — verify reassembly errors including MessageTooLarge (with or without specific message ID), IndexMismatch (out-of-order fragments), MessageMismatch (fragments from wrong logical message), SeriesComplete (duplicate or late fragments after completion), and IndexOverflow (fragment index exceeding limits).
• assert_fragment_reassembly_buffered_messages — verify buffered partial message count.
• assert_fragment_reassembly_evicted — verify timeout-based eviction.

Test observability

The wireframe_testing crate provides an ObservabilityHandle that combines log capture with metrics recording in a single fixture. This is useful for asserting that codec errors, recovery policies, and other instrumentation behave correctly under test.

use wireframe_testing::ObservabilityHandle;

let mut obs = ObservabilityHandle::new();
obs.clear();

// Record metrics via a thread-local recorder.
metrics::with_local_recorder(obs.recorder(), || {
    wireframe::metrics::inc_codec_error("framing", "drop");
});

// Take a snapshot and query the captured counter.
obs.snapshot();
assert_eq!(
    obs.counter(
        wireframe::metrics::CODEC_ERRORS,
        &[("error_type", "framing"), ("recovery_policy", "drop")],
    ),
    1
);

The handle serializes access to the global logger via a mutex, so tests using it should run serially (--test-threads=1 for the affected binary). Metrics are captured per-thread via metrics::with_local_recorder, so metric-emitting code must run on the same thread as the handle. For async tests, use #[tokio::test(flavor = "current_thread")] or tokio::runtime::Runtime::new()?.block_on(...).

Available assertion helpers:

• counter(name, labels) / counter_without_labels(name) — query counter values from the most recent snapshot.
• codec_error_counter(error_type, recovery_policy) — convenience for the wireframe_codec_errors_total metric.
• assert_counter(name, labels, expected) — assert a counter matches.
• assert_no_metric(name) — assert no metric with the given name exists.
• assert_codec_error_counter(error_type, recovery_policy, expected) — assert a codec error counter matches.
• assert_log_contains(substring) — assert any captured log contains a substring.
• assert_log_at_level(level, substring) — assert a log at a specific level contains a substring.

Simulating slow readers and writers

Back-pressure tests often need more than partial-frame delivery. The wireframe_testing crate also provides slow-I/O helpers that pace the client write side, the client read side, or both directions at once.

Use SlowIoPacing to define a chunk size and inter-chunk delay, then apply it through SlowIoConfig:

Enable the feature in Cargo.toml when importing from the main crate:

wireframe = { version = "0.3.0", features = ["testkit"] }

use std::{num::NonZeroUsize, time::Duration};
use wireframe::{
    app::{Envelope, WireframeApp},
    codec::examples::HotlineFrameCodec,
    serializer::{BincodeSerializer, Serializer},
};
use wireframe::testkit::{
    SlowIoConfig, SlowIoPacing, drive_with_slow_codec_payloads,
};

let codec = HotlineFrameCodec::new(4096);
let app = WireframeApp::new()?.with_codec(codec.clone());
let config = SlowIoConfig::new()
    .with_writer_pacing(SlowIoPacing::new(
        NonZeroUsize::new(8).expect("non-zero"),
        Duration::from_millis(5),
    ))
    .with_reader_pacing(SlowIoPacing::new(
        NonZeroUsize::new(32).expect("non-zero"),
        Duration::from_millis(5),
    ))
    .with_capacity(64);

let request = BincodeSerializer.serialize(&Envelope::new(
    1,
    Some(7),
    vec![1, 2, 3],
))?;
let payloads = drive_with_slow_codec_payloads(app, &codec, vec![request], config)
    .await?;

Available slow-I/O helper functions:

• drive_with_slow_frames — pre-framed bytes, returns raw output bytes.
• drive_with_slow_payloads — default length-delimited payloads, returns raw output bytes.
• drive_with_slow_codec_payloads — codec-aware payloads, returns decoded payload byte vectors.
• drive_with_slow_codec_frames — codec-aware frames, returns decoded F::Frame values.

These helpers are designed for deterministic tests under paused Tokio time. Small duplex capacities should be used together with reader pacing when the app's outbound writes must hit back-pressure quickly.

In-process server/client pair harness

The wireframe_testing::client_pair module provides a harness for starting a real WireframeServer and a connected WireframeClient inside one test process. Unlike the drive_with_* helpers, which exercise server-side behaviour over in-memory tokio::io::duplex streams, the pair harness communicates over a real loopback TCP socket so compatibility assertions exercise the full client/server network path.

Use the pair harness when a downstream crate needs to verify that its protocol implementation works end-to-end through a real Wireframe server. A shared echo_app_factory provides a ready-made app factory with invocation counting:

use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use wireframe_testing::{
    TestResult,
    echo_app_factory,
    spawn_wireframe_pair,
};

async fn example() -> TestResult<()> {
    let counter = Arc::new(AtomicUsize::new(0));
    let mut pair = spawn_wireframe_pair(
        echo_app_factory(&counter),
        |builder| builder.max_frame_length(2048),
    )
    .await?;

    // pair.client_mut()? returns &mut WireframeClient for
    // request/response operations. Streaming responses borrow the
    // client exclusively, keeping that constraint visible.
    let addr = pair.local_addr();

    pair.shutdown().await?;
    Ok(())
}

spawn_wireframe_pair_default is a convenience wrapper that connects with default client settings when no builder customization is needed. If the client connection fails, the server task is torn down automatically so no orphaned tasks leak into subsequent tests.

The harness depends on the wireframe_testing dev-dependency crate. It does not require the main-crate testkit feature unless the test also uses wireframe::testkit helpers directly.

Zero-copy payload extraction

For performance-critical codecs, use Bytes instead of Vec<u8> for payload storage and override frame_payload_bytes to avoid allocation:

use bytes::Bytes;
use wireframe::codec::FrameCodec;

pub struct MyFrame {
    pub metadata: u32,
    pub payload: Bytes,  // Use Bytes, not Vec<u8>
}

impl FrameCodec for MyCodec {
    type Frame = MyFrame;
    // ... other associated types ...

    fn frame_payload(frame: &Self::Frame) -> &[u8] {
        &frame.payload
    }

    fn frame_payload_bytes(frame: &Self::Frame) -> Bytes {
        frame.payload.clone()  // Cheap: atomic reference count increment
    }

    fn wrap_payload(&self, payload: Bytes) -> Self::Frame {
        MyFrame {
            metadata: 0,
            payload,  // Store directly, no copy
        }
    }

    // ... other methods ...
}

In the decoder, use BytesMut::freeze() instead of .to_vec():

use bytes::BytesMut;
use tokio_util::codec::Decoder;

fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
    // ... parse header ...
    let payload = src.split_to(payload_len).freeze();  // Zero-copy
    Ok(Some(MyFrame { metadata, payload }))
}

The default frame_payload_bytes implementation copies from the frame_payload() slice, ensuring backward compatibility for codecs that use Vec<u8> payloads.

The codec layer provides a structured error taxonomy via CodecError that enables recovery policies, structured logging, and proper EOF handling.

• FramingError - Wire-level frame boundary issues (oversized frames, invalid length encoding, incomplete headers, checksum mismatches, empty frames).
• ProtocolError - Semantic violations after frame extraction (unknown message types, invalid versions).
• io::Error - Transport layer failures (connection resets, timeouts).
• EofError - End-of-stream conditions with clean/mid-frame/mid-header variants.

Each error type has a default recovery policy:

Override recovery policies with a custom RecoveryPolicyHook:

use wireframe::codec::{
    CodecError, CodecErrorContext, RecoveryPolicy, RecoveryPolicyHook,
};

struct StrictRecovery;

impl RecoveryPolicyHook for StrictRecovery {
    fn recovery_policy(&self, _error: &CodecError, _ctx: &CodecErrorContext) -> RecoveryPolicy {
        // Disconnect on any codec error
        RecoveryPolicy::Disconnect
    }
}

CodecErrorContext provides connection metadata for policy decisions:

use wireframe::codec::CodecErrorContext;

let ctx = CodecErrorContext::new()
    .with_connection_id(42)
    .with_correlation_id(123)
    .with_codec_state("seq=5");

The WireframeProtocol trait includes an on_eof hook for handling EOF conditions during frame decoding:

use wireframe::{
    codec::EofError,
    hooks::{ConnectionContext, WireframeProtocol},
};

impl WireframeProtocol for MyProtocol {
    type Frame = Vec<u8>;
    type ProtocolError = String;

    fn on_eof(&self, error: &EofError, partial_data: &[u8], _ctx: &mut ConnectionContext) {
        match error {
            EofError::CleanClose => tracing::info!("connection closed cleanly"),
            EofError::MidFrame { bytes_received, expected } => {
                tracing::warn!(
                    received = bytes_received,
                    expected = expected,
                    partial_len = partial_data.len(),
                    "connection closed mid-frame"
                );
            }
            EofError::MidHeader { bytes_received, header_size } => {
                tracing::warn!(
                    received = bytes_received,
                    header_size = header_size,
                    "connection closed mid-header"
                );
            }
        }
    }
}

When the metrics feature is enabled, codec errors increment the wireframe_codec_errors_total counter with error_type and recovery_policy labels:

wireframe_codec_errors_total{error_type="framing",recovery_policy="drop"} 5
wireframe_codec_errors_total{error_type="eof",recovery_policy="disconnect"} 2