SchemaPrefetch

SchemaPrefetch resolves every registered subject at connect, which keeps the codecs' encode and decode free of I/O.

A codec is synchronous, and a registry lookup is not. The prefetch is the async half: every codec built from it, AvroCodec::registry or SchemaFramed::new, records its registered subjects on it. KafkaBroker::schema_prefetch attaches it, and the broker's connect resolves each subject before the app takes a delivery. On the consume path the prefetch resolves the writer schema that an arriving envelope names before the codec decodes the delivery, and leaves the payload as it arrived. A lookup that fails there is logged, and the decode that follows settles the delivery by the subscription's failure policy.

The registry appears twice in an app: once building the prefetch, once attaching it. A mount site never names it. Every codec comes from the one prefetch, and what varies per mount is the subject. A Router inside a broker scope overrides the scope's codec for the handlers it carries, by the usual codec cascade.

check_compatibility is on by default. At connect each registered type's schema is checked the way registering it would be, against the versions the subject's compatibility level names: the latest one, or every version under a _TRANSITIVE level. A model that has drifted from its subject stops the app, and the error carries the registry's own account of the difference. check_compatibility(false) turns the check off, for a registry whose level is NONE. A subject the registry does not hold is settled by MissingSubject.

The prefetch and the transcode are alternatives on one broker: the prefetch reads the envelope, and the transcode replaces it with JSON.

Example

Macros
Manual

examples/kafka_registry_prefetch.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema)]
struct Order {
    id: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order: i64,
    carrier: String,
}

#[subscriber("orders", reply)]
async fn ship(order: &Order) -> Shipment {
    tracing::info!(order.id, "order shipped");
    Shipment {
        order: order.id,
        carrier: "post".to_owned(),
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // At connect: resolve each registered subject and check the type's schema against it.
    let prefetch = SchemaPrefetch::new(registry).check_compatibility(true);
    let codec = AvroCodec::registry(&prefetch).register::<Shipment>("shipments-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("shipping-svc")
        .schema_prefetch(prefetch);

    let replies = codec.clone();
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(ship).out_reply(Publish::default()).codec(replies);
    })
}

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

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

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Shipment {
    order: i64,
    carrier: String,
}

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

impl MessageHeaders for Shipment {
    type Contract = NoHeaders;
}

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

impl Handle<Order, Shipment> for Ship {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Shipment, HandlerOutcome>> {
        tracing::info!(order.id, "order shipped");
        ready(Ok(Shipment {
            order: order.id,
            carrier: "post".to_owned(),
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // At connect: resolve each registered subject and check the type's schema against it.
    let prefetch = SchemaPrefetch::new(registry).check_compatibility(true);
    let codec = AvroCodec::registry(&prefetch).register::<Shipment>("shipments-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("shipping-svc")
        .schema_prefetch(prefetch);

    let replies = codec.clone();
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(subscriber("orders", Ship).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