tale_ndjson/readers/
buffered.rs

1//! A vanilla buffered file reader.
2
3use std::fs::File;
4use std::io::{BufRead, BufReader, Seek, SeekFrom};
5use std::path::Path;
6
7use super::FileProcessor;
8use crate::errors::TaleError;
9
10/// Standard buffered file reader implementation
11pub struct BufferedFileProcessor {
12    reader: BufReader<File>,
13    file_size: u64,
14    current_position: u64,
15}
16
17impl BufferedFileProcessor {
18    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self, TaleError> {
19        let mut file = File::open(&path)?;
20        let file_size = file.seek(SeekFrom::End(0))?;
21        file.seek(SeekFrom::Start(0))?;
22
23        let reader = BufReader::new(file);
24
25        Ok(Self {
26            reader,
27            file_size,
28            current_position: 0,
29        })
30    }
31}
32
33impl FileProcessor for BufferedFileProcessor {
34    fn process_lines<F>(&mut self, mut line_processor: F) -> Result<(), TaleError>
35    where
36        F: FnMut(&str) -> Result<(), TaleError>,
37    {
38        let mut line = String::new();
39        while self.reader.read_line(&mut line)? > 0 {
40            // Remove trailing newline
41            if line.ends_with('\n') {
42                line.pop();
43                if line.ends_with('\r') {
44                    line.pop();
45                }
46            }
47
48            line_processor(&line)?;
49            line.clear();
50        }
51        Ok(())
52    }
53
54    fn skip_lines(&mut self, count: u64) -> Result<(), TaleError> {
55        let mut line = String::new();
56        for _ in 0..count {
57            if self.reader.read_line(&mut line)? == 0 {
58                break; // EOF
59            }
60            line.clear();
61        }
62        Ok(())
63    }
64
65    fn file_size(&self) -> u64 {
66        self.file_size
67    }
68
69    fn seek(&mut self, pos: SeekFrom) -> Result<u64, TaleError> {
70        let new_pos = self.reader.get_mut().seek(pos).map_err(TaleError::from)?;
71        self.current_position = new_pos;
72        Ok(new_pos)
73    }
74
75    fn position(&self) -> u64 {
76        self.current_position
77    }
78}
tale_ndjson/readers/buffered.rs

tale_ndjson/readers/
buffered.rs
