Skip to main content

resolve_onto

resolve_onto reads every delivery onto a reader schema, so Avro fills the fields the writer never wrote from their defaults.

A datum read with its writer schema yields what the writer wrote. A consumer whose model has moved ahead of its producers names its own schema as the reader schema, and Avro's schema resolution fills each missing field from the default that schema declares. The model declares a default with #[avro(default = "..")], whose value is JSON. A field with no default that the writer never wrote fails the decode.

The reader schema applies to decoding only; a publish still writes the schema the wire carries. On a local codec the pair is checked when the codec is built, so a reader schema that cannot resolve against the writer schema is an error at startup. On a registry codec the writer schemas arrive with the deliveries, so a pair that cannot resolve fails the decode that meets it. A codec carrying a reader schema reads one type, so a handler that needs one mounts it in a Router with with_codec, and the scope's codec serves the other handlers.

Example​

examples/kafka_avro_resolve.rs
/// The schema the producers write, as the topic's owners publish it.
const WRITTEN: &str = r#"{"type": "record", "name": "Reading", "fields": [
    {"name": "sensor", "type": "string"},
    {"name": "celsius", "type": "double"}
]}"#;

/// This service's model: one field newer than what the producers write.
#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema)]
struct Reading {
    sensor: String,
    celsius: f64,
    #[avro(default = r#""unknown""#)]
    site: String,
}

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

fn codec() -> Result<AvroCodec, Box<dyn Error>> {
    let written = Schema::parse_str(WRITTEN)?;
    Ok(AvroCodec::local(written)?.resolve_onto(Reading::get_schema())?)
}

#[ruststream::app]
fn app() -> impl App {
    let codec = codec().expect("the model resolves against the written schema");
    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);
    })
}