Protobuf transcode

On a broker with schema_registry, framed Protobuf arrives as JSON and leaves framed, with no generated code in the service.

Every subscription of that broker reads a delivery's schema id, fetches the schema and compiles its .proto source into a descriptor. It decodes the datum into a dynamic message and hands the handler JSON, so the handler is a plain serde one on the default codec. On the way out, the SchemaFrame publish layer turns the JSON into the subject's Protobuf message and frames it. Field names in the serde types match the .proto messages. The message written is the schema's first; SchemaFrame::message pins another.

This path is for a service that does not carry generated types. It costs a JSON hop and a dynamic message per delivery, and nothing decodes while the registry is unreachable. A service with generated types reads them on the byte lanes with decode_confluent instead. The two do not mix on one broker: a broker with schema_registry transcodes every subscription, so a generated type there would be handed JSON.

Example

The .proto source of the reply's subject and the serde types.

examples/kafka_protobuf_transcode.rs
/// What the `order-confirmations-value` subject holds: the registry keeps Protobuf as `.proto`
/// source.
const CONFIRMATION_PROTO: &str = r#"
syntax = "proto3";
package shop;

message Confirmation {
  int64 id = 1;
  bool accepted = 2;
}
"#;

// Plain serde structs whose field names match the `.proto` messages.
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: i64,
    item: String,
}

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

The handler is an ordinary JSON one.

Macros
Manual

examples/kafka_protobuf_transcode.rs
/// An ordinary handler on the default JSON codec.
#[subscriber("orders", reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: !order.item.is_empty(),
    }
}

examples/manual/kafka_protobuf_transcode.rs
/// An ordinary handler on the default JSON codec.
struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation {
            id: order.id,
            accepted: !order.item.is_empty(),
        }))
    }
}

The broker transcodes the deliveries, the layer frames the replies, and startup registers the reply's subject.

Macros
Manual

examples/kafka_protobuf_transcode.rs
fn app(registry: &SchemaRegistry) -> impl App {
    // Every subscription of this broker turns a framed delivery into JSON.
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        .schema_registry(registry.clone());
    let subjects = registry.clone();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Writes framed Protobuf where the destination's subject holds a Protobuf schema.
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            subjects
                .register(
                    "order-confirmations-value",
                    SchemaType::Protobuf,
                    CONFIRMATION_PROTO,
                )
                .await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(confirm);
        })
}

examples/manual/kafka_protobuf_transcode.rs
fn app(registry: &SchemaRegistry) -> impl App {
    // Every subscription of this broker turns a framed delivery into JSON.
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        .schema_registry(registry.clone());
    let subjects = registry.clone();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Writes framed Protobuf where the destination's subject holds a Protobuf schema.
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            subjects
                .register(
                    "order-confirmations-value",
                    SchemaType::Protobuf,
                    CONFIRMATION_PROTO,
                )
                .await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(subscriber("orders", Confirm).reply().build());
        })
}

Example​

Example