Struct SchemaPrefetch 

pub struct SchemaPrefetch { /* private fields */ }

The async half of a registry-backed codec: it resolves, on the broker’s async edges, every schema the synchronous codec will read.

A Codec is synchronous on both ends and a registry lookup is not, and no amount of arranging makes those meet inside encode or decode: blocking a runtime worker from a sync function is not an option and guessing a schema is corruption. So the lookups move to the two places that are already async and already know when they have to happen - the broker’s connect, for the subjects a codec publishes under, and each subscription’s delivery path, for the writer schema an arriving envelope names. By the time a codec runs, what it needs is in the shared cache, and its own work is pure computation.

This is a distinct attachment from KafkaBroker::schema_registry on purpose. That one rewrites deliveries into JSON, which is the transcoding compatibility path; this one leaves every payload exactly as it arrived and only fills a cache. Attaching both to one broker is contradictory - the transcode would hand a JSON document to a codec expecting the Confluent wire format - and the prefetch runs first so it still sees the envelope.

Examples

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

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

#[derive(Serialize, AvroSchema, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: i64,
}

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

fn app() -> RustStream {
    let prefetch = SchemaPrefetch::new(SchemaRegistry::new("http://localhost:8081"));
    // Every type a codec registers records its subject here, so `connect` resolves it and a
    // subject that is missing is settled by the prefetch's policy rather than at first publish.
    let codec = AvroCodec::registry(&prefetch).register::<Shipment>("shipments-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("shipping-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("shipping", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(ship).out_reply(Publish::default()).codec(replies);
    })
}

Implementations

impl SchemaPrefetch

pub fn new(registry: SchemaRegistry) -> Self

Builds the prefetch over registry. Construction is synchronous and does no I/O; the resolving happens on the broker’s async edges.

pub fn check_compatibility(self, check: bool) -> Self

Whether connect checks each registered type’s schema the way registering it would be checked, and refuses to start when the registry says they are incompatible.

The subject’s compatibility level decides the versions the schema is held against: the latest one for BACKWARD, FORWARD and FULL, every version for their _TRANSITIVE forms. A schema this check passes is one the registry would accept.

It stands where Confluent’s latest.compatibility.strict does, and is on for the same reason it is on by default there: a model that has drifted from its subject is a fact worth learning while the app is starting rather than from a consumer that cannot read what it wrote. The registry’s own account of what differs, down to the field, travels in the error.

Turn it off for a registry whose compatibility level is deliberately NONE, or a client that cannot answer the question - a RegistryClient that does not implement is_compatible skips it regardless.

pub fn on_missing_subject(self, policy: MissingSubject) -> Self

What to do about a registered subject the registry does not hold; see MissingSubject. Defaults to Refuse.

Configure before building codecs from it: each copies the policy as it is built.

pub fn registry(&self) -> &SchemaRegistry

The client the codecs share a cache with.

Trait Implementations

impl Clone for SchemaPrefetch

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for SchemaPrefetch

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

Formats the value using the given formatter.