Protobuf

A prost message rides the byte lanes as itself, and decode_confluent steps over the Confluent envelope with no registry.

The protobuf feature adds it, with the protobuf module. A generated message is not a serde type, so it carries its own bytes: an incoming type derives Deserialized and names #[wire(decode = decode_confluent)], an outgoing one derives Serialized and names #[wire(encode = prost::Message::encode)]. The handler then takes and returns generated messages, and its signature says nothing about the wire.

A registry-backed producer writes the Confluent envelope in front of the message: a zero byte, the four-byte schema id, and the index path of the message within its schema. decode_confluent parses all three and reads the rest into the reader's own type. No lookup is needed, because Protobuf fields are addressed by tag: an unknown field is kept as unknown, and a missing one takes its default. A payload with no envelope is read as a bare message, so a topic where only some producers frame still decodes. What the reader does not check is which schema version a delivery claims.

Writing the envelope needs the schema id of the destination's subject, which a value cannot look up. The publish side adds it where the destination is known: framed at a mount site, or ProtobufFrame for the whole app. A publish without either sends the bare message. A subject is registered from .proto text through SchemaRegistry::register with SchemaType::Protobuf. The registry paths compile that text with the well-known types available, so a schema imports from them and from nothing else.

Example

The generated types, with their lanes.

examples/kafka_protobuf_plain.rs
// What `prost-build` emits, plus the lane derives: the order decodes past the envelope, and the
// confirmation encodes itself through `prost`.
#[derive(Clone, PartialEq, prost::Message, Deserialized)]
#[wire(decode = decode_confluent)]
struct Order {
    #[prost(int64, tag = "1")]
    id: i64,
    #[prost(string, tag = "2")]
    item: String,
}

#[derive(Clone, PartialEq, prost::Message, Serialized, Outgoing)]
#[wire(encode = prost::Message::encode)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    #[prost(int64, tag = "1")]
    id: i64,
    #[prost(bool, tag = "2")]
    accepted: bool,
}

The handler reads a framed or a bare order and returns the confirmation, which leaves as a bare message.

Macros
Manual

examples/kafka_protobuf_plain.rs
/// Confirms an order that names an item.
#[subscriber("orders", reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order = order.id, item = %order.item, "confirming");
    Confirmation {
        id: order.id,
        accepted: !order.item.is_empty(),
    }
}

examples/manual/kafka_protobuf_plain.rs
/// Confirms an order that names an item.
struct Confirm;

// Both positions are generated messages that carry their own bytes, so neither names a codec.
impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order = order.id, item = %order.item, "confirming");
        ready(Ok(Confirmation {
            id: order.id,
            accepted: !order.item.is_empty(),
        }))
    }
}

Macros
Manual

examples/kafka_protobuf_plain.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            b.include(confirm);
        },
    )
}

examples/manual/kafka_protobuf_plain.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            b.include(subscriber("orders", Confirm).reply().build());
        },
    )
}

Example​

Example