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        // `legacy()` is little-endian + fixint, byte-identical to bincode 1.x's
67        // default, so queues written before this upgrade still decode.
68        bincode::serde::encode_to_vec(value, bincode::config::legacy()).map_err(CodecError::new)
69    }
70
71    fn decode(&self, bytes: &[u8]) -> Result<T, CodecError> {
72        bincode::serde::decode_from_slice(bytes, bincode::config::legacy())
73            .map(|(value, _)| value)
74            .map_err(CodecError::new)
75    }
76}
77
78/// A [`Codec`] that encodes with rkyv. Requires the `rkyv` feature. The message type
79/// must derive rkyv's `Archive`, `Serialize`, and `Deserialize`.
80///
81/// ```
82/// use persistent_queue::{Builder, MemStore, Rkyv};
83///
84/// #[derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize, Debug, PartialEq)]
85/// struct Job {
86///     id: u64,
87///     name: String,
88/// }
89///
90/// let (tx, rx) = Builder::new(MemStore::new()).open_typed(Rkyv).unwrap();
91/// tx.push(&Job { id: 1, name: "build".into() }).unwrap();
92///
93/// let item = rx.reserve().unwrap().unwrap();
94/// assert_eq!(*item, Job { id: 1, name: "build".into() });
95/// item.ack().unwrap();
96/// ```
97#[cfg(feature = "rkyv")]
98#[derive(Clone, Copy, Debug, Default)]
99pub struct Rkyv;
100
101#[cfg(feature = "rkyv")]
102impl<T> Codec<T> for Rkyv
103where
104    T: rkyv::Archive
105        + for<'a> rkyv::Serialize<
106            rkyv::rancor::Strategy<
107                rkyv::ser::Serializer<
108                    rkyv::util::AlignedVec,
109                    rkyv::ser::allocator::ArenaHandle<'a>,
110                    rkyv::ser::sharing::Share,
111                >,
112                rkyv::rancor::Error,
113            >,
114        >,
115    T::Archived: for<'a> rkyv::bytecheck::CheckBytes<rkyv::api::high::HighValidator<'a, rkyv::rancor::Error>>
116        + rkyv::Deserialize<T, rkyv::api::high::HighDeserializer<rkyv::rancor::Error>>,
117{
118    fn encode(&self, value: &T) -> Result<Vec<u8>, CodecError> {
119        rkyv::to_bytes::<rkyv::rancor::Error>(value)
120            .map(|bytes| bytes.to_vec())
121            .map_err(CodecError::new)
122    }
123
124    fn decode(&self, bytes: &[u8]) -> Result<T, CodecError> {
125        // rkyv needs an aligned buffer; the store hands back a plain `Vec<u8>`.
126        let mut aligned = rkyv::util::AlignedVec::<16>::new();
127        aligned.extend_from_slice(bytes);
128        rkyv::from_bytes::<T, rkyv::rancor::Error>(&aligned).map_err(CodecError::new)
129    }
130}
persistent_queue/codec.rs

persistent_queue/
codec.rs
