laser_wire/
codecs.rs

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

laser_wire/
codecs.rs
