Struct ProtobufFrame 

pub struct ProtobufFrame { /* private fields */ }

Publish middleware putting the Confluent envelope on a prost message’s own bytes.

It is the encode-side half of the pair decode_confluent completes on the read side: with both in place a Protobuf handler returns the message and nothing else. Added app-wide with RustStream::publish_layer.

The envelope needs a schema id, and a value cannot fetch one: Serialized::wire_bytes is synchronous and has only &self to work with. The publish path can, because it knows the destination topic, which is exactly what names the subject - so the value writes bare Protobuf through the core’s own #[wire(encode = ::prost::Message::encode)] and this layer puts the id and the message-index path in front of it. That division is why nothing here needs a process-wide registry singleton keyed by message type, which is the other way a value could have reached an id.

What it touches, and what it leaves alone

A publish is framed only when its topic’s subject is registered and holds a Protobuf schema. Everything else passes through untouched: a topic with no subject (mixed registry/plain topologies need no configuration), and a topic whose subject is Avro or JSON, whose payload some other path already framed. That is what lets one app carry an Avro codec, a JSON codec and Protobuf at once with this layer installed app-wide.

A payload that already carries an envelope also passes through, so a message that came in framed and goes out again is not framed twice. The check is exact rather than a guess: a bare prost message begins with a field tag, whose field number is at least 1, so its first byte is never the zero magic byte.

It is an alternative to SchemaFrame rather than a companion. That layer’s contract is “the payload is a JSON document, transcode it to the subject’s flavor”; this one’s is “the payload is already the subject’s datum, put the envelope on”. Install one or the other.

The layer does not see a reply(..) reply

A handler’s outgoing message reaches this layer when it leaves through an Out slot, and not when it is returned as the reply of a reply(..) mount. That is the core’s own division rather than a gap here: a byte-for-byte reply goes straight to its paired publisher, deliberately, so a value that owns its bytes leaves exactly as it wrote them.

So there are two shapes, and which one to reach for is a question of where the framing should be declared. A handler that replies takes KafkaPublish::framed at its mount site, which frames on the way through the reply’s own publisher and keeps the body a plain async fn confirm(order: &Order) -> Confirmation. This layer is for setting framing once for a whole app, where publishes leave through slots and through publishers the mount sites never name.

The two compose rather than conflict: whichever runs first frames the payload, and the other sees an envelope already there and leaves it alone.

Examples

use ruststream_rdkafka::prelude::*;
use ruststream_rdkafka::{ProtobufFrame, SchemaRegistry};

#[derive(Clone, PartialEq, prost::Message, Deserialized)]
#[wire(decode = ruststream_rdkafka::protobuf::decode_confluent)]
struct Order {
    #[prost(int64, tag = "1")]
    id: i64,
}

#[derive(Clone, PartialEq, prost::Message, Serialized, Outgoing)]
#[wire(encode = ::prost::Message::encode)]
#[outgoing(name = "audit")]
struct Audit {
    #[prost(int64, tag = "1")]
    order_id: i64,
}

#[derive(OutSlot)]
#[publishes(Audit)]
struct Journal;

/// The audit record leaves through a slot, so the app-wide layer frames it.
#[subscriber("orders")]
async fn place(order: &Order, Out(journal): Out<impl Publisher, Journal>) -> HandlerOutcome {
    if journal.message(&Audit { order_id: order.id }).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let registry = SchemaRegistry::new("http://localhost:8081");
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(ProtobufFrame::new(registry))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
            |b| {
                b.include(place).out(Journal, Publish::default()).build();
            },
        )
}

Implementations

impl ProtobufFrame

pub fn new(registry: SchemaRegistry) -> Self

Builds the framing middleware over registry. Subjects default to the Confluent TopicName strategy ({topic}-value).

pub fn subject_strategy(self, strategy: SubjectStrategy) -> Self

How destination topics map onto subjects (default: SubjectStrategy::TopicName).

pub fn subject( self, topic: impl Into<String>, subject: impl Into<String>, ) -> Self

Pins topic’s subject explicitly, overriding the strategy.

pub fn message( self, topic: impl Into<String>, message: impl Into<String>, ) -> Self

Pins the message topic’s payloads are, by fully qualified name (package included).

The message-index path says which message of the schema was written, so it has to be the one the publishing type actually is. Without this pin the layer takes the schema’s first top-level message, which is the common case and the one Confluent optimises to a single zero byte - so a single-message .proto, and a multi-message one whose published type is declared first, need no pin. Anything else does: naming the wrong message puts a mis-addressed path on the wire, which consumers report as a decode failure.

Trait Implementations

impl Clone for ProtobufFrame

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ProtobufFrame

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl PublishLayer for ProtobufFrame

async fn on_publish<'a, N: PublishPipeline, P: Publisher>( &'a self, out: &'a mut Outgoing<'a>, next: PublishNext<'a, N, P>, ) -> Result<(), Box<dyn Error + Send + Sync>>

Handle the outgoing message, calling next to continue the pipeline.