krabka_client_streams/processor/
schema_serde.rs

1//! Bridges `krabka-schema-serde` typed serdes into the Streams `Serde<T>`
2//! boundary, and implements the membership `SchemaPrewarm` hook for
3//! `SchemaCache`. The `schema-serde` feature gates this module.
4
5use bytes::Bytes;
6use krabka_schema_serde::{
7    SchemaCache,
8    format::{SchemaDeserializer, SchemaSerializer, SchemaSubject},
9};
10
11use crate::{
12    error::StreamsClientError,
13    membership::SchemaPrewarm,
14    processor::serde::{Serde, SerdeAssociate, SerdeError, SerdeRole},
15};
16
17/// Wraps a schema-serde serializer and deserializer pair as a Streams
18/// `Serde<T>`.
19pub struct SchemaSerde<T, S> {
20    inner: S,
21    _marker: std::marker::PhantomData<fn() -> T>,
22}
23
24// Manual `Clone` (not derived) to avoid a spurious `T: Clone` bound;
25// `add_source`/`add_sink` require `Serde<_> + Clone`.
26impl<T, S: Clone> Clone for SchemaSerde<T, S> {
27    fn clone(&self) -> Self {
28        Self {
29            inner: self.inner.clone(),
30            _marker: std::marker::PhantomData,
31        }
32    }
33}
34
35impl<T, S> SchemaSerde<T, S> {
36    /// Wrap a schema-serde serde, such as an `AvroSerde<T>`.
37    pub fn new(inner: S) -> Self {
38        Self {
39            inner,
40            _marker: std::marker::PhantomData,
41        }
42    }
43}
44
45/// `Default` wraps the inner serde's `Default`, such as
46/// `AvroSerde<T>::default()`, which reads the process default registry. `T` can
47/// therefore declare a default schema serde with `DefaultSerde`, for use with
48/// `add_source` and `add_sink`.
49impl<T, S: Default> Default for SchemaSerde<T, S> {
50    fn default() -> Self {
51        Self::new(S::default())
52    }
53}
54
55impl<T: Send + Sync + 'static, S> SerdeAssociate for SchemaSerde<T, S> {
56    type Target = T;
57}
58
59impl<T, S> Serde<T> for SchemaSerde<T, S>
60where
61    T: Send + Sync + 'static,
62    S: SchemaSerializer<T> + SchemaDeserializer<T> + SchemaSubject,
63{
64    fn serialize(&self, topic: &str, value: &T) -> Bytes {
65        // The Streams sink path is infallible; a missing id means pre-warm was
66        // skipped — surface it loudly rather than writing a bad frame.
67        self.inner
68            .serialize(topic, value)
69            .expect("schema serialize failed (did membership prewarm run?)")
70    }
71
72    fn deserialize(&self, topic: &str, bytes: &[u8]) -> Result<T, SerdeError> {
73        self.inner
74            .deserialize(topic, bytes)
75            .map_err(|e| SerdeError(e.to_string()))
76    }
77
78    fn prepare(&self, topic: &str, _role: SerdeRole) {
79        // The serde was constructed with its key/value role; intern its subject
80        // for `topic` so membership pre-warm resolves the id.
81        self.inner.register_subject(topic);
82    }
83}
84
85#[async_trait::async_trait]
86impl SchemaPrewarm for SchemaCache {
87    async fn prewarm(&self) -> Result<(), StreamsClientError> {
88        // Call the inherent SchemaCache::prewarm (not this trait method).
89        // Inherent methods take priority in method resolution, so
90        // `SchemaCache::prewarm(self)` resolves to the inherent async method
91        // returning `Result<(), SchemaSerdeError>`, NOT this trait impl.
92        SchemaCache::prewarm(self)
93            .await
94            .map_err(|e| StreamsClientError::Runtime(e.to_string()))
95    }
96}
krabka_client_streams/processor/schema_serde.rs

krabka_client_streams/processor/
schema_serde.rs
