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}
persistent_queue/codec.rs

persistent_queue/
codec.rs
