synd-persistence 0.4.0

Persistence adapters for syndicationd
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

use std::{fmt, io, str::FromStr};

use thiserror::Error;

pub(crate) type CompressionResult<T> = Result<T, CompressionError>;

/// Error returned while compressing or decompressing stored bytes.
#[derive(Debug, Error)]
pub(crate) enum CompressionError {
    #[error("unsupported compression algorithm: {0}")]
    UnsupportedAlgo(String),
    #[error("compression failed for {algo}: {source}")]
    Compress {
        algo: CompressionAlgo,
        source: io::Error,
    },
    #[error("decompression failed for {algo}: {source}")]
    Decompress {
        algo: CompressionAlgo,
        source: io::Error,
    },
}

/// Compression algorithm used for persisted bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CompressionAlgo {
    Zstd,
}

impl CompressionAlgo {
    const ZSTD: &'static str = "zstd";

    /// Returns the stable storage value for this algorithm.
    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::Zstd => Self::ZSTD,
        }
    }
}

impl fmt::Display for CompressionAlgo {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl FromStr for CompressionAlgo {
    type Err = CompressionError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value {
            Self::ZSTD => Ok(Self::Zstd),
            value => Err(CompressionError::UnsupportedAlgo(value.to_owned())),
        }
    }
}

/// Compresses and decompresses byte payloads.
pub(crate) trait CompressionCodec {
    /// Compresses bytes using the requested algorithm.
    fn compress(&self, algo: CompressionAlgo, bytes: &[u8]) -> CompressionResult<Vec<u8>>;

    /// Decompresses stored bytes using the recorded algorithm.
    fn decompress(
        &self,
        algo: CompressionAlgo,
        stored: StoredCompressedBytes<'_>,
    ) -> CompressionResult<Vec<u8>>;
}

/// Default compression codec used by `SQLite` persistence.
#[derive(Debug, Default, Clone, Copy)]
pub(crate) struct DefaultCompressionCodec;

impl CompressionCodec for DefaultCompressionCodec {
    fn compress(&self, algo: CompressionAlgo, bytes: &[u8]) -> CompressionResult<Vec<u8>> {
        match algo {
            CompressionAlgo::Zstd => zstd::compress(bytes),
        }
    }

    fn decompress(
        &self,
        algo: CompressionAlgo,
        stored: StoredCompressedBytes<'_>,
    ) -> CompressionResult<Vec<u8>> {
        match algo {
            CompressionAlgo::Zstd => zstd::decompress(stored),
        }
    }
}

/// Stored compressed bytes and metadata needed to decode them.
#[derive(Debug, Clone, Copy)]
pub(crate) struct StoredCompressedBytes<'a> {
    pub(crate) bytes: &'a [u8],
    pub(crate) uncompressed_len: usize,
}

mod zstd {
    use super::{CompressionAlgo, CompressionError, CompressionResult, StoredCompressedBytes};

    const LEVEL: i32 = 3;

    pub(super) fn compress(bytes: &[u8]) -> CompressionResult<Vec<u8>> {
        ::zstd::bulk::compress(bytes, LEVEL).map_err(|source| CompressionError::Compress {
            algo: CompressionAlgo::Zstd,
            source,
        })
    }

    pub(super) fn decompress(stored: StoredCompressedBytes<'_>) -> CompressionResult<Vec<u8>> {
        ::zstd::bulk::decompress(stored.bytes, stored.uncompressed_len).map_err(|source| {
            CompressionError::Decompress {
                algo: CompressionAlgo::Zstd,
                source,
            }
        })
    }
}