laser_wire/
codecs.rs

1use crate::content::ContentType;
2use crate::error::DecodeError;
3use crate::kv::KvEntry;
4use crate::query::Row;
5use serde::Serialize;
6use serde::de::DeserializeOwned;
7
8/// Encoding strategy for a typed body. Four first-party codecs ship here, all
9/// self-describing so LaserData Cloud's projector can index their fields without a
10/// schema: [`Json`] (`serde_json`), [`Msgpack`] (`rmp_serde`), [`Cbor`]
11/// (`ciborium`), and [`Bson`] (`bson`, native-only feature). For a
12/// schema-first format (Avro or Protobuf), Arrow, or your own framing,
13/// implement `Codec` on a marker type. The codec advertises its
14/// `ContentType` so consumers can decode downstream.
15///
16/// The trait is generic on `T` so codecs can constrain the body type they
17/// accept. Serde-based codecs constrain `T: Serialize`, a Prost codec
18/// constrains `T: prost::Message`.
19///
20/// ```no_run
21/// # use laser_wire::codecs::Codec;
22/// # use laser_wire::content::ContentType;
23/// # use laser_wire::error::DecodeError;
24/// struct AvroCodec<S>(std::marker::PhantomData<S>);
25/// // Where `S` is your generated Avro schema:
26/// // impl<S: AvroSerialize> Codec<S> for AvroCodec<S> {
27/// //     fn content_type() -> ContentType { ContentType::Avro }
28/// //     fn encode(value: &S) -> Result<Vec<u8>, DecodeError> {
29/// //         avro::to_avro_bytes(value).map_err(|e| DecodeError::Encode(e.to_string()))
30/// //     }
31/// // }
32/// ```
33pub trait Codec<T: ?Sized> {
34    /// The wire-format tag stamped on `agdx.ct`.
35    fn content_type() -> ContentType;
36    /// Encode `value` into the bytes that ride the Iggy payload.
37    fn encode(value: &T) -> Result<Vec<u8>, DecodeError>;
38}
39
40/// The decode half of a codec. Separate from [`Codec`] because encoding is
41/// `?Sized` (you can encode a `&str`) while decoding must produce an owned,
42/// deserializable value. `Json`, `Msgpack`, `Cbor`, and `Bson` implement both.
43/// A custom codec (Avro, Protobuf, Arrow, or your own framing) implements only
44/// the half it needs.
45///
46/// ```no_run
47/// # use laser_wire::codecs::Decoder;
48/// # use laser_wire::error::DecodeError;
49/// struct AvroCodec<S>(std::marker::PhantomData<S>);
50/// // impl<S: AvroDeserialize> Decoder<S> for AvroCodec<S> {
51/// //     fn decode(payload: &[u8]) -> Result<S, DecodeError> {
52/// //         avro::from_avro_bytes(payload).map_err(|e| DecodeError::Decode(e.to_string()))
53/// //     }
54/// // }
55/// ```
56pub trait Decoder<T> {
57    /// Decode bytes previously produced by the matching [`Codec::encode`].
58    fn decode(payload: &[u8]) -> Result<T, DecodeError>;
59}
60
61/// Built-in JSON codec (`serde_json`). Constrains the body to `Serialize`.
62#[derive(Clone, Copy, Debug, Default)]
63pub struct Json;
64
65impl<T: Serialize + ?Sized> Codec<T> for Json {
66    fn content_type() -> ContentType {
67        ContentType::Json
68    }
69    fn encode(value: &T) -> Result<Vec<u8>, DecodeError> {
70        serde_json::to_vec(value)
71            .map_err(|error| DecodeError::Encode(format!("encode JSON payload: {error}")))
72    }
73}
74
75impl<T: DeserializeOwned> Decoder<T> for Json {
76    fn decode(payload: &[u8]) -> Result<T, DecodeError> {
77        serde_json::from_slice(payload)
78            .map_err(|error| DecodeError::Decode(format!("decode JSON payload: {error}")))
79    }
80}
81
82/// Built-in MessagePack codec (`rmp_serde`, named-map encoding so field
83/// names round-trip with JSON-shaped consumers). Constrains the body to
84/// `Serialize`.
85#[derive(Clone, Copy, Debug, Default)]
86pub struct Msgpack;
87
88impl<T: Serialize + ?Sized> Codec<T> for Msgpack {
89    fn content_type() -> ContentType {
90        ContentType::Msgpack
91    }
92    fn encode(value: &T) -> Result<Vec<u8>, DecodeError> {
93        rmp_serde::to_vec_named(value)
94            .map_err(|error| DecodeError::Encode(format!("encode msgpack payload: {error}")))
95    }
96}
97
98impl<T: DeserializeOwned> Decoder<T> for Msgpack {
99    fn decode(payload: &[u8]) -> Result<T, DecodeError> {
100        rmp_serde::from_slice(payload)
101            .map_err(|error| DecodeError::Decode(format!("decode msgpack payload: {error}")))
102    }
103}
104
105/// Built-in CBOR codec (`ciborium`). Self-describing like JSON, so LaserData Cloud's
106/// projector can index its fields without a schema. Constrains the body to
107/// `Serialize` / `DeserializeOwned`.
108#[derive(Clone, Copy, Debug, Default)]
109pub struct Cbor;
110
111impl<T: Serialize + ?Sized> Codec<T> for Cbor {
112    fn content_type() -> ContentType {
113        ContentType::Cbor
114    }
115    fn encode(value: &T) -> Result<Vec<u8>, DecodeError> {
116        let mut buffer = Vec::new();
117        ciborium::into_writer(value, &mut buffer)
118            .map_err(|error| DecodeError::Encode(format!("encode CBOR payload: {error}")))?;
119        Ok(buffer)
120    }
121}
122
123impl<T: DeserializeOwned> Decoder<T> for Cbor {
124    fn decode(payload: &[u8]) -> Result<T, DecodeError> {
125        ciborium::from_reader(payload)
126            .map_err(|error| DecodeError::Decode(format!("decode CBOR payload: {error}")))
127    }
128}
129
130/// Built-in BSON codec (`bson`, behind the native-only `bson` feature: its
131/// dependency tree does not build on `wasm32-unknown-unknown`). Self-describing
132/// like JSON. The top-level body must be a document (struct or map).
133#[cfg(feature = "bson")]
134#[derive(Clone, Copy, Debug, Default)]
135pub struct Bson;
136
137#[cfg(feature = "bson")]
138impl<T: Serialize> Codec<T> for Bson {
139    fn content_type() -> ContentType {
140        ContentType::Bson
141    }
142    fn encode(value: &T) -> Result<Vec<u8>, DecodeError> {
143        bson::serialize_to_vec(value)
144            .map_err(|error| DecodeError::Encode(format!("encode BSON payload: {error}")))
145    }
146}
147
148#[cfg(feature = "bson")]
149impl<T: DeserializeOwned> Decoder<T> for Bson {
150    fn decode(payload: &[u8]) -> Result<T, DecodeError> {
151        bson::deserialize_from_slice(payload)
152            .map_err(|error| DecodeError::Decode(format!("decode BSON payload: {error}")))
153    }
154}
155
156const NO_ROW_PAYLOAD: &str = "row has no payload, the publisher must call .inline_payload() and the query must call .with_payload()";
157
158impl Row {
159    /// Decodes the row's payload as JSON into `T`. Errors if no payload was
160    /// returned (publisher did not inline it, or the query did not request
161    /// it), or if the bytes fail to deserialize.
162    pub fn decode_json<T: DeserializeOwned>(&self) -> Result<T, DecodeError> {
163        self.decode_with::<Json, T>()
164    }
165
166    /// Decodes the row's payload as MessagePack into `T`. Same prerequisites
167    /// as [`decode_json`](Self::decode_json).
168    pub fn decode_msgpack<T: DeserializeOwned>(&self) -> Result<T, DecodeError> {
169        self.decode_with::<Msgpack, T>()
170    }
171
172    /// Decode the row's payload with any [`Decoder`] (`Json`, `Msgpack`, or
173    /// your own). `decode_json` and `decode_msgpack` are sugar for the
174    /// built-ins. Reach for this with a custom codec. Same prerequisites: the
175    /// publisher inlined the payload and the query requested it.
176    pub fn decode_with<C, T>(&self) -> Result<T, DecodeError>
177    where
178        C: Decoder<T>,
179    {
180        let payload = self
181            .payload
182            .as_deref()
183            .ok_or(DecodeError::MissingPayload(NO_ROW_PAYLOAD))?;
184        C::decode(payload)
185    }
186}
187
188impl KvEntry {
189    /// Decode the value as JSON into `T`. Sugar for `decode_value_with::<Json, _>`.
190    pub fn decode_value<T: DeserializeOwned>(&self) -> Result<T, DecodeError> {
191        self.decode_value_with::<Json, T>()
192    }
193
194    /// Decode the value with any [`Decoder`] (`Json`, `Msgpack`, or your own).
195    pub fn decode_value_with<C, T>(&self) -> Result<T, DecodeError>
196    where
197        C: Decoder<T>,
198    {
199        C::decode(&self.value)
200    }
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206
207    #[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug)]
208    struct Body {
209        id: u32,
210        name: String,
211    }
212
213    fn body() -> Body {
214        Body {
215            id: 7,
216            name: "alice".to_owned(),
217        }
218    }
219
220    #[test]
221    fn given_a_row_payload_when_decoded_with_each_codec_then_should_round_trip() {
222        let json_row = Row {
223            payload: Some(Json::encode(&body()).expect("json encode")),
224            ..Default::default()
225        };
226        assert_eq!(json_row.decode_with::<Json, Body>().expect("json"), body());
227        let msgpack_row = Row {
228            payload: Some(Msgpack::encode(&body()).expect("msgpack encode")),
229            ..Default::default()
230        };
231        assert_eq!(
232            msgpack_row.decode_with::<Msgpack, Body>().expect("msgpack"),
233            body()
234        );
235        let cbor_row = Row {
236            payload: Some(Cbor::encode(&body()).expect("cbor encode")),
237            ..Default::default()
238        };
239        assert_eq!(cbor_row.decode_with::<Cbor, Body>().expect("cbor"), body());
240    }
241
242    #[cfg(feature = "bson")]
243    #[test]
244    fn given_a_bson_payload_when_decoded_then_should_round_trip() {
245        let bson_row = Row {
246            payload: Some(Bson::encode(&body()).expect("bson encode")),
247            ..Default::default()
248        };
249        assert_eq!(bson_row.decode_with::<Bson, Body>().expect("bson"), body());
250        assert_eq!(<Bson as Codec<Body>>::content_type(), ContentType::Bson);
251    }
252
253    #[test]
254    fn given_each_codec_when_encoding_then_should_advertise_its_content_type() {
255        assert_eq!(<Json as Codec<str>>::content_type(), ContentType::Json);
256        assert_eq!(
257            <Msgpack as Codec<str>>::content_type(),
258            ContentType::Msgpack
259        );
260        assert_eq!(<Cbor as Codec<str>>::content_type(), ContentType::Cbor);
261    }
262
263    #[test]
264    fn given_a_row_without_payload_when_decoded_then_should_error() {
265        let row = Row::default();
266        assert!(matches!(
267            row.decode_with::<Json, String>(),
268            Err(DecodeError::MissingPayload(_))
269        ));
270    }
271
272    #[test]
273    fn given_a_msgpack_value_when_round_tripped_through_the_entry_then_should_decode_back() {
274        let encoded = Msgpack::encode(&vec!["a", "b"]).expect("encode");
275        let entry = KvEntry {
276            key: b"k".to_vec(),
277            value: encoded,
278            expires_at_micros: None,
279            version: 0,
280            scope: None,
281            source: None,
282        };
283        let decoded: Vec<String> = entry.decode_value_with::<Msgpack, _>().expect("decode");
284        assert_eq!(decoded, vec!["a".to_owned(), "b".to_owned()]);
285    }
286}
laser_wire/codecs.rs

laser_wire/
codecs.rs
