Registration

register, register_json and register_avro put a schema under a subject; warm resolves an existing subject and registers nothing.

Registering is an explicit step of the producer that owns the subject, usually in on_startup. The registry keeps the id of a schema it already holds, so registering the same schema again changes nothing. A schema that the subject's compatibility level refuses fails the call, and the error carries the registry's account of what differs. Each call caches the subject, so the SchemaFrame layer finds it without a request.

register_json::<T> registers the JSON Schema that JsonSchema (schemars, re-exported) derives for T. register_avro::<T> registers the Avro schema that AvroSchema derives, in full: field defaults, aliases and docs stay, because a later reader resolves an older writer's datum with them. register takes a SchemaType and the definition text, which is how a Protobuf subject is registered from its .proto source. The registry paths of this crate read Protobuf schemas whose imports come from the well-known types.

warm resolves a subject's latest version and caches it, and fails when the registry has no such subject. It is the startup check of a subject that another team owns. lookup_id returns the id a subject already gave one exact schema, and is_compatible asks whether the subject would accept a schema; neither registers anything.

A codec registers in its own sense: AvroCodec::register and SchemaFramed::register record the subject a type publishes under, and the SchemaPrefetch resolves those subjects when the broker connects. The registry itself is named twice in such an app, building the prefetch and attaching it, and never at a mount site.

Example

Macros
Manual

examples/kafka_registry_register.rs
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Payment {
    id: i64,
    amount: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[outgoing(name = "invoices")]
struct Invoice {
    payment: i64,
    amount: i64,
}

#[subscriber("payments", reply)]
async fn bill(payment: &Payment) -> Invoice {
    tracing::info!(payment.id, payment.amount, "invoice issued");
    Invoice {
        payment: payment.id,
        amount: payment.amount,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("invoices-svc")
        .schema_registry(registry.clone());

    RustStream::new(AppInfo::new("invoices", "0.1.0"))
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            // This service owns the invoices subject; an identical schema keeps its id.
            registry.register_json::<Invoice>("invoices-value").await?;
            // The payments subject belongs to its producer: warm checks it and registers nothing.
            registry.warm("payments-value").await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(bill);
        })
}

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

examples/manual/kafka_registry_register.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Payment {
    id: i64,
    amount: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Invoice {
    payment: i64,
    amount: i64,
}

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

impl MessageHeaders for Invoice {
    type Contract = NoHeaders;
}

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

impl Handle<Payment, Invoice> for Bill {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Invoice, HandlerOutcome>> {
        tracing::info!(payment.id, payment.amount, "invoice issued");
        ready(Ok(Invoice {
            payment: payment.id,
            amount: payment.amount,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("invoices-svc")
        .schema_registry(registry.clone());

    RustStream::new(AppInfo::new("invoices", "0.1.0"))
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            // This service owns the invoices subject; an identical schema keeps its id.
            registry.register_json::<Invoice>("invoices-value").await?;
            // The payments subject belongs to its producer: warm checks it and registers nothing.
            registry.warm("payments-value").await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(subscriber("payments", Bill).reply().build());
        })
}

// 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