Avro

AvroCodec::registry reads each delivery with its writer schema and frames each value with the id of its subject.

AvroCodec is a codec, so a handler on it takes and returns ordinary serde structs. A model derives AvroSchema from apache-avro, which the crate re-exports as avro::AvroSchema. The codec is built from a SchemaPrefetch, and the same prefetch is attached to the broker with KafkaBroker::schema_prefetch. The codec itself is synchronous and reads only the prefetch's cache, so its encode and decode reach no network.

Decoding needs no registration. The envelope of each delivery names the writer schema, the prefetch resolves it before the decode, and the codec reads the datum with it. One codec therefore decodes every record type that arrives on its topics.

register::<T>(subject) is the publish side: values of T are framed with the id of subject. The codec finds the subject by the name serde gives the value's type. A value of a type that no one registered fails its publish, and the error names the type and lists the registered ones. Two registered types with one serde name stop the app at startup.

The codec named with with_broker_codec decodes the deliveries of its scope. A reply names its codec at its own position, .out_reply(policy).codec(codec); a reply without one is encoded with the default codec. The codec and the transcode are alternatives on one broker: a broker with schema_registry hands every subscription plain JSON.

Example

Macros
Manual

examples/kafka_avro_codec.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema)]
struct Delivery {
    id: i64,
    address: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, Outgoing)]
#[outgoing(name = "delivery-receipts")]
struct Receipt {
    id: i64,
    delivered: bool,
}

#[subscriber("deliveries", reply)]
async fn deliver(delivery: &Delivery) -> Receipt {
    tracing::info!(delivery.id, delivery.address, "delivered");
    Receipt {
        id: delivery.id,
        delivered: true,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // The prefetch resolves subjects at connect and each writer schema before its decode.
    let prefetch = SchemaPrefetch::new(registry);
    // Registration is the publish side: the subject a type's values are framed under.
    let codec = AvroCodec::registry(&prefetch).register::<Receipt>("delivery-receipts-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("deliveries-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("deliveries", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(deliver)
            .out_reply(Publish::default())
            .codec(replies);
    })
}

#[ruststream::app]
fn service() -> impl App {
    app(SchemaRegistry::new("http://localhost:8081"))
}

examples/manual/kafka_avro_codec.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Delivery {
    id: i64,
    address: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Receipt {
    id: i64,
    delivered: bool,
}

// `#[derive(Outgoing)]` with `#[outgoing(name = "delivery-receipts")]`.
impl OutgoingDestination for Receipt {
    type Form = FixedName;
    const DESTINATION: &'static str = "delivery-receipts";
}

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

/// The definition `#[subscriber("deliveries", reply)]` would have minted.
struct Deliver;

impl Handle<Delivery, Receipt> for Deliver {
    fn handle(
        &self,
        delivery: &Delivery,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(delivery.id, delivery.address, "delivered");
        ready(Ok(Receipt {
            id: delivery.id,
            delivered: true,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // The prefetch resolves subjects at connect and each writer schema before its decode.
    let prefetch = SchemaPrefetch::new(registry);
    // Registration is the publish side: the subject a type's values are framed under.
    let codec = AvroCodec::registry(&prefetch).register::<Receipt>("delivery-receipts-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("deliveries-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("deliveries", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(subscriber("deliveries", Deliver).reply().build())
            .out_reply(Publish::default())
            .codec(replies);
    })
}

// What `#[ruststream::app]` writes: the command line on a multi-threaded Tokio runtime.
fn main() -> ExitCode {
    run_main(|| app(SchemaRegistry::new("http://localhost:8081")))
}

Example​

Example