ProtobufFrame

ProtobufFrame, added with publish_layer, frames every Protobuf publish of the app that leaves through a slot.

ProtobufFrame::new takes the registry, and the app adds the layer once, so no mount site names the framing. A publish is framed when its destination's subject holds a Protobuf schema. Every other publish passes as it is: a topic with no subject, and a subject holding Avro or JSON. So one app carries Protobuf, Avro and JSON topics with the layer installed. A payload that already carries an envelope also passes, so a message is never framed twice. subject_strategy, subject and message choose the subject and the message as they do on framed.

The layer sees what leaves through an Out slot and through a publisher the runtime pairs. A reply returned from a reply(..) handler goes straight to its paired publisher, so it is framed by naming framed at its mount site. The layer is an alternative to the transcode's SchemaFrame: this one expects the payload to be the Protobuf datum already, that one expects JSON.

Example

Every order writes an audit record through a slot, and the layer frames it with the id of order-audit-value.

Macros
Manual

examples/kafka_protobuf_frame.rs
#[derive(OutSlot)]
#[publishes(Audit)]
struct Journal;

/// Records every order in the audit journal; the app's layer frames the record.
#[subscriber("orders")]
async fn audit(order: &Order, Out(journal): Out<impl Publisher, Journal>) -> HandlerOutcome {
    let record = Audit { order_id: order.id };
    if journal.message(&record).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_protobuf_frame.rs
struct Journal;

impl OutSlot for Journal {
    const NAME: &'static str = "Journal";
    type Destination = Reads;
}

impl PublishedThrough<Journal> for Audit {}

/// Records every order in the audit journal; the app's layer frames the record.
struct AuditOrder;

impl<J> Handle<Order, (), Outs<(J,)>> for AuditOrder
where
    J: OutEntry<Journal, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(J,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let record = Audit { order_id: order.id };
        if outs.get(Journal).message(&record).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Macros
Manual

examples/kafka_protobuf_frame.rs
fn app(registry: &SchemaRegistry) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Frames every publish whose destination subject holds a Protobuf schema.
        .publish_layer(ProtobufFrame::new(registry.clone()))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
            |b| {
                b.include(audit).out(Journal, Publish::default()).build();
            },
        )
}

examples/manual/kafka_protobuf_frame.rs
fn app(registry: &SchemaRegistry) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Frames every publish whose destination subject holds a Protobuf schema.
        .publish_layer(ProtobufFrame::new(registry.clone()))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
            |b| {
                b.include(subscriber("orders", AuditOrder).build())
                    .out(Journal, Publish::default())
                    .build();
            },
        )
}

Example​

Example