v_queue 0.3.1

simple file based queue
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

use crate::compress::writer::dict_path;
use zstd::bulk::Decompressor;

// Lazily loaded per-part dictionary used to decompress v3 records.
pub struct PartDictionary {
    dict: Option<Vec<u8>>,
    decompressor: Option<Decompressor<'static>>,
    part_id: Option<u32>,
}

impl PartDictionary {
    pub fn new() -> Self {
        PartDictionary {
            dict: None,
            decompressor: None,
            part_id: None,
        }
    }

    pub fn ensure_loaded(&mut self, base_path: &str, queue_name: &str, part_id: u32) -> bool {
        if self.part_id == Some(part_id) && self.decompressor.is_some() {
            return true;
        }

        let path = dict_path(base_path, queue_name, part_id);
        let dict = match std::fs::read(&path) {
            Ok(d) => d,
            Err(e) => {
                error!("[queue:consumer] fail read dictionary {}, err={}", path, e);
                return false;
            },
        };

        match Decompressor::with_dictionary(&dict) {
            Ok(d) => {
                self.decompressor = Some(d);
                self.dict = Some(dict);
                self.part_id = Some(part_id);
                true
            },
            Err(e) => {
                error!("[queue:consumer] fail build decompressor, err={}", e);
                false
            },
        }
    }

    pub fn decompress(&mut self, compressed: &[u8], out: &mut [u8]) -> Result<usize, ()> {
        match self.decompressor.as_mut() {
            Some(d) => d.decompress_to_buffer(compressed, out).map_err(|_| ()),
            None => Err(()),
        }
    }
}