JSON transcode

schema_registry on the broker and the SchemaFrame layer convert framed payloads to and from plain JSON at the broker's edges.

The handler and its models stay on the default JSON codec and never see the envelope. On the way in, every subscription of the broker reads the envelope of a framed delivery, resolves the schema its id names and hands the handler a plain JSON document. A payload with no envelope passes through untouched. A lookup that fails leaves the delivery as it arrived, with a warning, and the handler's decode settles it. On the way out, the layer, added once with RustStream::publish_layer, resolves the subject of each publish's destination topic and frames the payload in that subject's format: a JSON subject keeps the document under the envelope, and an Avro or Protobuf subject is converted to its own format. The subject is {topic}-value unless SubjectStrategy or a pin names another.

The layer resolves a subject on its first publish to the topic and remembers the answer. A topic whose subject the registry does not hold publishes plain JSON, so registry-backed and plain topics share one app. A registry that cannot be reached, or a payload that does not fit the schema, fails the publish, and the publishing handler asks for a redelivery. The registry comes to the app as one SchemaRegistry value; the broker and the layer share it, and with it one cache. A producer owns its subjects, so the example registers the reply's subject at startup with register_json.

The transcode takes a hop through JSON per message. The codecs (SchemaFramed, AvroCodec) are the other path, and a broker uses one of the two.

Example

Macros
Manual

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[outgoing(name = "reservation-confirmations")]
struct Confirmation {
    id: i64,
    accepted: bool,
}

// An ordinary handler on the default JSON codec: the broker already took the envelope off.
#[subscriber("reservations", reply)]
async fn confirm(reservation: &Reservation) -> Confirmation {
    tracing::info!(reservation.id, "reservation confirmed");
    Confirmation {
        id: reservation.id,
        accepted: true,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // Incoming: the broker transcodes framed payloads to plain JSON.
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("reservations-svc")
        .schema_registry(registry.clone());

    RustStream::new(AppInfo::new("reservations", "0.1.0"))
        // Outgoing: the layer frames each publish by its topic's subject, `{topic}-value`.
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            registry
                .register_json::<Confirmation>("reservation-confirmations-value")
                .await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(confirm);
        })
}

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

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

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Confirmation {
    id: i64,
    accepted: bool,
}

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

impl MessageHeaders for Confirmation {
    type Contract = NoHeaders;
}

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

impl Handle<Reservation, Confirmation> for Confirm {
    // An ordinary handler on the default JSON codec: the broker already took the envelope off.
    fn handle(
        &self,
        reservation: &Reservation,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(reservation.id, "reservation confirmed");
        ready(Ok(Confirmation {
            id: reservation.id,
            accepted: true,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // Incoming: the broker transcodes framed payloads to plain JSON.
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("reservations-svc")
        .schema_registry(registry.clone());

    RustStream::new(AppInfo::new("reservations", "0.1.0"))
        // Outgoing: the layer frames each publish by its topic's subject, `{topic}-value`.
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            registry
                .register_json::<Confirmation>("reservation-confirmations-value")
                .await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(subscriber("reservations", Confirm).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