Codecs

A codec controls how a topic's messages cross the wire. Every topic carries a Codec associated type that decides how T::Message is encoded on publish and decoded on consume. The default is JSON, which means existing code keeps working without any changes; opting into a different encoding is a one-line edit on the topic definition.

The codec slot is per-topic, not per-broker. Two topics on the same backend may speak entirely different wire formats. This is the right boundary: a topic models a single event flow, and the encoding is a property of that flow, not of the transport.

When to swap the default

Three situations come up in practice:

• Protobuf, when you need a compact, schema-evolution-aware wire format, or when you share message contracts with services written in other languages.
• SBE, when latency dominates: Simple Binary Encoding frames publish and decode with zero copies, and handlers read fields in place through generated flyweight decoders.
• Raw bytes, when the payload is already encoded by an upstream component (Confluent Schema Registry's framed Avro, opaque blobs from a legacy system, a third-party format with its own framing).
• A custom codec, when none of the above fit — for example, CBOR or a domain-specific framing.

Outside those situations, stick with JSON; it is the default, requires no extra dependencies, and is the easiest format to debug.

The Codec trait

The trait is short:

pub trait Codec<M>: Send + Sync + 'static {
    const NAME: &'static str;
    fn encode(value: &M) -> Result<Vec<u8>>;
    fn decode(bytes: &[u8]) -> Result<M>;

    // Provided: override only for zero-copy codecs.
    fn encode_bytes(value: &M) -> Result<Bytes> { /* wraps encode */ }
    fn decode_owned(bytes: Bytes) -> Result<M> { /* borrows and delegates */ }
}

encode and decode are associated functions, not methods on a value. The codec carries no state; the macros plug the type in directly. NAME is a stable label used in logs.

encode_bytes and decode_owned are the owned-buffer variants the backends actually call where their clients allow it. The provided implementations delegate to encode/decode, so a custom codec only needs the two required functions; codecs whose message type already holds encoded bytes (SBE below) override them to pass the buffer through without copying.

JsonCodec — the default

JsonCodec is what you get when define_topic! is invoked without a codec = … clause:

define_topic!(
    OrderSettlement,
    SettlementEvent,
    TopologyBuilder::new("order-settlement").dlq().build()
);

This compiles down to a topic with type Codec = JsonCodec;. Payloads ride the wire as serde_json::to_vec output, and serde_json::from_slice is called on the consumer side. SettlementEvent must implement Serialize and DeserializeOwned.

ProtobufCodec

Enable the protobuf cargo feature, derive prost::Message on the payload, and pass the codec to the macro:

#[derive(Clone, PartialEq, ::prost::Message)]
struct OrderEvent {
    #[prost(string, tag = "1")]
    order_id: String,
    #[prost(double, tag = "2")]
    amount: f64,
}
 
shove::define_topic!(
    Orders,
    OrderEvent,
    TopologyBuilder::new("kafka-orders-proto").dlq().build(),
    codec = ProtobufCodec
);

ProtobufCodec is a single marker that implements Codec<M> for every M: prost::Message + Default. One topic can use codec = ProtobufCodec for OrderEvent, another for UserCreated, and so on — each call site picks the matching impl through <ProtobufCodec as Codec<T::Message>>. Encoding pre-sizes the buffer via encoded_len, so a single allocation covers the common case. Decoding failures surface as ShoveError::Codec { codec: "protobuf", .. } rather than Serialization, so handlers and middleware can distinguish protobuf-decode errors from JSON ones.

Runnable example: examples/kafka/protobuf_pubsub.rs.

Generating Rust types from a .proto file

The example above defines OrderEvent inline with #[derive(prost::Message)]. In production you usually drive the schema from a .proto file and let prost-build generate the Rust types at compile time. shove sees the generated type the same as a hand-written one — anything implementing prost::Message + Default works.

Add prost-build as a build dependency:

# Cargo.toml
[dependencies]
shove = { version = "0.11", features = ["kafka", "protobuf"] }
prost = "0.13"
 
[build-dependencies]
prost-build = "0.13"

Write the schema:

// proto/order.proto
syntax = "proto3";
package myservice;
 
message OrderEvent {
  string order_id = 1;
  double amount = 2;
}

Compile it at build time:

// build.rs
fn main() -> Result<(), Box<dyn std::error::Error>> {
    prost_build::compile_protos(&["proto/order.proto"], &["proto/"])?;
    Ok(())
}

Then include the generated module and bind it to a topic:

use shove::{define_topic, ProtobufCodec, TopologyBuilder};

// The module name matches the `.proto`'s `package` declaration.
mod myservice {
    include!(concat!(env!("OUT_DIR"), "/myservice.rs"));
}

define_topic!(
    Orders,
    myservice::OrderEvent,
    TopologyBuilder::new("orders").dlq().build(),
    codec = ProtobufCodec
);

For gRPC services, swap prost-build for tonic-build — the generated message types still implement prost::Message, so ProtobufCodec works on them unchanged. If you already depend on a separate crate that publishes generated protobuf types, depend on it as a normal dependency and skip the build.rs step.

shove deliberately does not bundle a .proto compiler — prost-build and tonic-build are the standard tools and would only duplicate them. The protobuf feature in shove is the runtime codec only.

SbeCodec — zero-copy

Enable the sbe cargo feature (it adds no dependencies). Simple Binary Encoding is a fixed-layout binary format from the low-latency trading world: instead of deserializing into an owned struct, generated flyweight codecs read and write fields directly on a byte buffer. shove preserves that model end to end.

An SBE topic carries SbeFrame<T>: an owned, header-validated wire frame. T is a zero-sized type tag that binds the topic to one SBE message type via the constants sbe-tool generates:

use shove::{define_topic, SbeCodec, SbeFrame, SbeMessage, TopologyBuilder};

struct Order;
impl SbeMessage for Order {
    const SCHEMA_ID: u16 = order_codec::SBE_SCHEMA_ID;
    const TEMPLATE_ID: u16 = order_codec::SBE_TEMPLATE_ID;
    // BYTE_ORDER defaults to little-endian, the SBE default.
}

define_topic!(
    Orders,
    SbeFrame<Order>,
    TopologyBuilder::new("orders").dlq().build(),
    codec = SbeCodec
);

On the publish side, encode once with the generated encoder (message header included) and hand the buffer to SbeFrame::new, which validates the header and takes ownership without copying:

let frame = SbeFrame::<Order>::new(encoded)?;
publisher.publish::<Orders>(&frame).await?;

On the consume side the handler receives the SbeFrame and wraps the generated decoder straight over frame.as_bytes(). There is no intermediate struct and, on backends that own the delivery buffer, no copy: the handler reads the same allocation the client received.

What the codec validates and what it deliberately does not:

• schemaId and templateId are checked against T on every decode (and at SbeFrame::new), so a frame from the wrong topic or message type surfaces as ShoveError::Codec { codec: "sbe", .. } instead of being misread.
• The wire version and blockLength are accepted as-is, following SBE extension semantics. The acting values are exposed through frame.header() so the generated decoder can handle frames from older or newer schema versions.

Per-backend copy behavior:

SNS/SQS and Redis Streams speak string payloads, so SbeCodec refuses them unconditionally at publish time with ShoveError::Codec { codec: "sbe", .. }. The rejection does not depend on the frame's bytes: even a frame that happens to be valid UTF-8 is refused, so failures are deterministic instead of data-dependent.

RawBytesCodec — the escape hatch

RawBytesCodec is a passthrough for payloads that are already encoded. The topic carries Vec<u8> and the handler owns every wire-format decision:

shove::define_topic!(
    RawTopic,
    Vec<u8>,
    TopologyBuilder::new("raw-bytes").dlq().build(),
    codec = RawBytesCodec
);

The canonical use case is Confluent Schema Registry. Records on that wire arrive as [magic_byte, schema_id (4 bytes), avro_payload...]. The library doesn't ship a Schema Registry client, but RawBytesCodec lets the handler strip the 5-byte framing header, look up the schema, and decode the Avro payload itself. Re-publishing works the same way in reverse: prepend the framing header before publishing.

Reach for RawBytesCodec only when the format genuinely isn't serde-friendly. Once the bytes leave the handler, the type system can no longer help you.

Runnable example: examples/nats/raw_bytes.rs.

Migrating hand-rolled Topic impls

If you implement Topic by hand instead of via define_topic!, you now need an explicit type Codec line:

impl Topic for OrderSettlement {
    type Message = SettlementEvent;
    type Codec = JsonCodec;
    fn topology() -> &'static QueueTopology { /* ... */ }
}

JsonCodec preserves the old behaviour exactly. Macro-generated topics already do this for you.

Custom codecs

Implement Codec<M> for your own type when neither JSON, Protobuf, nor raw bytes fits. The contract is round-trip safety: decode(encode(m).unwrap()).unwrap() == m for every m the codec supports. Encode and decode errors should surface as ShoveError::Codec { codec: "<name>", source } so consumers see a consistent error variant.