laser_sdk/
typed.rs

1use crate::cursor::Cursor;
2use crate::error::LaserError;
3use crate::message::Message;
4use crate::stream::{
5    BatchPublishRequest, Cbor, Codec, ContentType, Decoder, Json, PublishRequest, Topic,
6};
7use crate::types::MessageId;
8use serde::Serialize;
9use serde::de::DeserializeOwned;
10use std::collections::{BTreeMap, VecDeque};
11use std::marker::PhantomData;
12
13impl Topic {
14    /// The typed handle over this topic in the JSON serde form: no registry,
15    /// the body type's `Serialize`/`Deserialize` is the contract. Publishes
16    /// stamp `agdx.ct=json`, [`records`](TypedTopic::records) decodes each
17    /// payload back into `T`.
18    pub fn json<T>(&self) -> TypedTopic<T> {
19        TypedTopic::new(self.clone(), Form::Json)
20    }
21
22    /// The typed handle in the CBOR serde form: same contract as
23    /// [`json`](Self::json) with a binary self-describing body and
24    /// `agdx.ct=cbor`.
25    pub fn cbor<T>(&self) -> TypedTopic<T> {
26        TypedTopic::new(self.clone(), Form::Cbor)
27    }
28
29    /// The typed handle bound to a registered writer schema: resolves the
30    /// definition from LaserData Cloud's registry and compiles it ONCE for the
31    /// handle's lifetime, so every publish validates against it client-side
32    /// before any byte leaves the process and stamps `agdx.ct` + `agdx.sid`.
33    /// A body that stops matching the schema fails at the producer with the
34    /// schema's error, not downstream in the projector. Feature
35    /// `schema-codecs`, and the registry is managed, so this form needs
36    /// LaserData Cloud.
37    #[cfg(feature = "schema-codecs")]
38    pub async fn schema<T>(&self, schema_id: u32) -> Result<TypedTopic<T>, LaserError> {
39        let info = self
40            .laser()
41            .schemas()
42            .get(schema_id)
43            .await?
44            .ok_or_else(|| LaserError::Invalid(format!("schema {schema_id} is not registered")))?;
45        let compiled = crate::schema_codecs::CompiledSchema::compile(&info.schema)?;
46        Ok(TypedTopic::new(
47            self.clone(),
48            Form::Schema {
49                id: schema_id,
50                compiled,
51            },
52        ))
53    }
54}
55
56/// One topic seen through one body type: encode, validate, and stamp on the
57/// way in, decode with the log position attached on the way out. Build it with
58/// [`Topic::json`], [`Topic::cbor`], or `Topic::schema`. The handle is a
59/// plain wrapper over [`Topic`], so it is free to construct and clone, and the
60/// raw verbs on the untyped handle stay one accessor away.
61#[derive(Clone)]
62pub struct TypedTopic<T> {
63    topic: Topic,
64    form: Form,
65    body: PhantomData<fn() -> T>,
66}
67
68impl<T> TypedTopic<T> {
69    /// Encode `body` under the handle's form and open the publish builder with
70    /// the payload, `agdx.ct`, and (for the schema-bound form) `agdx.sid`
71    /// already stamped. Chain `.index(..)`, `.partition_key(..)`, or any other
72    /// record option, then `.send()`. Encoding and schema validation happen
73    /// HERE, so a body the schema rejects never reaches the wire.
74    pub fn publish(&self, body: &T) -> Result<PublishRequest<'_>, LaserError>
75    where
76        T: Serialize,
77    {
78        let (payload, content_type, schema_id) = self.form.encode(body)?;
79        let mut request = self.topic.publish().raw_bytes(payload, content_type);
80        if let Some(id) = schema_id {
81            request = request.schema_id(id);
82        }
83        Ok(request)
84    }
85
86    /// Encode every body in `bodies` under the handle's form and open the
87    /// batch publish builder with the payloads, `agdx.ct`, and (for the
88    /// schema-bound form) `agdx.sid` already stamped. Chain batch options,
89    /// then `.send()`. A body that does not encode fails here, so a partly
90    /// valid batch never reaches the wire.
91    pub fn publish_batch<'b>(
92        &self,
93        bodies: impl IntoIterator<Item = &'b T>,
94    ) -> Result<BatchPublishRequest<'_>, LaserError>
95    where
96        T: Serialize + 'b,
97    {
98        let mut request = self.topic.publish_batch();
99        for body in bodies {
100            let (payload, content_type, schema_id) = self.form.encode(body)?;
101            request = request.add_raw_bytes(payload, content_type);
102            if let Some(id) = schema_id {
103                request = request.schema_id(id);
104            }
105        }
106        Ok(request)
107    }
108
109    /// The typed reader over this topic: a resumable [`Cursor`] under the
110    /// caller-chosen identity `reader_name`, decoding each record into `T` as
111    /// it is drained. This is not Apache Iggy consumer-group delivery. Own the
112    /// offsets exactly like the raw cursor: persist
113    /// [`offsets`](TypedRecords::offsets) and resume with
114    /// [`from_offsets`](TypedRecords::from_offsets), so the read is bounded by
115    /// construction and never a silent rescan from zero.
116    pub fn records(&self, reader_name: &str) -> Result<TypedRecords<T>, LaserError>
117    where
118        T: DeserializeOwned,
119    {
120        let cursor = self.topic.replay()?.consumer_named(reader_name)?;
121        Ok(TypedRecords {
122            cursor,
123            buffered: VecDeque::new(),
124            form: self.form.clone(),
125            body: PhantomData,
126        })
127    }
128
129    /// The untyped handle underneath, for the verbs the typed form does not
130    /// wrap (raw send, batches, the substrate builders).
131    pub fn topic(&self) -> &Topic {
132        &self.topic
133    }
134
135    fn new(topic: Topic, form: Form) -> Self {
136        Self {
137            topic,
138            form,
139            body: PhantomData,
140        }
141    }
142}
143
144/// The typed consume side of a [`TypedTopic`]: each [`next`](Self::next)
145/// yields the next record decoded as `T`, or the position-carrying
146/// [`TypedDecodeError`] for a record that does not decode (the reader moves
147/// past it, so one bad record never wedges the stream). `None` means caught
148/// up: everything appended so far has been yielded, call again later to see
149/// new records.
150///
151/// The reliable consumer path composes instead of duplicating this: an
152/// `Agent` handler decodes inside the handler with the
153/// same error shape, and an undecodable record routes to the existing
154/// dead-letter policy.
155pub struct TypedRecords<T> {
156    cursor: Cursor,
157    buffered: VecDeque<Message>,
158    form: Form,
159    body: PhantomData<fn() -> T>,
160}
161
162impl<T: DeserializeOwned> TypedRecords<T> {
163    /// Resume from previously persisted per-partition offsets, exactly what an
164    /// earlier [`offsets`](Self::offsets) returned.
165    #[must_use]
166    pub fn from_offsets(mut self, offsets: Vec<u64>) -> Self {
167        self.cursor = self.cursor.from_offsets(offsets);
168        self
169    }
170
171    /// Read at most `batch` messages per partition per underlying poll.
172    #[must_use]
173    pub fn batch(mut self, batch: u32) -> Self {
174        self.cursor = self.cursor.batch(batch);
175        self
176    }
177
178    /// The next offset to read on each partition. Persist this to resume later
179    /// with [`from_offsets`](Self::from_offsets).
180    pub fn offsets(&self) -> &[u64] {
181        self.cursor.offsets()
182    }
183
184    /// The next record, decoded. `None` when the reader is caught up. A record
185    /// that does not decode yields the error with its log position (taken from
186    /// the record's own header, never a batch watermark) and the reader keeps
187    /// going. A failed poll surfaces the same way with no position, and the
188    /// next call polls again.
189    pub async fn next(&mut self) -> Option<Result<TypedRecord<T>, TypedDecodeError>> {
190        if self.buffered.is_empty() {
191            match self.cursor.poll().await {
192                Ok(messages) => self.buffered.extend(messages),
193                Err(error) => {
194                    return Some(Err(TypedDecodeError {
195                        position: None,
196                        source: Box::new(error),
197                    }));
198                }
199            }
200        }
201        let message = self.buffered.pop_front()?;
202        Some(decode_message(&self.form, message))
203    }
204
205    /// This reader as a [`Stream`](futures::Stream) of decoded records, one at a
206    /// time, ending once caught up (the same shape as the async typed reader in
207    /// the Python binding). Each item is a value or its position-carrying decode
208    /// error, so one poison record does not end the stream. Drive it with
209    /// `futures::StreamExt`.
210    pub fn stream(self) -> impl futures::Stream<Item = Result<TypedRecord<T>, TypedDecodeError>> {
211        futures::stream::unfold(self, |mut records| async move {
212            records.next().await.map(|item| (item, records))
213        })
214    }
215
216    /// One bounded poll, decoded: everything appended since the last read (at
217    /// most [`batch`](Self::batch) messages per partition), each record either
218    /// a value or its position-carrying decode error. An empty vec means
219    /// caught up. The batch altitude under [`next`](Self::next), for callers
220    /// that checkpoint per poll.
221    pub async fn poll(
222        &mut self,
223    ) -> Result<Vec<Result<TypedRecord<T>, TypedDecodeError>>, LaserError> {
224        let mut decoded: Vec<_> = self
225            .buffered
226            .drain(..)
227            .map(|message| decode_message(&self.form, message))
228            .collect();
229        if decoded.is_empty() {
230            decoded = self
231                .cursor
232                .poll()
233                .await?
234                .into_iter()
235                .map(|message| decode_message(&self.form, message))
236                .collect();
237        }
238        Ok(decoded)
239    }
240}
241
242/// One record decoded off the log: the typed body plus where it sits
243/// (partition and offset) and its user headers.
244#[derive(Clone, Debug)]
245pub struct TypedRecord<T> {
246    /// The payload decoded as the handle's body type.
247    pub value: T,
248    /// The record's log position, from its own message header.
249    pub position: MessageId,
250    /// The record's user headers decoded to strings.
251    pub headers: BTreeMap<String, String>,
252}
253
254/// A typed read that could not produce a value: a record whose payload does
255/// not decode as the handle's type (`position` names the exact record on the
256/// log), or a poll that failed before reaching any record (`position` is
257/// `None`).
258#[derive(Debug, thiserror::Error)]
259pub struct TypedDecodeError {
260    /// The failed record's log position. `None` when the poll itself failed.
261    pub position: Option<MessageId>,
262    /// What went wrong: a codec failure for a positioned record, the transport
263    /// error for a failed poll.
264    pub source: Box<LaserError>,
265}
266
267/// Folds into the codec family so `records.next()` composes with `?` in a
268/// `Result<_, LaserError>` function, the position kept in the message.
269impl From<TypedDecodeError> for LaserError {
270    fn from(error: TypedDecodeError) -> Self {
271        match error.position {
272            Some(_) => LaserError::Codec(error.to_string()),
273            None => *error.source,
274        }
275    }
276}
277
278impl std::fmt::Display for TypedDecodeError {
279    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
280        match &self.position {
281            Some(position) => write!(
282                formatter,
283                "record at {position} does not decode: {}",
284                self.source
285            ),
286            None => write!(formatter, "typed read failed to poll: {}", self.source),
287        }
288    }
289}
290
291fn decode_message<T: DeserializeOwned>(
292    form: &Form,
293    message: Message,
294) -> Result<TypedRecord<T>, TypedDecodeError> {
295    match form.decode::<T>(&message.payload) {
296        Ok(value) => Ok(TypedRecord {
297            value,
298            position: message.id,
299            headers: message.headers,
300        }),
301        Err(source) => Err(TypedDecodeError {
302            position: Some(message.id),
303            source: Box::new(source),
304        }),
305    }
306}
307
308// How the handle turns bodies into payloads and back. The compiled schema is
309// resolved and parsed once at handle construction, cloned (cheaply, the
310// expensive parts are shared) into each reader.
311#[derive(Clone)]
312enum Form {
313    Json,
314    Cbor,
315    #[cfg(feature = "schema-codecs")]
316    Schema {
317        id: u32,
318        compiled: crate::schema_codecs::CompiledSchema,
319    },
320}
321
322impl Form {
323    fn encode<T: Serialize>(
324        &self,
325        body: &T,
326    ) -> Result<(Vec<u8>, ContentType, Option<u32>), LaserError> {
327        match self {
328            Self::Json => Ok((Json::encode(body)?, ContentType::Json, None)),
329            Self::Cbor => Ok((Cbor::encode(body)?, ContentType::Cbor, None)),
330            #[cfg(feature = "schema-codecs")]
331            Self::Schema { id, compiled } => {
332                use crate::schema_codecs::CompiledSchema;
333                match compiled {
334                    CompiledSchema::Avro(_) => {
335                        Ok((compiled.encode_avro(body)?, ContentType::Avro, Some(*id)))
336                    }
337                    CompiledSchema::Json(_) => {
338                        let value = serde_json::to_value(body).map_err(|error| {
339                            LaserError::Codec(format!("body does not lower to JSON: {error}"))
340                        })?;
341                        if !compiled.validate_value(&value) {
342                            return Err(LaserError::Codec(format!(
343                                "body fails schema {id}'s JSON Schema"
344                            )));
345                        }
346                        let payload = serde_json::to_vec(&value).map_err(|error| {
347                            LaserError::Codec(format!("encode JSON payload: {error}"))
348                        })?;
349                        Ok((payload, ContentType::Json, Some(*id)))
350                    }
351                    CompiledSchema::Protobuf(_) => Err(LaserError::Invalid(format!(
352                        "schema {id} is Protobuf: encode the body with prost and publish it via \
353                         .raw_bytes(bytes, ContentType::Protobuf).schema_id({id}). The typed \
354                         handle still decodes Protobuf records"
355                    ))),
356                }
357            }
358        }
359    }
360
361    fn decode<T: DeserializeOwned>(&self, payload: &[u8]) -> Result<T, LaserError> {
362        match self {
363            Self::Json => Ok(<Json as Decoder<T>>::decode(payload)?),
364            Self::Cbor => Ok(<Cbor as Decoder<T>>::decode(payload)?),
365            #[cfg(feature = "schema-codecs")]
366            Self::Schema { compiled, .. } => {
367                let value = compiled.decode(payload)?;
368                serde_json::from_value(value).map_err(|error| {
369                    LaserError::Codec(format!(
370                        "decoded record does not fit the body type: {error}"
371                    ))
372                })
373            }
374        }
375    }
376}
377
378#[cfg(test)]
379mod tests {
380    use super::*;
381    use serde::Deserialize;
382
383    #[derive(Debug, PartialEq, Serialize, Deserialize)]
384    struct Reading {
385        host: String,
386        cpu: i64,
387    }
388
389    fn reading() -> Reading {
390        Reading {
391            host: "node-7".to_owned(),
392            cpu: 42,
393        }
394    }
395
396    #[test]
397    fn given_typed_bodies_when_batched_then_should_add_one_record_per_body() {
398        let laser = crate::laser::Laser::from_client(iggy::prelude::IggyClient::default());
399        let readings = laser.topic("readings").json::<Reading>();
400        let batch = readings
401            .publish_batch(&[reading(), reading()])
402            .expect("bodies encode");
403        assert_eq!(batch.len(), 2);
404        assert!(readings.publish_batch(&[]).expect("empty").is_empty());
405    }
406
407    #[test]
408    fn given_the_json_form_when_encoded_then_should_round_trip_with_the_json_tag() {
409        let form = Form::Json;
410        let (payload, content_type, schema_id) = form.encode(&reading()).expect("body encodes");
411        assert_eq!(content_type, ContentType::Json);
412        assert_eq!(schema_id, None);
413        let back: Reading = form.decode(&payload).expect("payload decodes");
414        assert_eq!(back, reading());
415    }
416
417    #[test]
418    fn given_the_cbor_form_when_encoded_then_should_round_trip_with_the_cbor_tag() {
419        let form = Form::Cbor;
420        let (payload, content_type, schema_id) = form.encode(&reading()).expect("body encodes");
421        assert_eq!(content_type, ContentType::Cbor);
422        assert_eq!(schema_id, None);
423        let back: Reading = form.decode(&payload).expect("payload decodes");
424        assert_eq!(back, reading());
425    }
426
427    #[test]
428    fn given_a_wrong_payload_when_decoded_then_should_fail_as_codec() {
429        let form = Form::Json;
430        let result: Result<Reading, _> = form.decode(b"not json");
431        assert!(matches!(result, Err(LaserError::Codec(_))));
432    }
433
434    #[cfg(feature = "schema-codecs")]
435    mod schema_bound {
436        use super::*;
437        use crate::query::{SchemaDef, SchemaSource};
438        use crate::schema_codecs::CompiledSchema;
439
440        const READING_AVRO: &str = r#"{
441            "type":"record","name":"Reading",
442            "fields":[
443                {"name":"host","type":"string"},
444                {"name":"cpu","type":"long"}
445            ]
446        }"#;
447
448        fn schema_form(source: SchemaSource) -> Form {
449            let compiled = CompiledSchema::compile(&SchemaDef {
450                id: 7,
451                source,
452                name: None,
453                version: None,
454            })
455            .expect("schema compiles");
456            Form::Schema { id: 7, compiled }
457        }
458
459        #[test]
460        fn given_an_avro_schema_when_encoded_then_should_round_trip_stamping_the_id() {
461            let form = schema_form(SchemaSource::Avro {
462                schema: READING_AVRO.to_owned(),
463            });
464            let (payload, content_type, schema_id) = form.encode(&reading()).expect("body encodes");
465            assert_eq!(content_type, ContentType::Avro);
466            assert_eq!(schema_id, Some(7));
467            let back: Reading = form.decode(&payload).expect("datum decodes");
468            assert_eq!(back, reading());
469        }
470
471        #[test]
472        fn given_a_json_schema_when_the_body_fails_it_then_should_refuse_before_send() {
473            let form = schema_form(SchemaSource::JsonSchema {
474                schema: r#"{
475                    "type":"object",
476                    "required":["host","cpu"],
477                    "properties":{"cpu":{"type":"integer","minimum":100}}
478                }"#
479                .to_owned(),
480            });
481            assert!(matches!(form.encode(&reading()), Err(LaserError::Codec(_))));
482        }
483
484        #[test]
485        fn given_a_json_schema_when_the_body_passes_then_should_stamp_json_plus_id() {
486            let form = schema_form(SchemaSource::JsonSchema {
487                schema: r#"{"type":"object","required":["host","cpu"]}"#.to_owned(),
488            });
489            let (payload, content_type, schema_id) = form.encode(&reading()).expect("body encodes");
490            assert_eq!(content_type, ContentType::Json);
491            assert_eq!(schema_id, Some(7));
492            let back: Reading = form.decode(&payload).expect("payload decodes");
493            assert_eq!(back, reading());
494        }
495    }
496
497    #[test]
498    fn given_a_positioned_failure_when_displayed_then_should_name_the_record() {
499        let error = TypedDecodeError {
500            position: Some(MessageId::new(2, 17)),
501            source: Box::new(LaserError::Codec("bad".to_owned())),
502        };
503        assert!(error.to_string().contains("2:17"));
504    }
505}
laser_sdk/typed.rs

laser_sdk/
typed.rs
