zstd_complete/encoding/
streaming_encoder.rs

1use alloc::{string::ToString, vec::Vec};
2#[cfg(feature = "hash")]
3use core::hash::Hasher;
4use core::{convert::TryFrom, fmt};
5use std::io::{self, Read, Write};
6
7use super::{levels::c_port, CompressionLevel, CompressionTuning, ContentSizePolicy};
8use crate::{
9    blocks::block::BlockType,
10    common::MAX_BLOCK_SIZE,
11    encoding::{block_header::BlockHeader, frame_header::FrameHeader},
12};
13
14/// Default maximum estimated resident working set: 96 MiB.
15pub const DEFAULT_MEMORY_LIMIT: usize = 96 * 1024 * 1024;
16/// Default amount of input stored in each independent Zstandard frame: 8 MiB.
17pub const DEFAULT_FRAME_CHUNK_SIZE: usize = 8 * 1024 * 1024;
18
19/// Configuration for the bounded-memory streaming encoder.
20#[derive(Clone)]
21pub struct EncoderOptions {
22    level: CompressionLevel,
23    memory_limit: usize,
24    frame_chunk_size: usize,
25    dictionary: Option<EncoderDictionary>,
26    checksum: bool,
27    tuning: CompressionTuning,
28    pledged_source_size: Option<u64>,
29    content_size_policy: ContentSizePolicy,
30}
31
32impl EncoderOptions {
33    /// Creates options using the 96 MiB memory budget and 8 MiB frame chunks.
34    pub const fn new(level: CompressionLevel) -> Self {
35        Self {
36            level,
37            memory_limit: DEFAULT_MEMORY_LIMIT,
38            frame_chunk_size: DEFAULT_FRAME_CHUNK_SIZE,
39            dictionary: None,
40            checksum: false,
41            tuning: CompressionTuning::new(),
42            pledged_source_size: None,
43            content_size_policy: ContentSizePolicy::Include,
44        }
45    }
46
47    pub const fn level(&self) -> CompressionLevel {
48        self.level
49    }
50
51    pub const fn memory_limit(&self) -> usize {
52        self.memory_limit
53    }
54
55    pub const fn frame_chunk_size(&self) -> usize {
56        self.frame_chunk_size
57    }
58
59    pub fn dictionary(&self) -> Option<&EncoderDictionary> {
60        self.dictionary.as_ref()
61    }
62
63    pub const fn checksum(&self) -> bool {
64        self.checksum
65    }
66
67    pub const fn tuning(&self) -> CompressionTuning {
68        self.tuning
69    }
70
71    pub const fn pledged_source_size(&self) -> Option<u64> {
72        self.pledged_source_size
73    }
74
75    pub const fn content_size_policy(&self) -> ContentSizePolicy {
76        self.content_size_policy
77    }
78
79    /// Returns the conservative memory estimate checked by [`Encoder::new`].
80    pub fn estimated_memory_usage(&self) -> usize {
81        let required = c_port::estimated_frame_memory(self.level.c_level(), self.frame_chunk_size);
82        self.dictionary.as_ref().map_or(required, |dictionary| {
83            let retained_copies = if cfg!(feature = "multithreading") {
84                3
85            } else {
86                2
87            };
88            required.saturating_add(dictionary.raw_size().saturating_mul(retained_copies))
89        })
90    }
91
92    /// Sets the maximum conservative working-memory estimate.
93    pub const fn with_memory_limit(mut self, bytes: usize) -> Self {
94        self.memory_limit = bytes;
95        self
96    }
97
98    /// Sets the input bytes collected per independent Zstandard frame.
99    ///
100    /// Smaller chunks reduce latency and memory use but can reduce compression
101    /// ratio because matches do not cross frame boundaries.
102    pub const fn with_frame_chunk_size(mut self, bytes: usize) -> Self {
103        self.frame_chunk_size = bytes;
104        self
105    }
106
107    /// Uses a prepared dictionary for every frame in the output archive.
108    pub fn with_dictionary(mut self, dictionary: EncoderDictionary) -> Self {
109        self.dictionary = Some(dictionary);
110        self
111    }
112
113    /// Enables or disables the standard 32-bit frame content checksum.
114    pub const fn with_checksum(mut self, enabled: bool) -> Self {
115        self.checksum = enabled;
116        self
117    }
118
119    /// Applies validated, typed overrides to the selected level preset.
120    pub const fn with_tuning(mut self, tuning: CompressionTuning) -> Self {
121        self.tuning = tuning;
122        self
123    }
124
125    /// Declares the exact number of input bytes the encoder must receive.
126    pub const fn with_pledged_source_size(mut self, bytes: u64) -> Self {
127        self.pledged_source_size = Some(bytes);
128        self
129    }
130
131    /// Includes or omits the decoded size in each emitted frame header.
132    pub const fn with_content_size_policy(mut self, policy: ContentSizePolicy) -> Self {
133        self.content_size_policy = policy;
134        self
135    }
136
137    pub(super) fn validate(&self) -> Result<(), EncodeError> {
138        if self.frame_chunk_size == 0 {
139            return Err(EncodeError::InvalidOptions(
140                "frame chunk size must be greater than zero",
141            ));
142        }
143        if self.level.is_uncompressed() && self.dictionary.is_some() {
144            return Err(EncodeError::InvalidOptions(
145                "an uncompressed encoder cannot use a dictionary",
146            ));
147        }
148        if self.level.is_uncompressed() && self.tuning != CompressionTuning::new() {
149            return Err(EncodeError::InvalidOptions(
150                "an uncompressed encoder cannot use compression tuning",
151            ));
152        }
153        self.tuning
154            .validate()
155            .map_err(EncodeError::InvalidOptions)?;
156        if self.checksum && !cfg!(feature = "hash") {
157            return Err(EncodeError::InvalidOptions(
158                "checksums require the zstd-complete `hash` feature",
159            ));
160        }
161        let required = self.estimated_memory_usage();
162        if required > self.memory_limit {
163            return Err(EncodeError::MemoryLimitExceeded {
164                limit: self.memory_limit,
165                required,
166            });
167        }
168        Ok(())
169    }
170}
171
172impl fmt::Debug for EncoderOptions {
173    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
174        formatter
175            .debug_struct("EncoderOptions")
176            .field("level", &self.level)
177            .field("memory_limit", &self.memory_limit)
178            .field("frame_chunk_size", &self.frame_chunk_size)
179            .field("dictionary", &self.dictionary)
180            .field("checksum", &self.checksum)
181            .field("tuning", &self.tuning)
182            .field("pledged_source_size", &self.pledged_source_size)
183            .field("content_size_policy", &self.content_size_policy)
184            .finish()
185    }
186}
187
188/// An owned, parsed dictionary reusable across independent frames.
189#[derive(Clone)]
190pub struct EncoderDictionary {
191    prepared: c_port::PreparedDictionary,
192    raw_size: usize,
193    #[cfg(feature = "multithreading")]
194    raw: Vec<u8>,
195}
196
197impl EncoderDictionary {
198    /// Copies and validates raw-content or formatted Zstandard dictionary data.
199    pub fn copy(dictionary: &[u8]) -> Result<Self, DictionaryError> {
200        let prepared = c_port::PreparedDictionary::from_bytes(dictionary)
201            .map_err(|error| match error {
202                c_port::DictionaryParseError::WrongDictionary => DictionaryError::WrongDictionary,
203                c_port::DictionaryParseError::CorruptedDictionary => {
204                    DictionaryError::CorruptedDictionary
205                }
206            })?
207            .ok_or(DictionaryError::WrongDictionary)?;
208        Ok(Self {
209            prepared,
210            raw_size: dictionary.len(),
211            #[cfg(feature = "multithreading")]
212            raw: dictionary.to_vec(),
213        })
214    }
215
216    pub const fn raw_size(&self) -> usize {
217        self.raw_size
218    }
219
220    #[cfg(feature = "multithreading")]
221    pub(super) fn raw(&self) -> &[u8] {
222        &self.raw
223    }
224}
225
226impl fmt::Debug for EncoderDictionary {
227    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
228        formatter
229            .debug_struct("EncoderDictionary")
230            .field("raw_size", &self.raw_size)
231            .finish_non_exhaustive()
232    }
233}
234
235/// Error returned when dictionary bytes are not a usable Zstandard dictionary.
236#[derive(Clone, Copy, Debug, Eq, PartialEq)]
237pub enum DictionaryError {
238    WrongDictionary,
239    CorruptedDictionary,
240}
241
242impl fmt::Display for DictionaryError {
243    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
244        formatter.write_str(match self {
245            Self::WrongDictionary => "input is not a usable Zstandard dictionary",
246            Self::CorruptedDictionary => "Zstandard dictionary metadata is corrupted",
247        })
248    }
249}
250
251impl std::error::Error for DictionaryError {}
252
253impl Default for EncoderOptions {
254    fn default() -> Self {
255        Self::new(CompressionLevel::DEFAULT)
256    }
257}
258
259/// Errors produced by the bounded streaming API.
260#[derive(Debug)]
261pub enum EncodeError {
262    Io(io::Error),
263    InvalidOptions(&'static str),
264    MemoryLimitExceeded {
265        limit: usize,
266        required: usize,
267    },
268    PledgedSourceSizeMismatch {
269        pledged: u64,
270        actual: u64,
271    },
272    #[cfg(feature = "multithreading")]
273    WorkerFailed,
274}
275
276impl fmt::Display for EncodeError {
277    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
278        match self {
279            Self::Io(error) => error.fmt(formatter),
280            Self::InvalidOptions(message) => formatter.write_str(message),
281            Self::MemoryLimitExceeded { limit, required } => write!(
282                formatter,
283                "encoder needs an estimated {required} bytes, exceeding the {limit}-byte memory limit"
284            ),
285            Self::PledgedSourceSizeMismatch { pledged, actual } => write!(
286                formatter,
287                "encoder received {actual} bytes but was pledged exactly {pledged} bytes"
288            ),
289            #[cfg(feature = "multithreading")]
290            Self::WorkerFailed => formatter.write_str("a compression worker failed"),
291        }
292    }
293}
294
295impl std::error::Error for EncodeError {
296    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
297        match self {
298            Self::Io(error) => Some(error),
299            _ => None,
300        }
301    }
302}
303
304impl From<io::Error> for EncodeError {
305    fn from(error: io::Error) -> Self {
306        Self::Io(error)
307    }
308}
309
310/// A bounded-memory Zstandard encoder implementing [`Write`].
311///
312/// Each full input chunk becomes an independent frame in the output archive.
313/// Call [`Encoder::finish`] to emit the final frame and recover the writer.
314pub struct Encoder<W: Write> {
315    inner: W,
316    options: EncoderOptions,
317    input: Vec<u8>,
318    emitted_frame: bool,
319    accepted_input: u64,
320}
321
322impl<W: Write> Encoder<W> {
323    pub fn new(inner: W, options: EncoderOptions) -> Result<Self, EncodeError> {
324        options.validate()?;
325        let input = Vec::with_capacity(options.frame_chunk_size);
326        Ok(Self {
327            inner,
328            options,
329            input,
330            emitted_frame: false,
331            accepted_input: 0,
332        })
333    }
334
335    pub fn get_ref(&self) -> &W {
336        &self.inner
337    }
338
339    pub fn get_mut(&mut self) -> &mut W {
340        &mut self.inner
341    }
342
343    /// Emits the final frame and returns the wrapped writer.
344    pub fn finish(mut self) -> Result<W, EncodeError> {
345        self.verify_pledged_size()?;
346        if !self.input.is_empty() || !self.emitted_frame {
347            self.emit_frame()?;
348        }
349        self.inner.flush()?;
350        Ok(self.inner)
351    }
352
353    fn emit_frame(&mut self) -> io::Result<()> {
354        let mut compressed = encode_frame_payload(&self.input, &self.options);
355        apply_content_size_policy(&mut compressed, self.options.content_size_policy)?;
356        if self.options.checksum {
357            append_checksum(&mut compressed, &self.input);
358        }
359        self.inner.write_all(&compressed)?;
360        self.input.clear();
361        self.emitted_frame = true;
362        Ok(())
363    }
364
365    fn verify_pledged_size(&self) -> Result<(), EncodeError> {
366        if let Some(pledged) = self.options.pledged_source_size {
367            if pledged != self.accepted_input {
368                return Err(EncodeError::PledgedSourceSizeMismatch {
369                    pledged,
370                    actual: self.accepted_input,
371                });
372            }
373        }
374        Ok(())
375    }
376}
377
378fn encode_frame_payload(source: &[u8], options: &EncoderOptions) -> Vec<u8> {
379    if options.level.is_uncompressed() {
380        return encode_uncompressed_frame(source);
381    }
382    let has_tuning = options.tuning != CompressionTuning::new();
383    match (options.dictionary.as_ref(), has_tuning) {
384        (Some(dictionary), false) => c_port::encode_frame_with_prepared_dictionary(
385            source,
386            options.level.c_level(),
387            &dictionary.prepared,
388        ),
389        (Some(dictionary), true) => c_port::encode_frame_with_prepared_dictionary_and_tuning(
390            source,
391            options.level.c_level(),
392            &dictionary.prepared,
393            options.tuning,
394        ),
395        (None, false) => c_port::encode_frame_no_dict(source, options.level.c_level()),
396        (None, true) => c_port::encode_frame_no_dict_with_tuning(
397            source,
398            options.level.c_level(),
399            options.tuning,
400        ),
401    }
402}
403
404fn apply_content_size_policy(frame: &mut Vec<u8>, policy: ContentSizePolicy) -> io::Result<()> {
405    if policy == ContentSizePolicy::Include {
406        return Ok(());
407    }
408    let (parsed, old_header_size) = crate::decoding::frame::read_frame_header(frame.as_slice())
409        .map_err(|error| io::Error::other(error.to_string()))?;
410    let window_size = parsed.window_size().unwrap_or(1024).max(1024);
411    let mut header = Vec::with_capacity(18);
412    FrameHeader {
413        frame_content_size: None,
414        single_segment: false,
415        content_checksum: false,
416        dictionary_id: parsed.dictionary_id().map(u64::from),
417        window_size: Some(window_size),
418    }
419    .serialize(&mut header).expect("owned legacy codec allocation failed");
420    frame.splice(..usize::from(old_header_size), header);
421    Ok(())
422}
423
424fn append_checksum(frame: &mut Vec<u8>, source: &[u8]) {
425    #[cfg(feature = "hash")]
426    {
427        debug_assert!(frame.len() >= 5);
428        frame[4] |= 1 << 2;
429        let mut hasher = twox_hash::XxHash64::with_seed(0);
430        hasher.write(source);
431        frame.extend_from_slice(&(hasher.finish() as u32).to_le_bytes());
432    }
433    #[cfg(not(feature = "hash"))]
434    {
435        let _ = (frame, source);
436        unreachable!("checksum options are rejected without the hash feature");
437    }
438}
439
440#[cfg(feature = "multithreading")]
441pub(super) fn encode_frame_for_options(source: &[u8], options: &EncoderOptions) -> Vec<u8> {
442    let mut compressed = encode_frame_payload(source, options);
443    apply_content_size_policy(&mut compressed, options.content_size_policy)
444        .expect("validated frame header can be rewritten");
445    if options.checksum {
446        append_checksum(&mut compressed, source);
447    }
448    compressed
449}
450
451fn encode_uncompressed_frame(source: &[u8]) -> Vec<u8> {
452    let mut output = Vec::with_capacity(source.len().saturating_add(16));
453    FrameHeader {
454        frame_content_size: Some(source.len() as u64),
455        single_segment: true,
456        content_checksum: false,
457        dictionary_id: None,
458        window_size: None,
459    }
460    .serialize(&mut output).expect("owned legacy codec allocation failed");
461
462    if source.is_empty() {
463        BlockHeader {
464            last_block: true,
465            block_type: BlockType::Raw,
466            block_size: 0,
467        }
468        .serialize(&mut output).expect("owned legacy codec allocation failed");
469        return output;
470    }
471
472    for (index, block) in source.chunks(MAX_BLOCK_SIZE as usize).enumerate() {
473        BlockHeader {
474            last_block: (index + 1) * MAX_BLOCK_SIZE as usize >= source.len(),
475            block_type: BlockType::Raw,
476            block_size: block.len() as u32,
477        }
478        .serialize(&mut output).expect("owned legacy codec allocation failed");
479        output.extend_from_slice(block);
480    }
481    output
482}
483
484impl<W: Write> Write for Encoder<W> {
485    fn write(&mut self, source: &[u8]) -> io::Result<usize> {
486        if source.is_empty() {
487            return Ok(0);
488        }
489        if self.input.len() == self.options.frame_chunk_size {
490            self.emit_frame()?;
491        }
492        let available = self.options.frame_chunk_size - self.input.len();
493        let consumed = available.min(source.len());
494        if let Some(pledged) = self.options.pledged_source_size {
495            let remaining = pledged.saturating_sub(self.accepted_input);
496            if remaining == 0 {
497                return Err(io::Error::other(EncodeError::PledgedSourceSizeMismatch {
498                    pledged,
499                    actual: self.accepted_input.saturating_add(source.len() as u64),
500                }));
501            }
502            let consumed = consumed.min(usize::try_from(remaining).unwrap_or(usize::MAX));
503            self.input.extend_from_slice(&source[..consumed]);
504            self.accepted_input += consumed as u64;
505            return Ok(consumed);
506        }
507        self.input.extend_from_slice(&source[..consumed]);
508        self.accepted_input += consumed as u64;
509        Ok(consumed)
510    }
511
512    /// Closes the current frame before flushing the wrapped writer.
513    fn flush(&mut self) -> io::Result<()> {
514        if !self.input.is_empty() {
515            self.emit_frame()?;
516        }
517        self.inner.flush()
518    }
519}
520
521/// Streams all input into a bounded-memory encoder.
522pub fn encode<R: Read, W: Write>(
523    mut source: R,
524    target: W,
525    options: EncoderOptions,
526) -> Result<(), EncodeError> {
527    let mut encoder = Encoder::new(target, options)?;
528    io::copy(&mut source, &mut encoder)?;
529    encoder.finish()?;
530    Ok(())
531}
532
533/// Compresses all input and returns the resulting Zstandard archive.
534pub fn encode_all<R: Read>(source: R, options: EncoderOptions) -> Result<Vec<u8>, EncodeError> {
535    let mut output = Vec::new();
536    encode(source, &mut output, options)?;
537    Ok(output)
538}
539
540#[cfg(test)]
541mod tests {
542    use super::*;
543    use crate::decoding::FrameDecoder;
544    use alloc::vec;
545
546    #[test]
547    fn multi_frame_stream_roundtrips() {
548        let input = b"abcdefgh".repeat(4096);
549        let options = EncoderOptions::new(CompressionLevel::FASTEST)
550            .with_frame_chunk_size(1024)
551            .with_memory_limit(16 * 1024 * 1024);
552        let compressed = encode_all(input.as_slice(), options).unwrap();
553        let mut decoded = Vec::with_capacity(input.len());
554        FrameDecoder::new()
555            .decode_all_to_vec(&compressed, &mut decoded)
556            .unwrap();
557        assert_eq!(decoded, input);
558    }
559
560    #[test]
561    fn empty_input_is_a_valid_frame() {
562        let compressed = encode_all(&[][..], EncoderOptions::default()).unwrap();
563        let mut decoded = Vec::with_capacity(1);
564        FrameDecoder::new()
565            .decode_all_to_vec(&compressed, &mut decoded)
566            .unwrap();
567        assert!(decoded.is_empty());
568    }
569
570    #[test]
571    fn uncompressed_stream_roundtrips() {
572        let input = vec![42; 300_000];
573        let options =
574            EncoderOptions::new(CompressionLevel::UNCOMPRESSED).with_frame_chunk_size(150_000);
575        let compressed = encode_all(input.as_slice(), options).unwrap();
576        assert_eq!(zstd::decode_all(compressed.as_slice()).unwrap(), input);
577    }
578
579    #[test]
580    fn archive_is_interoperable_with_the_c_decoder() {
581        let input = b"interoperable streaming input".repeat(10_000);
582        let options = EncoderOptions::new(CompressionLevel::DEFAULT)
583            .with_frame_chunk_size(32 * 1024)
584            .with_memory_limit(32 * 1024 * 1024);
585        let compressed = encode_all(input.as_slice(), options).unwrap();
586        assert_eq!(zstd::decode_all(compressed.as_slice()).unwrap(), input);
587    }
588
589    #[test]
590    fn default_options_keep_the_existing_specialized_frame_path() {
591        let input = b"unchanged default single-thread frame path".repeat(10_000);
592        let expected = c_port::encode_frame_no_dict(&input, CompressionLevel::DEFAULT.c_level());
593        let actual = encode_all(input.as_slice(), EncoderOptions::default()).unwrap();
594        assert_eq!(actual, expected);
595    }
596
597    #[cfg(feature = "hash")]
598    #[test]
599    fn checksummed_frames_roundtrip() {
600        let input = b"checksummed content".repeat(10_000);
601        let options = EncoderOptions::new(CompressionLevel::DEFAULT)
602            .with_frame_chunk_size(32 * 1024)
603            .with_memory_limit(32 * 1024 * 1024)
604            .with_checksum(true);
605        let compressed = encode_all(input.as_slice(), options).unwrap();
606        assert_eq!(zstd::decode_all(compressed.as_slice()).unwrap(), input);
607    }
608
609    #[test]
610    fn prepared_dictionary_stream_roundtrips() {
611        let dictionary_bytes = b"dictionary words and repeated prefixes for the test".repeat(8);
612        let input = dictionary_bytes.repeat(200);
613        let dictionary = EncoderDictionary::copy(&dictionary_bytes).unwrap();
614        let options = EncoderOptions::new(CompressionLevel::DEFAULT)
615            .with_frame_chunk_size(16 * 1024)
616            .with_memory_limit(32 * 1024 * 1024)
617            .with_dictionary(dictionary);
618        let compressed = encode_all(input.as_slice(), options).unwrap();
619
620        let mut decoded = Vec::with_capacity(input.len());
621        let mut decoder =
622            zstd::stream::read::Decoder::with_dictionary(compressed.as_slice(), &dictionary_bytes)
623                .unwrap();
624        decoder.read_to_end(&mut decoded).unwrap();
625        assert_eq!(decoded, input);
626    }
627
628    #[test]
629    fn too_short_dictionary_is_rejected_instead_of_silently_ignored() {
630        assert_eq!(
631            EncoderDictionary::copy(b"short").unwrap_err(),
632            DictionaryError::WrongDictionary
633        );
634    }
635
636    #[test]
637    fn invalid_budget_is_reported_before_writing() {
638        let options = EncoderOptions::new(CompressionLevel::MAXIMUM);
639        assert!(matches!(
640            Encoder::new(io::sink(), options),
641            Err(EncodeError::MemoryLimitExceeded { .. })
642        ));
643    }
644
645    #[test]
646    fn target_io_errors_are_preserved() {
647        struct FailingWriter;
648
649        impl Write for FailingWriter {
650            fn write(&mut self, _source: &[u8]) -> io::Result<usize> {
651                Err(io::Error::new(io::ErrorKind::BrokenPipe, "test failure"))
652            }
653
654            fn flush(&mut self) -> io::Result<()> {
655                Ok(())
656            }
657        }
658
659        let error = encode(
660            b"accepted input".as_slice(),
661            FailingWriter,
662            EncoderOptions::default(),
663        )
664        .unwrap_err();
665        assert!(matches!(
666            error,
667            EncodeError::Io(ref error) if error.kind() == io::ErrorKind::BrokenPipe
668        ));
669    }
670
671    #[test]
672    fn input_buffer_never_exceeds_the_configured_chunk() {
673        let options = EncoderOptions::new(CompressionLevel::FASTEST)
674            .with_frame_chunk_size(1024)
675            .with_memory_limit(16 * 1024 * 1024);
676        let mut encoder = Encoder::new(io::sink(), options).unwrap();
677        for _ in 0..1024 {
678            encoder.write_all(&[7; 4096]).unwrap();
679            assert!(encoder.input.len() <= 1024);
680            assert_eq!(encoder.input.capacity(), 1024);
681        }
682        encoder.finish().unwrap();
683    }
684
685    #[test]
686    fn large_logical_source_does_not_require_a_large_input_allocation() {
687        struct RepeatingReader {
688            remaining: usize,
689        }
690
691        impl Read for RepeatingReader {
692            fn read(&mut self, target: &mut [u8]) -> io::Result<usize> {
693                let count = self.remaining.min(target.len());
694                target[..count].fill(19);
695                self.remaining -= count;
696                Ok(count)
697            }
698        }
699
700        let options = EncoderOptions::new(CompressionLevel::UNCOMPRESSED)
701            .with_frame_chunk_size(1024 * 1024)
702            .with_memory_limit(4 * 1024 * 1024);
703        encode(
704            RepeatingReader {
705                remaining: 64 * 1024 * 1024,
706            },
707            io::sink(),
708            options,
709        )
710        .unwrap();
711    }
712
713    #[test]
714    fn negative_fast_levels_interoperate() {
715        let input = b"fast acceleration content".repeat(20_000);
716        let level = CompressionLevel::fast(8).unwrap();
717        let compressed = encode_all(input.as_slice(), EncoderOptions::new(level)).unwrap();
718        assert_eq!(zstd::decode_all(compressed.as_slice()).unwrap(), input);
719    }
720
721    #[test]
722    fn typed_tuning_changes_real_compressor_parameters() {
723        let input = b"typed compressor tuning".repeat(20_000);
724        let tuning = CompressionTuning::new()
725            .with_strategy(super::super::CompressionStrategy::Greedy)
726            .with_target_compressed_block_size(16 * 1024);
727        let options = EncoderOptions::default().with_tuning(tuning);
728        let compressed = encode_all(input.as_slice(), options).unwrap();
729        assert_eq!(zstd::decode_all(compressed.as_slice()).unwrap(), input);
730    }
731
732    #[test]
733    fn omitted_content_size_interoperates() {
734        let input = b"content size omitted".repeat(10_000);
735        let options = EncoderOptions::default().with_content_size_policy(ContentSizePolicy::Omit);
736        let compressed = encode_all(input.as_slice(), options).unwrap();
737        let (header, _) = crate::decoding::frame::read_frame_header(compressed.as_slice()).unwrap();
738        assert!(!header.descriptor.single_segment_flag());
739        assert_eq!(header.frame_content_size(), 0);
740        assert_eq!(zstd::decode_all(compressed.as_slice()).unwrap(), input);
741    }
742
743    #[test]
744    fn pledged_size_must_match_exactly() {
745        let options = EncoderOptions::default().with_pledged_source_size(4);
746        let error = encode_all(b"abc".as_slice(), options).unwrap_err();
747        assert!(matches!(
748            error,
749            EncodeError::PledgedSourceSizeMismatch {
750                pledged: 4,
751                actual: 3
752            }
753        ));
754    }
755}
zstd_complete/encoding/streaming_encoder.rs

zstd_complete/encoding/
streaming_encoder.rs
