Struct SchemaFrame 

pub struct SchemaFrame { /* private fields */ }

Publish middleware framing outgoing JSON in the Confluent wire format - the publish-side half of the registry integration, added app-wide with RustStream::publish_layer.

For every publish it resolves the destination topic’s value subject (lazily: the shared cache first, the registry once on a miss) and frames the codec’s plain-JSON payload by the subject’s registered flavor: JSON keeps its bytes under the envelope, Avro and Protobuf transcode against the schema (with their cargo features). A topic with no registered subject publishes untouched - mixed registry/plain topologies need no configuration. The miss is cached (and logged) once per subject, so plain topics pay no per-publish registry round-trip; a subject registered later is picked up after a restart or an explicit SchemaRegistry::warm. A registry outage or a payload that does not fit the schema fails the publish, so a publishing handler nacks and the delivery is retried rather than a mis-framed record reaching the topic.

Subjects follow SubjectStrategy::TopicName by default; subject_strategy switches the naming, and subject pins a topic’s subject explicitly (the RecordName layouts, where the topic alone cannot name the subject).

Examples

use ruststream_rdkafka::prelude::*;
use ruststream_rdkafka::{SchemaFrame, SchemaRegistry};
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
}

/// Returns plain JSON; the layer frames it for `confirmations-value` on the way out.
#[subscriber("orders", reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

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

Implementations

impl SchemaFrame

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). The RecordName strategies need the record’s name, which the publish path does not know; pin those subjects per topic with subject instead.

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 Protobuf message topic’s payloads serialize as, by fully qualified name (package included) - for subjects whose .proto schema declares several messages. Without it a Protobuf subject serializes as the schema’s first top-level message.

Trait Implementations

impl Clone for SchemaFrame

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 SchemaFrame

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

Formats the value using the given formatter.

impl PublishLayer for SchemaFrame

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

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