Codec

A codec of your own implements the Codec trait over any serde format and mounts wherever a built-in codec does.

The trait has two required methods: encode writes a value into a new buffer, and decode reads a value from the delivery's bytes. encode_into writes into a buffer the caller passes. The publish path calls it for a broker that only reads the payload, and inside a dispatch loop that buffer is reused, so a codec that implements it allocates nothing per publish. Its default encodes into a buffer of its own and copies. Both sides return a CodecError: the format's error goes in CodecError::Decode or CodecError::Encode as the source. CONTENT_TYPE is the media type the generated AsyncAPI document reports.

A codec generic over another composes with it: the inner codec decides the format, and the wrapper changes the bytes around it, such as a version byte in front. Both methods are synchronous, so a codec can only use what it holds and the bytes at hand. Work that needs I/O, such as a schema registry lookup, goes around the codec: in the broker's delivery path on the way in, and in a PublishLayer on the way out.

The example writes MessagePack with field names: a struct travels as a map, the form a peer in a dynamic language reads, where the built-in MessagePack codec writes an array. It uses rmp-serde, the crate behind the msgpack feature.

Example

The example builds with msgpack as the only codec feature:

cargo test --example custom_codec --no-default-features --features testing,macros,memory,msgpack

examples/custom_codec.rs
/// `MessagePack` that writes a struct as a map of field names, not as an array of values.
#[derive(Debug, Clone, Copy, Default)]
struct NamedMsgpack;

impl Codec for NamedMsgpack {
    const CONTENT_TYPE: &'static str = "application/msgpack";

    fn encode<T: Serialize>(&self, value: &T) -> Result<BytesMut, CodecError> {
        let mut buf = BytesMut::new();
        self.encode_into(value, &mut buf)?;
        Ok(buf)
    }

    fn encode_into<T: Serialize>(&self, value: &T, buf: &mut BytesMut) -> Result<(), CodecError> {
        // Straight into the publish path's buffer: nothing else is allocated.
        rmp_serde::encode::write_named(&mut buf.writer(), value)
            .map_err(|err| CodecError::Encode(Box::new(err)))
    }

    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, CodecError> {
        rmp_serde::from_slice(bytes).map_err(|err| CodecError::Decode(Box::new(err)))
    }
}

The codec mounts at the scope and at the reply position, like a built-in one.

Macros
Manual

examples/custom_codec.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker_codec(
        MemoryBroker::new(),
        NamedMsgpack,
        |b| {
            b.include(checkout).out_reply(Publish).codec(NamedMsgpack);
        },
    )
}

examples/manual/custom_codec.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker_codec(
        MemoryBroker::new(),
        NamedMsgpack,
        |b| {
            b.include(
                subscriber("orders", Checkout)
                    .reply()
                    .to("receipts")
                    .build(),
            )
            .out_reply(Publish)
            .codec(NamedMsgpack);
        },
    )
}

Example​

Example