aggligator_wrapper_zstd/
lib.rs

1#![warn(missing_docs)]
2#![cfg_attr(docsrs, feature(doc_cfg))]
3#![doc(
4    html_logo_url = "https://raw.githubusercontent.com/remoc-rs/aggligator/master/.misc/aggligator.png",
5    html_favicon_url = "https://raw.githubusercontent.com/remoc-rs/aggligator/master/.misc/aggligator.png",
6    issue_tracker_base_url = "https://github.com/remoc-rs/aggligator/issues/"
7)]
8
9//! [Aggligator](aggligator) transport wrapper using zstd compression
10//!
11//! This compresses the data sent over each Aggligator link using [zstd](https://facebook.github.io/zstd/).
12//!
13//! The wrapper must be used on both ends of a connection, since compression is not negotiated.
14//! The compression level only affects the sending side and thus may differ between both ends.
15//!
16//! Compression only pays off when the transferred data is compressible and the links,
17//! not the CPU, are the bottleneck. Data that is already compressed or encrypted does not
18//! become smaller. Alternatively, the application can compress the data of the aggregated
19//! connection itself, which compresses the data only once instead of separately on each link.
20//!
21//! # Combining with TLS
22//! Wrappers are applied in the order they are added, thus the first wrapper is applied directly
23//! to the transport. To compress data before it is encrypted, add the TLS wrapper first and then
24//! this wrapper.
25//!
26//! Compressing data before encrypting it may reveal information about the data through
27//! the length of the encrypted data. This allows attacks if data controlled by an attacker is
28//! transferred together with secret data over the same connection.
29//!
30//! # Example
31//! ```no_run
32//! use aggligator::transport::ConnectorBuilder;
33//! use aggligator_wrapper_zstd::Zstd;
34//!
35//! let mut builder = ConnectorBuilder::new(Default::default());
36//! builder.wrap(Zstd::new());
37//! let connector = builder.build();
38//! ```
39
40use async_compression::{
41    Level,
42    tokio::{bufread::ZstdDecoder, write::ZstdEncoder},
43    zstd::{CParameter, DParameter},
44};
45use async_trait::async_trait;
46use std::io::{Error, ErrorKind, Result};
47use tokio::io::BufReader;
48
49use aggligator::{
50    io::{IoBox, StreamBox},
51    transport::{AcceptingWrapper, ConnectingWrapper},
52};
53
54static NAME: &str = "zstd";
55
56/// zstd compression wrapper for outgoing and incoming links.
57///
58/// Only IO-based streams are supported; wrapping a packet-based stream fails.
59///
60/// Add this to a [`ConnectorBuilder`](aggligator::transport::ConnectorBuilder) and the
61/// [`AcceptorBuilder`](aggligator::transport::AcceptorBuilder) of the remote endpoint
62/// to compress all links of the connection.
63#[derive(Debug, Clone)]
64#[must_use = "you must pass this wrapper to the connector or acceptor"]
65pub struct Zstd {
66    level: i32,
67    window_log: Option<u32>,
68}
69
70impl Default for Zstd {
71    fn default() -> Self {
72        Self::new()
73    }
74}
75
76impl Zstd {
77    /// Default compression level.
78    ///
79    /// This is the fastest regular level of zstd.
80    pub const DEFAULT_LEVEL: i32 = 1;
81
82    /// Creates a new zstd compression wrapper using the [default level](Self::DEFAULT_LEVEL).
83    pub fn new() -> Self {
84        Self::with_level(Self::DEFAULT_LEVEL)
85    }
86
87    /// Creates a new zstd compression wrapper using the specified compression level.
88    ///
89    /// Higher levels compress better but are slower. Negative levels are faster than
90    /// level 1 but compress worse. The level is clamped to the range supported by zstd.
91    pub fn with_level(level: i32) -> Self {
92        Self { level, window_log: None }
93    }
94
95    /// The compression level.
96    pub fn level(&self) -> i32 {
97        self.level
98    }
99
100    /// Sets the base 2 logarithm of the maximum window size.
101    ///
102    /// This limits the memory used for compressing and decompressing each link.
103    /// Received data that was compressed using a larger window cannot be decompressed,
104    /// thus both ends must use the same setting.
105    ///
106    /// If `None` (default), the window size is determined by the compression level
107    /// for compressing and the default limit of zstd (128 MiB) applies to decompressing.
108    pub fn set_window_log(&mut self, window_log: Option<u32>) {
109        self.window_log = window_log;
110    }
111
112    /// Wraps the stream of a link.
113    fn wrap(&self, stream: StreamBox) -> Result<StreamBox> {
114        let StreamBox::Io(io) = stream else {
115            return Err(Error::new(ErrorKind::Unsupported, "zstd wrapper only supports IO-based streams"));
116        };
117        let (read, write) = io.into_split();
118
119        let cparams: Vec<_> = self.window_log.map(CParameter::window_log).into_iter().collect();
120        let encoder = ZstdEncoder::with_quality_and_params(write, Level::Precise(self.level), &cparams);
121
122        let dparams: Vec<_> = self.window_log.map(DParameter::window_log_max).into_iter().collect();
123        let mut decoder = ZstdDecoder::with_params(BufReader::new(read), &dparams);
124        decoder.multiple_members(true);
125
126        Ok(IoBox::new(decoder, encoder).into())
127    }
128}
129
130#[async_trait]
131impl ConnectingWrapper for Zstd {
132    fn name(&self) -> &str {
133        NAME
134    }
135
136    async fn wrap(&self, stream: StreamBox) -> Result<StreamBox> {
137        Zstd::wrap(self, stream)
138    }
139}
140
141#[async_trait]
142impl AcceptingWrapper for Zstd {
143    fn name(&self) -> &str {
144        NAME
145    }
146
147    async fn wrap(&self, stream: StreamBox) -> Result<StreamBox> {
148        Zstd::wrap(self, stream)
149    }
150}
151
152#[cfg(test)]
153mod tests;
aggligator_wrapper_zstd/lib.rs

aggligator_wrapper_zstd/
lib.rs
