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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
|
use super::common::*;
use rev_buf_reader::RevBufReader;
use std::fs::{self};
use std::io::{self, BufRead, Read, Seek, SeekFrom};
/// There are three special characters that need to be handled:
/// Here: SC = escape char, FS = field separator.
/// - FS FS SC -> actual record separator
/// - SC FS SC -> literal FS
/// - SC SC SC -> literal SC
fn validate_special(buf: &[u8]) -> Option<SpecialSequence> {
match buf {
SEQ_RECORD_SEP => Some(SpecialSequence::RecordSeparator),
SEQ_LIT_FIELD_SEP => Some(SpecialSequence::LiteralFieldSeparator),
SEQ_LIT_ESCAPE => Some(SpecialSequence::LiteralEscape),
_ => None,
}
}
pub struct ReverseLogReader<'a> {
rev_reader: RevBufReader<&'a mut fs::File>,
}
impl<'a> ReverseLogReader<'a> {
pub fn new(file: &mut fs::File) -> Result<ReverseLogReader, io::Error> {
let rev_reader = RevBufReader::new(file);
Ok(ReverseLogReader { rev_reader })
}
fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
if self.rev_reader.stream_position()? == 0 {
return Ok(None);
}
// Check that the record starts with the record separator
if self.read_special_sequence()? != SpecialSequence::RecordSeparator {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Record candidate does not end with record separator",
));
}
// The buffer that stores all the bytes of the record read so far in reverse order.
let mut result_buf: Vec<u8> = Vec::new();
// The buffer that stores the bytes read from the file.
let mut read_buf: Vec<u8> = Vec::new();
loop {
read_buf.clear();
self.rev_reader
.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
result_buf.extend(read_buf.iter().rev());
if self.rev_reader.stream_position()? == 0 {
// If we've reached the beginning of the file, we've read the entire record.
break;
}
// Otherwise, we must have encountered an escape character.
match self.read_special_sequence()? {
SpecialSequence::RecordSeparator => {
// The record is complete, so we can break out of the loop.
// Move the cursor back to the beginning of the special sequence.
self.rev_reader.seek_relative(SEQ_RECORD_SEP.len() as i64)?;
break;
}
SpecialSequence::LiteralFieldSeparator => {
// The field separator is escaped, so we need to add it to the result buffer.
result_buf.push(FIELD_SEPARATOR);
}
SpecialSequence::LiteralEscape => {
// The escape character is escaped, so we need to add it to the result buffer.
result_buf.push(ESCAPE_CHARACTER);
}
}
}
result_buf.reverse();
let record = Record::deserialize(&result_buf);
Ok(Some(record))
}
fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
self.rev_reader.read_exact(&mut special_buf)?;
match validate_special(&special_buf.as_slice()) {
Some(special) => Ok(special),
None => Err(io::Error::new(
io::ErrorKind::InvalidData,
"Not a special sequence",
)),
}
}
}
impl Iterator for ReverseLogReader<'_> {
type Item = Record;
fn next(&mut self) -> Option<Self::Item> {
match self.read_record() {
Ok(Some(record)) => Some(record),
Ok(None) => None,
Err(err) => panic!("Error reading record: {:?}", err),
}
}
}
pub struct ForwardLogReader<'a> {
reader: io::BufReader<&'a mut fs::File>,
}
impl<'a> ForwardLogReader<'a> {
pub fn new(file: &mut fs::File) -> ForwardLogReader {
let reader = io::BufReader::new(file);
ForwardLogReader { reader }
}
fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
// The buffer that stores the bytes read from the file.
let mut read_buf: Vec<u8> = Vec::new();
// The buffer that stores all the bytes of the record read so far in reverse order.
let mut result_buf: Vec<u8> = Vec::new();
// Try reading a byte from the file.
// If we've reached the end of the file, return None.
let mut peek_buf = vec![0, 1];
match self.reader.read_exact(&mut peek_buf) {
Ok(_) => {
// Go back one byte (not sure why you need to seek by -2 here?)
self.reader.seek_relative(-2)?;
}
Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => {
return Ok(None);
}
Err(e) => {
return Err(e);
}
}
loop {
read_buf.clear();
self.reader.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
self.reader.seek_relative(-1)?;
result_buf.extend(&read_buf[..read_buf.len() - 1]);
// Otherwise, we must have encountered an escape character.
match self.read_special_sequence()? {
SpecialSequence::RecordSeparator => {
// The record is complete, so we can break out of the loop.
break;
}
SpecialSequence::LiteralFieldSeparator => {
// The field separator is escaped, so we need to add it to the result buffer.
result_buf.push(FIELD_SEPARATOR);
}
SpecialSequence::LiteralEscape => {
// The escape character is escaped, so we need to add it to the result buffer.
result_buf.push(ESCAPE_CHARACTER);
}
}
}
let record = Record::deserialize(&result_buf);
Ok(Some(record))
}
fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
self.reader.read_exact(&mut special_buf)?;
match validate_special(&special_buf.as_slice()) {
Some(special) => Ok(special),
None => Err(io::Error::new(
io::ErrorKind::InvalidData,
"Not a special sequence",
)),
}
}
}
impl Iterator for ForwardLogReader<'_> {
type Item = Record;
fn next(&mut self) -> Option<Self::Item> {
match self.read_record() {
Ok(Some(record)) => Some(record),
Ok(None) => None,
Err(err) => panic!("Error reading record: {:?}", err),
}
}
}
|
