hayson_engine/buffer/
parser.rs

1use std::fs;
2use std::io;
3use std::path::Path;
4
5use crate::errors::HaysonCorruptLogError;
6
7use super::entry::{parse_buffer_entry, BufferEntry};
8
9#[derive(Debug)]
10pub enum BufferParseError {
11    Io(io::Error),
12    Corrupt(HaysonCorruptLogError),
13}
14
15impl From<io::Error> for BufferParseError {
16    fn from(value: io::Error) -> Self {
17        BufferParseError::Io(value)
18    }
19}
20
21impl From<HaysonCorruptLogError> for BufferParseError {
22    fn from(value: HaysonCorruptLogError) -> Self {
23        BufferParseError::Corrupt(value)
24    }
25}
26
27/// Reads every `*.jsonl` file directly inside `folder` (not recursive), in filename order, and
28/// every line inside each file in file order — this is the enumeration order later fed to `merge`
29/// before `buffered_at` sorting takes over.
30pub fn parse_buffer_folder(folder: &Path) -> Result<Vec<BufferEntry>, BufferParseError> {
31    let mut files: Vec<_> = fs::read_dir(folder)?
32        .filter_map(|entry| entry.ok())
33        .map(|entry| entry.path())
34        .filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("jsonl"))
35        .collect();
36    files.sort();
37
38    let mut entries = Vec::new();
39    for file in files {
40        let contents = fs::read_to_string(&file)?;
41        let path_label = file.to_string_lossy().into_owned();
42        for (line_no, line) in contents.lines().enumerate() {
43            if line.is_empty() {
44                continue;
45            }
46            entries.push(parse_buffer_entry(line, &path_label, line_no)?);
47        }
48    }
49    Ok(entries)
50}
51
52/// Reads every folder in `folders` via `parse_buffer_folder` and concatenates the results in the
53/// order the folders were given — `merge` is the one that imposes the real `buffered_at` order.
54pub fn parse_buffer_folders(folders: &[impl AsRef<Path>]) -> Result<Vec<BufferEntry>, BufferParseError> {
55    let mut entries = Vec::new();
56    for folder in folders {
57        entries.extend(parse_buffer_folder(folder.as_ref())?);
58    }
59    Ok(entries)
60}
61
62#[cfg(test)]
63mod tests {
64    use serde_json::{Map, Value};
65    use tempfile::tempdir;
66
67    use super::*;
68    use crate::buffer::entry::{serialize_buffer_entry, BufferEntry, BufferOp};
69
70    fn write_folder(dir: &Path, file_name: &str, entries: &[BufferEntry]) {
71        let lines: Vec<String> = entries.iter().map(serialize_buffer_entry).collect();
72        fs::write(dir.join(file_name), lines.join("\n") + "\n").unwrap();
73    }
74
75    fn put(table: &str, key: &str, source: &str, buffered_at: i64, name: &str) -> BufferEntry {
76        let mut data = Map::new();
77        data.insert("name".into(), Value::String(name.into()));
78        BufferEntry {
79            table: table.into(),
80            key: key.into(),
81            op: BufferOp::Put,
82            data: Some(data),
83            buffered_at,
84            source: source.into(),
85        }
86    }
87
88    #[test]
89    fn reads_two_folders_and_keeps_each_entrys_own_source() {
90        let dir_a = tempdir().unwrap();
91        let dir_b = tempdir().unwrap();
92        write_folder(
93            dir_a.path(),
94            "buf.jsonl",
95            &[put("users", "u1", "branch-a", 100, "Ania")],
96        );
97        write_folder(
98            dir_b.path(),
99            "buf.jsonl",
100            &[put("users", "u2", "branch-b", 200, "Piotr")],
101        );
102
103        let entries = parse_buffer_folders(&[dir_a.path(), dir_b.path()]).unwrap();
104
105        assert_eq!(entries.len(), 2);
106        assert_eq!(entries[0].source, "branch-a");
107        assert_eq!(entries[0].key, "u1");
108        assert_eq!(entries[1].source, "branch-b");
109        assert_eq!(entries[1].key, "u2");
110    }
111
112    #[test]
113    fn ignores_files_without_the_jsonl_extension() {
114        let dir = tempdir().unwrap();
115        write_folder(dir.path(), "buf.jsonl", &[put("users", "u1", "branch-a", 100, "Ania")]);
116        fs::write(dir.path().join("readme.txt"), "not a buffer file").unwrap();
117
118        let entries = parse_buffer_folder(dir.path()).unwrap();
119        assert_eq!(entries.len(), 1);
120    }
121}
hayson_engine/buffer/parser.rs

hayson_engine/buffer/
parser.rs
