Trait Codec 

pub trait Codec: Send + Sync {
    const CONTENT_TYPE: &'static str = "application/octet-stream";

    // Required methods
    fn encode<T: Serialize>(&self, value: &T) -> Result<BytesMut, CodecError>;
    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, CodecError>;

    // Provided method
    fn encode_into<T: Serialize>(
        &self,
        value: &T,
        buf: &mut BytesMut,
    ) -> Result<(), CodecError> { ... }
}

A serializer that converts Rust values to and from bytes.

Implementations are stateless and cheap to clone. The trait uses generic methods rather than associated types so a single codec instance can handle any Serialize / DeserializeOwned value. This means dyn Codec is not object-safe; use generics or boxed concrete codecs at the call site.

encode and decode are synchronous: a codec holds only what a constant and the bytes at hand decide. An integration that needs I/O to serialize (a schema registry, a key service) goes around the codec, on the async edges: the broker’s delivery path on the way in, a PublishLayer on the way out.

Examples

use ruststream::BytesMut;
use ruststream::codec::{Codec, CodecError};
use ruststream::memory::prelude::*;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

/// JSON under the vendor media type the order contract was published with.
#[derive(Debug, Clone, Copy, Default)]
struct OrdersJson;

impl Codec for OrdersJson {
    const CONTENT_TYPE: &'static str = "application/vnd.orders.v1+json";

    fn encode<T: Serialize>(&self, value: &T) -> Result<BytesMut, CodecError> {
        let bytes =
            serde_json::to_vec(value).map_err(|error| CodecError::Encode(Box::new(error)))?;
        Ok(BytesMut::from(bytes.as_slice()))
    }

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

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    // The scope decodes with the codec, and the document reports its media type.
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker_codec(
        MemoryBroker::new(),
        OrdersJson,
        |b| {
            b.include(accept);
        },
    )
}

A codec generic over another composes: the inner one decides the format, and the wrapper transforms the bytes around it (a versioned envelope, a registry header, an encrypting layer). A failure of the wrapper becomes CodecError::Decode or CodecError::Encode with its own error as the source. The wrapper writes its version byte into the caller’s buffer in encode_into and hands the rest to the inner codec:

use std::fmt;

use ruststream::BytesMut;
use ruststream::codec::{Codec, CodecError, JsonCodec};
use serde::Serialize;
use serde::de::DeserializeOwned;

/// Frames another codec's bytes with a one-byte format version.
#[derive(Clone, Copy)]
struct Versioned<C>(C);

const VERSION: u8 = 1;

#[derive(Debug)]
struct BadVersion(u8);

impl fmt::Display for BadVersion {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "envelope version {} is not the version this build reads", self.0)
    }
}

impl std::error::Error for BadVersion {}

impl<C: Codec> Codec for Versioned<C> {
    const CONTENT_TYPE: &'static str = C::CONTENT_TYPE;

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

    fn encode_into<T: Serialize>(&self, value: &T, buf: &mut BytesMut) -> Result<(), CodecError> {
        buf.extend_from_slice(&[VERSION]);
        self.0.encode_into(value, buf)
    }

    fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, CodecError> {
        match bytes {
            [VERSION, payload @ ..] => self.0.decode(payload),
            [version, ..] => Err(CodecError::Decode(Box::new(BadVersion(*version)))),
            [] => Err(CodecError::Decode(Box::new(BadVersion(0)))),
        }
    }
}

let codec = Versioned(JsonCodec);
let bytes = codec.encode(&7u32)?;
assert_eq!(bytes[0], VERSION);
let back: u32 = codec.decode(&bytes)?;
assert_eq!(back, 7);

Provided Associated Constants

const CONTENT_TYPE: &'static str = "application/octet-stream"

The media type of the bytes this codec produces, reported as the contentType of every message it encodes or decodes in the generated AsyncAPI document.

The built-in codecs name application/json, application/cbor (RFC 8949) and application/msgpack. MessagePack has no registered media type; of the two spellings in use, application/msgpack is the one this crate reports.

The default is application/octet-stream: what a codec whose output carries no media type of its own should keep.

Examples

use ruststream::BytesMut;
use ruststream::codec::{Codec, CodecError};
use ruststream::memory::prelude::*;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

/// JSON under the vendor media type the order contract was published with: the generated
/// document reports it as the `contentType` of every message this codec carries.
#[derive(Debug, Clone, Copy, Default)]
struct OrdersJson;

impl Codec for OrdersJson {
    const CONTENT_TYPE: &'static str = "application/vnd.orders.v1+json";

    fn encode<T: Serialize>(&self, value: &T) -> Result<BytesMut, CodecError> {
        let bytes = serde_json::to_vec(value)
            .map_err(|error| CodecError::Encode(Box::new(error)))?;
        Ok(BytesMut::from(bytes.as_slice()))
    }

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

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    // The scope decodes with the codec, and the document reports its media type.
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker_codec(
        MemoryBroker::new(),
        OrdersJson,
        |b| {
            b.include(accept);
        },
    )
}

Required Methods

fn encode<T: Serialize>(&self, value: &T) -> Result<BytesMut, CodecError>

Encodes value into a mutable byte buffer.

Returning BytesMut lets the encoded buffer move into the publish pipeline (an Outgoing payload) without a copy, while still allowing publish middleware to mutate it in place. It is what a publish to a transport that keeps the payload (Take) calls, once per message.

Errors

Returns CodecError::Encode when the underlying serializer fails.

fn decode<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, CodecError>

Decodes bytes into a Rust value of type T.

Errors

Returns CodecError::Decode when the underlying deserializer fails.

Provided Methods

fn encode_into<T: Serialize>( &self, value: &T, buf: &mut BytesMut, ) -> Result<(), CodecError>

Encodes value into the end of a buffer the caller owns.

What a publish to a transport that only reads the payload (Lend) calls: inside a dispatch loop that buffer is the loop’s own, reused for every message, so a codec that writes into it leaves the publish path with nothing to allocate. Implement it wherever the serializer can write into a sink; the default encodes into a buffer of its own and copies, which is correct but pays the allocation this exists to avoid.

Errors

Returns CodecError::Encode when the underlying serializer fails. The buffer’s contents are then unspecified: the publish path discards what a failed encode wrote.

Examples

use ruststream::BytesMut;
use ruststream::codec::{Codec, CodecError, JsonCodec};
use ruststream::runtime::{Serialized, WireBytes};
use serde::Serialize;

#[derive(Serialize)]
struct Order {
    id: u64,
}

/// A value that brings its own wire format: a one-byte schema tag, then the JSON body, both
/// written straight into the buffer the publish lends it.
struct Tagged(Order);

impl Serialized for Tagged {
    type Error = CodecError;

    fn wire_bytes(&self, buf: &mut BytesMut) -> Result<WireBytes<'_>, CodecError> {
        buf.extend_from_slice(&[1]);
        JsonCodec.encode_into(&self.0, buf)?;
        Ok(WireBytes::InBuffer)
    }
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl Codec for CborCodec

const CONTENT_TYPE: &'static str = "application/cbor"

impl Codec for JsonCodec

const CONTENT_TYPE: &'static str = "application/json"

impl Codec for MsgpackCodec

const CONTENT_TYPE: &'static str = "application/msgpack"