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
use bytes::{Bytes, BytesMut};
use snap::raw::*;
use log::error;
use crate::protocol::buf::{ByteBuf, ByteBufMut};
use crate::protocol::{EncodeError, DecodeError};
use super::{Compressor, Decompressor};
pub struct Snappy;
impl<B: ByteBufMut> Compressor<B> for Snappy {
type BufMut = BytesMut;
fn compress<R, F>(buf: &mut B, f: F) -> Result<R, EncodeError>
where
F: FnOnce(&mut Self::BufMut) -> Result<R, EncodeError>
{
let mut tmp = BytesMut::new();
let res = f(&mut tmp)?;
let start_pos = buf.offset();
let compress_gap = buf.put_gap(max_compress_len(tmp.len()));
let actual_len = Encoder::new().compress(&tmp, buf.gap_buf(compress_gap)).map_err(|e| {
error!("Failed to compress buffer: {}", e);
EncodeError
})?;
buf.seek(start_pos + actual_len);
Ok(res)
}
}
impl<B: ByteBuf> Decompressor<B> for Snappy {
type Buf = Bytes;
fn decompress<R, F>(buf: &mut B, f: F) -> Result<R, DecodeError>
where
F: FnOnce(&mut Self::Buf) -> Result<R, DecodeError>
{
let buf = buf.copy_to_bytes(buf.remaining());
let actual_len = decompress_len(&buf).map_err(|e| {
error!("Failed to decompress buffer: {}", e);
DecodeError
})?;
let mut tmp = BytesMut::new();
tmp.resize(actual_len, 0);
Decoder::new().decompress(&buf, &mut tmp).map_err(|e| {
error!("Failed to decompress buffer: {}", e);
DecodeError
})?;
f(&mut tmp.into())
}
}