Third-party clients

Another registry stack does the schema work and the byte lanes carry its framed bytes, so no type crosses between the two.

A service that already uses a registry crate, such as schema_registry_converter, keeps it. The handler takes the delivery as a #[derive(Deserialized)] view of its bytes and hands them to the converter, which resolves the schema the envelope names through its own client and cache. The reply is the framed buffer the converter returns, in a #[derive(Serialized)] type, and the lane publishes it as it is. No codec runs on either side.

Such a crate pins its own apache-avro and prost, and those versions may differ from this crate's. On this path only bytes pass between the two, so the versions never meet. The converter is application state, built in on_startup and read with State. The broker carries neither schema_registry nor schema_prefetch: the schemas are the converter's business. The same shape fits any client that produces bytes, a binding to a client in another language included.

Example

Macros
Manual

examples/kafka_third_party_registry.rs
/// The delivery as it arrived: a view of the bytes, handed straight to the converter.
#[derive(Deserialized)]
struct Framed<'a>(&'a [u8]);

/// What the converter produced; the lane sends the bytes as they are.
#[derive(Serialized, Outgoing)]
#[outgoing(name = "rental-confirmations")]
struct Encoded(Vec<u8>);

#[derive(Serialize)]
struct Confirmation {
    id: i64,
    accepted: bool,
}

/// The service's own registry stack, held as application state.
#[derive(Clone)]
struct Converter {
    decoder: Arc<AvroDecoder<'static>>,
    encoder: Arc<AvroEncoder<'static>>,
}

#[derive(FromRef)]
struct Rentals {
    converter: Converter,
}

#[subscriber("rentals", reply)]
async fn confirm(
    framed: &Framed<'_>,
    State(converter): State<Converter>,
) -> Result<Encoded, HandlerOutcome> {
    // The converter resolves the schema the envelope names through its own client and cache.
    let decoded = converter
        .decoder
        .decode(Some(framed.0))
        .await
        .map_err(|error| {
            tracing::warn!(topic = "rentals", %error, "the converter could not decode a record");
            HandlerOutcome::drop()
        })?;
    let AvroValue::Record(fields) = &decoded.value else {
        return Err(HandlerOutcome::drop());
    };
    let id = fields
        .iter()
        .find_map(|(name, value)| match (name.as_str(), value) {
            ("id", AvroValue::Long(id)) => Some(*id),
            _ => None,
        })
        .ok_or(HandlerOutcome::drop())?;
    tracing::info!(id, "rental confirmed");

    let bytes = converter
        .encoder
        .encode_struct(
            Confirmation { id, accepted: true },
            &SubjectNameStrategy::TopicNameStrategy("rental-confirmations".to_owned(), false),
        )
        .await
        .map_err(|error| {
            tracing::warn!(topic = "rental-confirmations", %error, "the converter could not encode");
            HandlerOutcome::retry()
        })?;
    Ok(Encoded(bytes))
}

fn app(registry_url: String) -> impl App {
    // A plain broker: the schemas are the converter's business.
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("rentals-svc");
    RustStream::new(AppInfo::new("rentals", "0.1.0"))
        .on_startup(async move |()| {
            let settings = SrSettings::new(registry_url);
            Ok::<_, Infallible>(Rentals {
                converter: Converter {
                    decoder: Arc::new(AvroDecoder::new(settings.clone())),
                    encoder: Arc::new(AvroEncoder::new(settings)),
                },
            })
        })
        .with_broker(broker, |b| {
            b.include(confirm);
        })
}

#[ruststream::app]
fn service() -> impl App {
    app("http://localhost:8081".to_owned())
}

examples/manual/kafka_third_party_registry.rs
/// The delivery as it arrived: a view of the bytes, handed straight to the converter.
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>>;
}

/// What the converter produced; the lane sends the bytes as they are.
struct Encoded(Vec<u8>);

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

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

impl MessageWire for Encoded {
    type Wire = SerializedWire;
}

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

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

impl MessageHeaders for Encoded {
    type Contract = NoHeaders;
}

#[derive(Serialize)]
struct Confirmation {
    id: i64,
    accepted: bool,
}

/// The service's own registry stack, held as application state.
#[derive(Clone)]
struct Converter {
    decoder: Arc<AvroDecoder<'static>>,
    encoder: Arc<AvroEncoder<'static>>,
}

/// The application state; the handler's state axis names it.
struct Rentals {
    converter: Converter,
}

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

impl Handle<Framed<'_>, Encoded, (), (), Rentals> for Confirm {
    async fn handle(
        &self,
        framed: &Framed<'_>,
        _outs: &(),
        ctx: &mut Context<'_, (), Rentals>,
    ) -> Result<Encoded, HandlerOutcome> {
        let converter = &ctx.state().converter;
        // The converter resolves the schema the envelope names through its own client and cache.
        let decoded = converter
        .decoder
        .decode(Some(framed.0))
        .await
        .map_err(|error| {
            tracing::warn!(topic = "rentals", %error, "the converter could not decode a record");
            HandlerOutcome::drop()
        })?;
        let AvroValue::Record(fields) = &decoded.value else {
            return Err(HandlerOutcome::drop());
        };
        let id = fields
            .iter()
            .find_map(|(name, value)| match (name.as_str(), value) {
                ("id", AvroValue::Long(id)) => Some(*id),
                _ => None,
            })
            .ok_or(HandlerOutcome::drop())?;
        tracing::info!(id, "rental confirmed");

        let bytes = converter
        .encoder
        .encode_struct(
            Confirmation { id, accepted: true },
            &SubjectNameStrategy::TopicNameStrategy("rental-confirmations".to_owned(), false),
        )
        .await
        .map_err(|error| {
            tracing::warn!(topic = "rental-confirmations", %error, "the converter could not encode");
            HandlerOutcome::retry()
        })?;
        Ok(Encoded(bytes))
    }
}

fn app(registry_url: String) -> impl App {
    // A plain broker: the schemas are the converter's business.
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("rentals-svc");
    RustStream::new(AppInfo::new("rentals", "0.1.0"))
        .on_startup(async move |()| {
            let settings = SrSettings::new(registry_url);
            Ok::<_, Infallible>(Rentals {
                converter: Converter {
                    decoder: Arc::new(AvroDecoder::new(settings.clone())),
                    encoder: Arc::new(AvroEncoder::new(settings)),
                },
            })
        })
        .with_broker(broker, |b| {
            b.include(subscriber("rentals", Confirm).reply().build());
        })
}

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

Example​

Example