Local

AvroCodec::local pins one schema and writes a bare datum, with no envelope, no registry and no I/O.

The schema passed in is the one the wire carries: T::get_schema() when the model defines it, or a parsed .avsc file when the schema text is the source of truth. Every message is written and read with that schema, so producers and consumers of the topic agree on it outside the service. local returns an error when the schema refers to a name it does not define. The codec registers nothing; register belongs to the registry form.

avro::encode and avro::decode are the same conversion as free functions, for a tool that works on datum bytes outside an app. Both use the type's own schema.

Example

Macros

examples/kafka_avro_local.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema)]
struct Reading {
    sensor: String,
    celsius: f64,
}

#[subscriber("readings")]
async fn record(reading: &Reading) -> HandlerOutcome {
    tracing::info!(reading.sensor, reading.celsius, "reading recorded");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The schema the wire carries, pinned here; nothing reaches a registry.
    let codec = AvroCodec::local(Reading::get_schema()).expect("a derived schema resolves");
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("readings-svc");
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(record);
    })
}

Manual

examples/manual/kafka_avro_local.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Reading {
    sensor: String,
    celsius: f64,
}

/// The definition `#[subscriber("readings")]` would have minted.
struct Record;

impl Handle<Reading> for Record {
    fn handle(
        &self,
        reading: &Reading,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(reading.sensor, reading.celsius, "reading recorded");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    // The schema the wire carries, pinned here; nothing reaches a registry.
    let codec = AvroCodec::local(Reading::get_schema()).expect("a derived schema resolves");
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("readings-svc");
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(subscriber("readings", Record).build());
    })
}

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