SubjectStrategy

SubjectStrategy names a subject from the topic, the record name or both, and SchemaFrame::subject pins the subject of one topic.

The strategies are Confluent's three. TopicName, the default, is {topic}-value: one value schema per topic. RecordName is the record's fully qualified name: one subject per record type, shared by every topic that carries it. TopicRecordName is {topic}-{record}: several record types on one topic, each with its own subject. subject(topic, record) builds the name.

The SchemaFrame layer derives a topic's subject with its strategy, TopicName unless subject_strategy names another. The publish path knows the topic and not the record name, so a topic under a record strategy gets its subject pinned with subject(topic, subject), which overrides the strategy for that topic. The example pins the subject that RecordName gives the record. The Protobuf publish policy and its layer take the same three settings.

Example

Macros
Manual

examples/kafka_registry_subject.rs
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Return {
    order: i64,
    amount: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[outgoing(name = "refunds")]
struct Refund {
    order: i64,
    amount: i64,
}

#[subscriber("returns", reply)]
async fn reimburse(item: &Return) -> Refund {
    tracing::info!(item.order, item.amount, "refund issued");
    Refund {
        order: item.order,
        amount: item.amount,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // `billing.Refund`: the record's own subject, whichever topic carries it.
    let refunds = SubjectStrategy::RecordName.subject("refunds", "billing.Refund");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("refunds-svc")
        .schema_registry(registry.clone());
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .publish_layer(SchemaFrame::new(registry).subject("refunds", refunds))
        .with_broker(broker, |b| {
            b.include(reimburse);
        })
}

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

examples/manual/kafka_registry_subject.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Return {
    order: i64,
    amount: i64,
}

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

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

impl MessageHeaders for Refund {
    type Contract = NoHeaders;
}

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

impl Handle<Return, Refund> for Reimburse {
    fn handle(
        &self,
        item: &Return,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Refund, HandlerOutcome>> {
        tracing::info!(item.order, item.amount, "refund issued");
        ready(Ok(Refund {
            order: item.order,
            amount: item.amount,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // `billing.Refund`: the record's own subject, whichever topic carries it.
    let refunds = SubjectStrategy::RecordName.subject("refunds", "billing.Refund");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("refunds-svc")
        .schema_registry(registry.clone());
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .publish_layer(SchemaFrame::new(registry).subject("refunds", refunds))
        .with_broker(broker, |b| {
            b.include(subscriber("returns", Reimburse).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