persistent_queue/
codec.rs

1//! The [`Codec`] trait and built-in codecs for the typed queue layer.
2
3use std::error::Error as StdError;
4use std::fmt;
5
6/// Encodes a message type to bytes for the store and decodes it back.
7///
8/// The typed layer ([`TypedProducer`](crate::TypedProducer) /
9/// [`TypedConsumer`](crate::TypedConsumer)) is generic over this trait: implement it
10/// for a custom format, or use a built-in like [`Bincode`] behind the `serde` feature.
11pub trait Codec<T> {
12    /// Encode `value` to bytes.
13    fn encode(&self, value: &T) -> Result<Vec<u8>, CodecError>;
14    /// Decode a value from `bytes`.
15    fn decode(&self, bytes: &[u8]) -> Result<T, CodecError>;
16}
17
18/// An encode or decode failure, carrying the underlying codec's message.
19#[derive(Debug)]
20pub struct CodecError(String);
21
22impl CodecError {
23    /// Build a codec error from anything printable, e.g. the codec's own error.
24    pub fn new(error: impl fmt::Display) -> Self {
25        Self(error.to_string())
26    }
27}
28
29impl fmt::Display for CodecError {
30    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
31        write!(f, "codec error: {}", self.0)
32    }
33}
34
35impl StdError for CodecError {}
36
37/// A [`Codec`] that encodes with serde and bincode. Requires the `serde` feature.
38///
39/// ```
40/// use persistent_queue::{Bincode, Builder, MemStore};
41/// use serde::{Deserialize, Serialize};
42///
43/// #[derive(Serialize, Deserialize, Debug, PartialEq)]
44/// struct Job {
45///     id: u64,
46///     name: String,
47/// }
48///
49/// let (tx, rx) = Builder::new(MemStore::new()).open_typed(Bincode).unwrap();
50/// tx.push(&Job { id: 1, name: "build".into() }).unwrap();
51///
52/// let item = rx.reserve().unwrap().unwrap();
53/// assert_eq!(*item, Job { id: 1, name: "build".into() });
54/// item.ack().unwrap();
55/// ```
56#[cfg(feature = "serde")]
57#[derive(Clone, Copy, Debug, Default)]
58pub struct Bincode;
59
60#[cfg(feature = "serde")]
61impl<T> Codec<T> for Bincode
62where
63    T: serde::Serialize + serde::de::DeserializeOwned,
64{
65    fn encode(&self, value: &T) -> Result<Vec<u8>, CodecError> {
66        bincode::serialize(value).map_err(CodecError::new)
67    }
68
69    fn decode(&self, bytes: &[u8]) -> Result<T, CodecError> {
70        bincode::deserialize(bytes).map_err(CodecError::new)
71    }
72}
73
74/// A [`Codec`] that encodes with rkyv. Requires the `rkyv` feature. The message type
75/// must derive rkyv's `Archive`, `Serialize`, and `Deserialize`.
76///
77/// ```
78/// use persistent_queue::{Builder, MemStore, Rkyv};
79///
80/// #[derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize, Debug, PartialEq)]
81/// struct Job {
82///     id: u64,
83///     name: String,
84/// }
85///
86/// let (tx, rx) = Builder::new(MemStore::new()).open_typed(Rkyv).unwrap();
87/// tx.push(&Job { id: 1, name: "build".into() }).unwrap();
88///
89/// let item = rx.reserve().unwrap().unwrap();
90/// assert_eq!(*item, Job { id: 1, name: "build".into() });
91/// item.ack().unwrap();
92/// ```
93#[cfg(feature = "rkyv")]
94#[derive(Clone, Copy, Debug, Default)]
95pub struct Rkyv;
96
97#[cfg(feature = "rkyv")]
98impl<T> Codec<T> for Rkyv
99where
100    T: rkyv::Archive
101        + for<'a> rkyv::Serialize<
102            rkyv::rancor::Strategy<
103                rkyv::ser::Serializer<
104                    rkyv::util::AlignedVec,
105                    rkyv::ser::allocator::ArenaHandle<'a>,
106                    rkyv::ser::sharing::Share,
107                >,
108                rkyv::rancor::Error,
109            >,
110        >,
111    T::Archived: for<'a> rkyv::bytecheck::CheckBytes<rkyv::api::high::HighValidator<'a, rkyv::rancor::Error>>
112        + rkyv::Deserialize<T, rkyv::api::high::HighDeserializer<rkyv::rancor::Error>>,
113{
114    fn encode(&self, value: &T) -> Result<Vec<u8>, CodecError> {
115        rkyv::to_bytes::<rkyv::rancor::Error>(value)
116            .map(|bytes| bytes.to_vec())
117            .map_err(CodecError::new)
118    }
119
120    fn decode(&self, bytes: &[u8]) -> Result<T, CodecError> {
121        // rkyv needs an aligned buffer; the store hands back a plain `Vec<u8>`.
122        let mut aligned = rkyv::util::AlignedVec::<16>::new();
123        aligned.extend_from_slice(bytes);
124        rkyv::from_bytes::<T, rkyv::rancor::Error>(&aligned).map_err(CodecError::new)
125    }
126}
persistent_queue/codec.rs

persistent_queue/
codec.rs
