Envelope

parse_envelope and encode_envelope read and write the Confluent wire format: a zero byte, the schema id, the datum.

The schema id takes four bytes, big-endian. parse_envelope splits a payload into the id and a view of the datum, with no copy. It returns None for a payload that does not open with the zero byte or is shorter than five bytes, which reads as "not framed". encode_envelope writes the five bytes and the datum into a new buffer. Neither function reaches a registry.

A handler that works on the framed bytes takes them through the byte lanes: a #[derive(Deserialized)] view of the delivery in, a #[derive(Serialized)] buffer out. No codec runs on either side, so the bytes travel as they are. The example moves records between two registries and rewrites the id each record names.

Example

Macros
Manual

examples/kafka_registry_envelope.rs
/// A record as it arrived: a view of its bytes, envelope included.
#[derive(Deserialized)]
struct Framed<'a>(&'a [u8]);

/// The same datum under the id the new registry gave its schema.
#[derive(Serialized, Outgoing)]
#[outgoing(name = "orders-migrated")]
struct Reframed(Vec<u8>);

/// Each schema id of the old registry, and the id the new registry gave the same schema.
const MIGRATED_IDS: &[(u32, u32)] = &[(7, 107), (8, 108)];

#[subscriber("orders", reply)]
async fn reframe(record: &Framed<'_>) -> Result<Reframed, HandlerOutcome> {
    let Some((old, datum)) = parse_envelope(record.0) else {
        tracing::warn!(topic = "orders", "a record without the Confluent envelope");
        return Err(HandlerOutcome::drop());
    };
    let Some(&(_, new)) = MIGRATED_IDS.iter().find(|(from, _)| *from == old) else {
        tracing::warn!(
            topic = "orders",
            schema_id = old,
            "a schema id with no migration"
        );
        return Err(HandlerOutcome::drop());
    };
    tracing::info!(old, new, "record reframed");
    Ok(Reframed(encode_envelope(new, datum)))
}

#[ruststream::app]
fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("registry-migration");
    RustStream::new(AppInfo::new("registry-migration", "0.1.0")).with_broker(broker, |b| {
        b.include(reframe);
    })
}

examples/manual/kafka_registry_envelope.rs
/// A record as it arrived: a view of its bytes, envelope included.
struct Framed<'a>(&'a [u8]);

// `#[derive(Deserialized)]` over a borrowed byte slice: the view is the payload itself.
impl Deserialized for Framed<'_> {
    type Output<'a> = Framed<'a>;
    type Error = Infallible;

    fn from_payload(payload: &[u8]) -> Result<Framed<'_>, Infallible> {
        Ok(Framed(payload))
    }
}

impl Input for Framed<'_> {
    type Axis = SoloDeserialized<Framed<'static>>;
}

/// The same datum under the id the new registry gave its schema.
struct Reframed(Vec<u8>);

// `#[derive(Serialized)]`: the bytes are the payload, so no codec runs on them.
impl Serialized for Reframed {
    type Error = Infallible;

    fn wire_bytes(&self, _buf: &mut BytesMut) -> Result<WireBytes<'_>, Infallible> {
        Ok(WireBytes::Own(&self.0))
    }
}

impl MessageWire for Reframed {
    type Wire = SerializedWire;
}

impl ReplyShape for Reframed {
    type Body = Self;
    type Headers = ();
    type Wire = SerializedReply;
}

// `#[derive(Outgoing)]` with `#[outgoing(name = "orders-migrated")]`.
impl OutgoingDestination for Reframed {
    type Form = FixedName;
    const DESTINATION: &'static str = "orders-migrated";
}

impl MessageHeaders for Reframed {
    type Contract = NoHeaders;
}

/// Each schema id of the old registry, and the id the new registry gave the same schema.
const MIGRATED_IDS: &[(u32, u32)] = &[(7, 107), (8, 108)];

/// The definition `#[subscriber("orders", reply)]` would have minted.
struct Reframe;

impl Handle<Framed<'_>, Reframed> for Reframe {
    fn handle(
        &self,
        record: &Framed<'_>,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Reframed, HandlerOutcome>> {
        ready(reframe(record.0))
    }
}

/// The body of the handler: it awaits nothing, so the future above is ready at once.
fn reframe(record: &[u8]) -> Result<Reframed, HandlerOutcome> {
    let Some((old, datum)) = parse_envelope(record) else {
        tracing::warn!(topic = "orders", "a record without the Confluent envelope");
        return Err(HandlerOutcome::drop());
    };
    let Some(&(_, new)) = MIGRATED_IDS.iter().find(|(from, _)| *from == old) else {
        tracing::warn!(
            topic = "orders",
            schema_id = old,
            "a schema id with no migration"
        );
        return Err(HandlerOutcome::drop());
    };
    tracing::info!(old, new, "record reframed");
    Ok(Reframed(encode_envelope(new, datum)))
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("registry-migration");
    RustStream::new(AppInfo::new("registry-migration", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Reframe).reply().build());
    })
}

// What `#[ruststream::app]` writes: the command line on a multi-threaded Tokio runtime.
fn main() -> ExitCode {
    run_main(app)
}

Example​

Example