Cache

SchemaCachePolicy bounds the id entries and ages the subject entries, since an id never changes and a subject's latest version moves.

The registry assigns an id per distinct schema, so an entry keyed by id is never stale; it takes a bound and no expiry. A subject's latest version moves whenever someone registers one, so its entry takes an expiry and no bound. Cached holds up to capacity ids, drops the oldest past it, and trusts a subject's version for subject_ttl. The default keeps 1000 ids and never looks a subject up again, as Confluent's own clients do. Disabled remembers nothing: the transcode works under it, and the synchronous codecs, which read the cache and cannot await a miss, do not.

subject_ttl governs the async lookups: warm and the publish side of the transcode. A codec resolves its subject once, at connect, and keeps it. Ids are numbered per schema context, so one client stays inside one context; an app that reaches across contexts uses a client per context.

SchemaRegistry::cache_policy sets the policy of the default cache, MemorySchemaCache. with_cache puts a SchemaCache of your own in its place. That trait is synchronous, because a codec reads it while it encodes or decodes; fetching belongs to the RegistryClient.

Example

Macros
Manual

examples/kafka_registry_cache.rs
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Purchase {
    id: i64,
    total: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    purchase: i64,
    total: i64,
}

#[subscriber("purchases", reply)]
async fn acknowledge(purchase: &Purchase) -> Receipt {
    tracing::info!(purchase.id, purchase.total, "receipt issued");
    Receipt {
        purchase: purchase.id,
        total: purchase.total,
    }
}

fn registry(url: &str) -> SchemaRegistry {
    // Up to 256 schemas by id; a subject's latest version is looked up again after 5 minutes,
    // so a version registered meanwhile reaches the replies without a restart.
    SchemaRegistry::new(url).cache_policy(SchemaCachePolicy::Cached {
        capacity: nonzero!(256usize),
        subject_ttl: Some(Duration::from_secs(300)),
    })
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("receipts-svc")
        .schema_registry(registry.clone());
    RustStream::new(AppInfo::new("receipts", "0.1.0"))
        .publish_layer(SchemaFrame::new(registry))
        .with_broker(broker, |b| {
            b.include(acknowledge);
        })
}

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

examples/manual/kafka_registry_cache.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Purchase {
    id: i64,
    total: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Receipt {
    purchase: i64,
    total: i64,
}

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

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

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

impl Handle<Purchase, Receipt> for Acknowledge {
    fn handle(
        &self,
        purchase: &Purchase,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(purchase.id, purchase.total, "receipt issued");
        ready(Ok(Receipt {
            purchase: purchase.id,
            total: purchase.total,
        }))
    }
}

fn registry(url: &str) -> SchemaRegistry {
    // Up to 256 schemas by id; a subject's latest version is looked up again after 5 minutes,
    // so a version registered meanwhile reaches the replies without a restart.
    SchemaRegistry::new(url).cache_policy(SchemaCachePolicy::Cached {
        capacity: nonzero!(256usize),
        subject_ttl: Some(Duration::from_secs(300)),
    })
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("receipts-svc")
        .schema_registry(registry.clone());
    RustStream::new(AppInfo::new("receipts", "0.1.0"))
        .publish_layer(SchemaFrame::new(registry))
        .with_broker(broker, |b| {
            b.include(subscriber("purchases", Acknowledge).reply().build());
        })
}

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

Example​

Example