framed

KafkaPublish::framed puts the Confluent envelope on every message its policy publishes, with the destination subject's schema id.

It returns a KafkaFramedPublish, a policy like any other, so it goes where a policy goes: .out_reply(..) or an Out slot. Its live form is a KafkaFramedPublisher. The subject comes from the destination topic, {topic}-value by default; subject_strategy and subject choose another. The index path names the schema's first message, which Confluent writes as one zero byte. message pins another message of the schema by its full name, for a topic whose type is not declared first. KafkaFramedPublish::over wraps a configured KafkaPublish instead of the default one.

The subject is resolved on the first publish to each destination and cached. That is the first moment the destination is known: it comes from the reply clause or the slot publish, not from the policy. So a subject the registry does not hold, or one holding another format, fails that publish with the subject and the format, and the handler's failure policy settles the delivery. A payload that already carries an envelope is left as it is, so the policy and ProtobufFrame together frame a message once.

Example

The receipt declares no destination, so the mount site's reply("order-receipts") names it, and the policy frames it with the id of order-receipts-value.

Macros
Manual

examples/kafka_protobuf_framed.rs
// The receipt names no destination, so the mount site's `reply(..)` does.
#[derive(Clone, PartialEq, prost::Message, Serialized, Outgoing)]
#[wire(encode = prost::Message::encode)]
struct Receipt {
    #[prost(int64, tag = "1")]
    order_id: i64,
    #[prost(bool, tag = "2")]
    accepted: bool,
}

#[subscriber("orders", reply("order-receipts"))]
async fn receipt(order: &Order) -> Receipt {
    Receipt {
        order_id: order.id,
        accepted: !order.item.is_empty(),
    }
}

examples/manual/kafka_protobuf_framed.rs
// The receipt names no destination, so the mount site's `reply(..)` does.
#[derive(Clone, PartialEq, prost::Message, Serialized, Outgoing)]
#[wire(encode = prost::Message::encode)]
struct Receipt {
    #[prost(int64, tag = "1")]
    order_id: i64,
    #[prost(bool, tag = "2")]
    accepted: bool,
}

struct Confirm;

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

Macros
Manual

examples/kafka_protobuf_framed.rs
fn app(registry: &SchemaRegistry) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            // The subject is the destination topic's, `order-receipts-value`, resolved on the
            // first reply and cached.
            b.include(receipt).out_reply(Publish::framed(registry));
        },
    )
}

examples/manual/kafka_protobuf_framed.rs
fn app(registry: &SchemaRegistry) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            let orders = subscriber("orders", Confirm).reply().to("order-receipts");
            // The subject is the destination topic's, `order-receipts-value`, resolved on the
            // first reply and cached.
            b.include(orders.build())
                .out_reply(Publish::framed(registry));
        },
    )
}

Example​

Example