跳到主要内容
本页尚未翻译

下面显示的是英文原文。

RegistryClient

RegistryClient is the transport the registry speaks through; HttpRegistryClient is the default, and with_client supplies another.

SchemaRegistry is a facade over two parts: a RegistryClient that asks the registry, and a SchemaCache that remembers the answers. The codecs, the prefetch and the transcode are written against the facade, so a client of your own reaches all of them. It suits a different HTTP stack, a published client crate, a binding to a client in another language, or schemas shipped with the service.

A client answers four questions, each as a boxed future: a schema by id, a subject's latest version, registering a schema, and the id of a schema a subject already holds. latest answers None for a subject the registry does not hold, which the transcode reads as a plain topic. is_compatible has a default that answers None, "cannot tell", and the prefetch then skips its compatibility check. The client is shared as Arc<dyn RegistryClient>, so each round trip costs one virtual call and one allocation beside the network, and only a lookup the cache does not answer makes one. basic_auth, bearer_token and request_timeout configure the shipped HTTP client, and a supplied client carries its own credentials and deadlines. Errors are KafkaError::SchemaRegistry, with a message that names what was asked.

Example​

examples/kafka_registry_custom_client.rs
/// The schemas this service reads, by the id their records carry.
struct Catalog(HashMap<u32, Arc<RegisteredSchema>>);

impl Catalog {
    fn bundled() -> Self {
        let order =
            r#"{"type": "record", "name": "Order", "fields": [{"name": "id", "type": "long"}]}"#;
        Self(HashMap::from([(
            1,
            Arc::new(RegisteredSchema::new(1, SchemaType::Avro, order)),
        )]))
    }

    fn read_only() -> Result<u32, KafkaError> {
        Err(KafkaError::SchemaRegistry(
            "the bundled catalog is read-only".into(),
        ))
    }
}

impl RegistryClient for Catalog {
    fn schema_by_id(&self, id: u32) -> BoxFuture<'_, Result<Arc<RegisteredSchema>, KafkaError>> {
        let found = self.0.get(&id).cloned().ok_or_else(|| {
            KafkaError::SchemaRegistry(format!("schema id {id} is not in the catalog").into())
        });
        Box::pin(ready(found))
    }

    fn latest(
        &self,
        _subject: &str,
    ) -> BoxFuture<'_, Result<Option<Arc<RegisteredSchema>>, KafkaError>> {
        // No subjects: every topic this service publishes to is a plain one.
        Box::pin(ready(Ok(None)))
    }

    fn register(
        &self,
        _subject: &str,
        _schema_type: SchemaType,
        _definition: String,
    ) -> BoxFuture<'_, Result<u32, KafkaError>> {
        Box::pin(ready(Self::read_only()))
    }

    fn lookup_id(
        &self,
        _subject: &str,
        _schema_type: SchemaType,
        _definition: String,
    ) -> BoxFuture<'_, Result<u32, KafkaError>> {
        Box::pin(ready(Self::read_only()))
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Order {
    id: i64,
}

#[subscriber("orders")]
async fn take(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // Everything above the client (the transcode, the codecs, the cache) works unchanged.
    let registry = SchemaRegistry::with_client(Arc::new(Catalog::bundled()));
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        .schema_registry(registry);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(take);
    })
}