kacrab-protocol 0.1.0

Kafka wire protocol types — generated from upstream message schemas by `kacrab-codegen`.
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

//! Gzip codec (`flate2` Rust backend).

use std::io::{Read, Write};

use super::{Compression, CompressionError, CompressionErrorKind, Result};

const DEFAULT_GZIP_LEVEL: u32 = 6;

/// Compress `payload` at the given level (`None` -> codec default `6`). Only the lower bound is
/// clamped (negative levels become `0`); higher values pass through to `flate2` unchanged.
pub fn compress_with_level(payload: &[u8], level: Option<i32>) -> Result<Vec<u8>> {
    let lvl = level.map_or(DEFAULT_GZIP_LEVEL, |l| l.max(0).cast_unsigned());
    let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::new(lvl));
    encoder
        .write_all(payload)
        .map_err(|e| encode_err(e.to_string()))?;
    encoder.finish().map_err(|e| encode_err(e.to_string()))
}

/// Decompress `payload`.
pub fn decompress(payload: &[u8]) -> Result<Vec<u8>> {
    let mut decoder = flate2::read::GzDecoder::new(payload);
    let mut output = Vec::new();
    let _read = decoder
        .read_to_end(&mut output)
        .map_err(|e| decode_err(e.to_string()))?;
    Ok(output)
}

const fn encode_err(message: String) -> CompressionError {
    CompressionError::new(
        Compression::Gzip,
        CompressionErrorKind::EncodeFailed { message },
    )
}

const fn decode_err(message: String) -> CompressionError {
    CompressionError::new(
        Compression::Gzip,
        CompressionErrorKind::DecodeFailed { message },
    )
}