SchemaFramed

SchemaFramed wraps a codec such as JsonCodec in the Confluent envelope, so a registry-backed JSON topic is a codec choice.

A JSON document reads without its schema, so the envelope separates from the format: the inner codec writes and reads the document, and SchemaFramed puts the envelope on and takes it off. Decoding strips the envelope and looks nothing up. Encoding frames with the id of the subject that register::<T>(subject) records for T, found by the name serde gives the value's type. The JSON Schema of T is captured with it, for MissingSubject::AutoRegister. The wrapper frames and does not validate: the registry enforces compatibility between versions, and a service that validates each message passes an inner codec that does.

SchemaFramed::new takes the SchemaPrefetch that the broker carries, which resolves the registered subjects at connect. The codec named with with_broker_codec decodes the deliveries of its scope, and a reply names it at its own position with .out_reply(policy).codec(codec). An Avro datum cannot be read without its writer schema, so Avro has a codec of its own, AvroCodec.

Example

Macros
Manual

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

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema, Outgoing)]
#[outgoing(name = "statements")]
struct Statement {
    account: i64,
    balance: i64,
}

#[subscriber("accounts", reply)]
async fn issue(account: &Account) -> Statement {
    tracing::info!(account.id, "statement issued");
    Statement {
        account: account.id,
        balance: 0,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    let prefetch = SchemaPrefetch::new(registry);
    // Decoding strips the envelope; encoding frames with the id of the type's subject.
    let codec = SchemaFramed::new(&prefetch, JsonCodec).register::<Statement>("statements-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("statements-svc")
        .schema_prefetch(prefetch);

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

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

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

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Statement {
    account: i64,
    balance: i64,
}

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

impl MessageHeaders for Statement {
    type Contract = NoHeaders;
}

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

impl Handle<Account, Statement> for Issue {
    fn handle(
        &self,
        account: &Account,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Statement, HandlerOutcome>> {
        tracing::info!(account.id, "statement issued");
        ready(Ok(Statement {
            account: account.id,
            balance: 0,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    let prefetch = SchemaPrefetch::new(registry);
    // Decoding strips the envelope; encoding frames with the id of the type's subject.
    let codec = SchemaFramed::new(&prefetch, JsonCodec).register::<Statement>("statements-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("statements-svc")
        .schema_prefetch(prefetch);

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