reduct-base 1.19.1

Base crate for ReductStore
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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use crate::error::ReductError;
use crate::io::{BoxedReadRecord, ReadRecord, RecordMeta};
use bytes::Bytes;
use std::collections::VecDeque;
use std::io::{Read, Seek, SeekFrom};

/// A record that reads data from an iterator of byte chunks
/// Each item from the iterator is a Result<Bytes, ReductError>
#[derive(Debug)]
pub struct IterRecord {
    inner: VecDeque<Result<Bytes, ReductError>>,
    meta: RecordMeta,
}

impl IterRecord {
    pub fn new<R: Iterator<Item = Result<Bytes, ReductError>>>(inner: R, meta: RecordMeta) -> Self {
        Self {
            inner: inner.collect(),
            meta,
        }
    }

    pub fn boxed<R: Iterator<Item = Result<Bytes, ReductError>> + 'static>(
        inner: R,
        meta: RecordMeta,
    ) -> BoxedReadRecord {
        Box::new(Self::new(inner, meta))
    }
}
impl Read for IterRecord {
    fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
        unimplemented!()
    }
}

impl Seek for IterRecord {
    fn seek(&mut self, _pos: SeekFrom) -> std::io::Result<u64> {
        unimplemented!()
    }
}

impl ReadRecord for IterRecord {
    fn read_chunk(&mut self) -> Option<Result<Bytes, ReductError>> {
        if self.inner.is_empty() {
            None
        } else {
            Some(self.inner.pop_front().unwrap())
        }
    }

    fn meta(&self) -> &RecordMeta {
        &self.meta
    }

    fn meta_mut(&mut self) -> &mut RecordMeta {
        &mut self.meta
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::io::BoxedReadRecord;
    use rstest::*;

    #[rstest]
    fn test_iter_record(mut record: BoxedReadRecord) {
        assert_eq!(record.meta().timestamp(), 123);
        assert_eq!(record.meta().content_type(), "text/plain");
        assert_eq!(record.meta().content_length(), 11);

        let chunk1 = record.read_chunk();
        assert!(chunk1.is_some());
        assert_eq!(chunk1.unwrap().unwrap(), Bytes::from("Hello "));

        let chunk2 = record.read_chunk();
        assert!(chunk2.is_some());
        assert_eq!(chunk2.unwrap().unwrap(), Bytes::from("World"));

        let chunk3 = record.read_chunk();
        assert!(chunk3.is_none());

        // for code coverage
        let _ = record.meta_mut();
    }

    #[rstest]
    #[should_panic(expected = "not implemented")]
    fn test_seek(record: BoxedReadRecord) {
        let mut record = record;
        let _ = record.seek(SeekFrom::Start(0));
    }

    #[rstest]
    #[should_panic(expected = "not implemented")]
    fn test_read(record: BoxedReadRecord) {
        let mut record = record;
        let mut buf = [0; 10];
        let _ = record.read(&mut buf);
    }

    #[fixture]
    fn record() -> BoxedReadRecord {
        let chunks = vec![Ok(Bytes::from("Hello ")), Ok(Bytes::from("World"))];
        let meta = RecordMeta::builder()
            .timestamp(123)
            .state(1)
            .content_type("text/plain".to_string())
            .content_length(11)
            .build();
        IterRecord::boxed(chunks.into_iter(), meta)
    }
}