Struct AvroCodec 

pub struct AvroCodec { /* private fields */ }

An Avro Codec: the schema lives in the codec, and the messages riding it are ordinary serde types.

Avro is a schema-driven format with a serde front end, so the codec position is where it belongs: apache-avro converts a value through its own dynamic representation in both directions, guided by the schema the codec holds, and asks nothing of the message type beyond Serialize / Deserialize. A handler on this codec is an ordinary handler and its models are ordinary structs - no derive of ours, no wire type in the signature.

The two schema sources

local pins one schema at construction. The wire carries a bare Avro datum, nothing reaches a registry, and the whole path is I/O-free - which is what a topic with a fixed schema, and every unit test, wants.

registry speaks the Confluent wire format: the encode side frames with the id its subject holds, and the decode side reads each delivery with the writer schema that delivery’s envelope names, so a producer still on an older version stays readable. It needs a SchemaPrefetch attached to the broker, because those lookups are async and a Codec is not; see that type for why the split falls where it does.

Schema evolution

Reading a datum with its writer schema recovers what the writer wrote, and no more. A field the writer never had is filled from a reader schema’s default, which is Avro’s own schema resolution, so a consumer that has moved ahead of its producers names the schema it expects with resolve_onto. Without it a model with a field the writer lacks fails to deserialize, which is the honest outcome: the value is genuinely not on the wire.

Examples

A topic with one fixed schema and no registry, consumed through an ordinary handler:

use apache_avro::AvroSchema;
use ruststream_rdkafka::KafkaError;
use ruststream_rdkafka::avro::AvroCodec;
use ruststream_rdkafka::prelude::*;
use serde::Deserialize;

#[derive(Deserialize, AvroSchema)]
struct Order {
    id: i64,
    item: String,
}

#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, item = %order.item, "placed");
    HandlerOutcome::ack()
}

fn app() -> Result<RustStream, KafkaError> {
    let codec = AvroCodec::local(Order::get_schema())?;
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("orders-svc");
    Ok(RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(place);
    }))
}

Implementations

impl AvroCodec

pub fn local(schema: Schema) -> Result<Self, KafkaError>

A codec over one schema, known here and used for every message.

Nothing on this path reaches a registry and nothing frames an envelope: the wire carries the datum alone. Pass the schema the wire is supposed to carry - T::get_schema() when the model defines it, a parsed .avsc when the schema is the source of truth.

Errors

Returns KafkaError::WireFormat when the schema’s own names cannot be resolved (a reference to a record the schema does not define).

pub fn registry(prefetch: &SchemaPrefetch) -> Self

A codec on the Confluent wire format, publishing what register records.

Construction is synchronous and does no I/O; the subjects are resolved when the broker connects, and each delivery’s writer schema on its consume path, both by the prefetch this is built from.

Examples

Each delivery decodes with the writer schema its envelope names, so a datum an older producer wrote stays readable; each reply is framed with its subject’s id:

use apache_avro::AvroSchema;
use ruststream_rdkafka::avro::AvroCodec;
use ruststream_rdkafka::prelude::*;
use ruststream_rdkafka::{SchemaPrefetch, SchemaRegistry};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, AvroSchema)]
struct Order {
    id: i64,
}

#[derive(Serialize, Deserialize, AvroSchema, Outgoing)]
struct Confirmation {
    id: i64,
}

#[subscriber("orders", reply("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

fn app() -> RustStream {
    let prefetch = SchemaPrefetch::new(SchemaRegistry::new("http://localhost:8081"));
    let codec = AvroCodec::registry(&prefetch).register::<Confirmation>("confirmations-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        .schema_prefetch(prefetch);
    // The scope's codec decodes every delivery; the reply names it again, at its own
    // position.
    let replies = codec.clone();
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(confirm).out_reply(Publish::default()).codec(replies);
    })
}

pub fn register<T: Serialize + AvroSchema>( self, subject: impl Into<String>, ) -> Self

Records that values of T publish under subject.

One codec serves as many message types as a router mounts through it, so this is a list rather than a second codec each time. The subject is looked up per publish by the name serde gives the value’s type, which is the only identity Codec::encode can recover - TypeId needs T: 'static, and encode bounds T by Serialize alone.

T::get_schema() is captured too, and used for exactly one thing: putting the subject back under MissingSubject::AutoRegister. What a datum is written with is always the schema the framed id names, so a drifted local model cannot silently produce records the id contradicts.

Panics

Panics when two registered types share a serde name, which would make the subject of a publish ambiguous. That is a startup failure by design: the alternative is picking one of them per message and writing records under the wrong subject. Rename one with #[serde(rename = "..")]. Also panics when the prefetch’s internal mutex is poisoned, or when T’s own schema cannot be serialized.

Examples

Registration is the publish side: decoding needs none of it, since the writer schema comes off the envelope’s id. The registry is named once, building the prefetch every codec is minted from:

use apache_avro::AvroSchema;
use ruststream_rdkafka::avro::AvroCodec;
use ruststream_rdkafka::prelude::*;
use ruststream_rdkafka::schema_registry::{SchemaPrefetch, SchemaRegistry};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, AvroSchema)]
pub struct Order {
    pub id: i64,
}

#[derive(Serialize, Deserialize, AvroSchema, Outgoing)]
#[outgoing(name = "confirmations")]
pub struct Confirmation {
    pub id: i64,
}

#[subscriber("orders", reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

pub fn app() -> RustStream {
    let prefetch = SchemaPrefetch::new(SchemaRegistry::new("http://localhost:8081"));
    let codec = AvroCodec::registry(&prefetch).register::<Confirmation>("confirmations-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        .schema_prefetch(prefetch);
    // The scope's codec decodes every delivery; the reply names it again, at its own
    // position.
    let replies = codec.clone();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker_codec(broker, codec, |b| {
            b.include(confirm).out_reply(Publish::default()).codec(replies);
        })
}

pub fn resolve_onto(self, schema: Schema) -> Result<Self, KafkaError>

Reads every delivery onto schema instead of the schema it was written with, so Avro’s resolution fills the fields this consumer expects and the writer never wrote.

This is the reader schema of Avro’s resolution rules, and it is the decode side only: a publish writes what the wire is supposed to carry, which is the writer schema either way.

Errors

Returns KafkaError::WireFormat when the schema cannot be prepared against the local writer schema. A registry codec’s writer schemas arrive later, so a pairing that cannot resolve is reported by the decode that meets it.

Trait Implementations

impl Clone for AvroCodec

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Codec for AvroCodec

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

Encodes value into a mutable byte buffer.

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

Decodes bytes into a Rust value of type T.

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.

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

where T: Serialize,

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

impl Debug for AvroCodec

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.