use std::io::{Read, Seek, SeekFrom};
use crate::error::Result;
use crate::usn::{parse_usn_record_v2, parse_usn_record_v3, UsnRecord};
const BUF_SIZE: usize = 64 * 1024;
fn read_u32_le(data: &[u8], offset: usize) -> u32 {
let mut b = [0u8; 4];
if let Some(s) = data.get(offset..offset + 4) {
b.copy_from_slice(s);
}
u32::from_le_bytes(b)
}
fn read_u16_le(data: &[u8], offset: usize) -> u16 {
let mut b = [0u8; 2];
if let Some(s) = data.get(offset..offset + 2) {
b.copy_from_slice(s);
}
u16::from_le_bytes(b)
}
pub struct UsnJournalReader<R: Read + Seek> {
reader: R,
buf: Vec<u8>,
buf_len: usize,
buf_offset: usize,
stream_pos: u64,
total_size: u64,
done: bool,
}
impl<R: Read + Seek> UsnJournalReader<R> {
pub fn new(mut reader: R) -> Result<Self> {
let total_size = reader.seek(SeekFrom::End(0))?;
reader.seek(SeekFrom::Start(0))?;
Ok(Self {
reader,
buf: vec![0u8; BUF_SIZE],
buf_len: 0,
buf_offset: 0,
stream_pos: 0,
total_size,
done: false,
})
}
fn fill_buffer(&mut self) -> Result<bool> {
if self.stream_pos >= self.total_size {
self.done = true;
return Ok(false);
}
if self.buf_offset > 0 && self.buf_offset < self.buf_len {
let remaining = self.buf_len - self.buf_offset;
self.buf.copy_within(self.buf_offset..self.buf_len, 0);
self.buf_len = remaining;
} else {
self.buf_len = 0;
}
self.buf_offset = 0;
let space = BUF_SIZE - self.buf_len;
if space > 0 {
let Some(dst) = self.buf.get_mut(self.buf_len..self.buf_len + space) else {
self.done = true; return Ok(self.buf_len > 0); };
let n = self.reader.read(dst)?;
if n == 0 {
self.done = true; return Ok(self.buf_len > 0); }
self.buf_len += n;
self.stream_pos += n as u64;
}
Ok(true)
}
fn skip_zeros(&mut self) -> Result<bool> {
loop {
while self.buf_offset + 8 <= self.buf_len {
match self.buf.get(self.buf_offset..self.buf_offset + 8) {
Some([0, 0, 0, 0, 0, 0, 0, 0]) => self.buf_offset += 8,
_ => return Ok(true),
}
}
if !self.fill_buffer()? {
return Ok(false);
}
if self.buf_len == 0 {
return Ok(false); }
}
}
}
impl<R: Read + Seek> Iterator for UsnJournalReader<R> {
type Item = Result<UsnRecord>;
fn next(&mut self) -> Option<Self::Item> {
if self.done {
return None;
}
if self.buf_offset >= self.buf_len {
match self.fill_buffer() {
Ok(true) => {}
Ok(false) => return None,
Err(e) => return Some(Err(e)),
}
}
match self.skip_zeros() {
Ok(true) => {}
Ok(false) => return None,
Err(e) => return Some(Err(e)),
}
if self.buf_offset + 8 > self.buf_len {
match self.fill_buffer() {
Ok(true) if self.buf_offset + 8 <= self.buf_len => {} _ => return None, }
}
let record_len = read_u32_le(&self.buf, self.buf_offset) as usize;
if !(8..=65536).contains(&record_len) {
self.buf_offset += 8;
return self.next();
}
if self.buf_offset + record_len > self.buf_len {
match self.fill_buffer() {
Ok(true) if self.buf_offset + record_len <= self.buf_len => {}
_ => {
self.buf_offset += 8;
return self.next();
}
}
}
let version = read_u16_le(&self.buf, self.buf_offset + 4);
let Some(record_data) = self.buf.get(self.buf_offset..self.buf_offset + record_len) else {
self.buf_offset += 8; return self.next(); };
let record_data = record_data.to_vec();
let aligned = (record_len + 7) & !7;
self.buf_offset += aligned;
match version {
2 => match parse_usn_record_v2(&record_data) {
Ok(r) => Some(Ok(r)),
Err(_) => self.next(),
},
3 => match parse_usn_record_v3(&record_data) {
Ok(r) => Some(Ok(r)),
Err(_) => self.next(),
},
_ => self.next(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::usn::UsnReason;
use std::io::Cursor;
fn build_v2_record_bytes(
entry: u64,
seq: u16,
parent: u64,
parent_seq: u16,
reason: u32,
name: &str,
) -> Vec<u8> {
let name_utf16: Vec<u16> = name.encode_utf16().collect();
let name_bytes_len = name_utf16.len() * 2;
let record_len = 0x3C + name_bytes_len;
let aligned_len = (record_len + 7) & !7;
let mut buf = vec![0u8; aligned_len];
buf[0..4].copy_from_slice(&(record_len as u32).to_le_bytes());
buf[4..6].copy_from_slice(&2u16.to_le_bytes());
let file_ref = entry | (u64::from(seq) << 48);
buf[0x08..0x10].copy_from_slice(&file_ref.to_le_bytes());
let parent_ref = parent | (u64::from(parent_seq) << 48);
buf[0x10..0x18].copy_from_slice(&parent_ref.to_le_bytes());
buf[0x18..0x20].copy_from_slice(&100i64.to_le_bytes());
let ts: i64 = 133_500_480_000_000_000;
buf[0x20..0x28].copy_from_slice(&ts.to_le_bytes());
buf[0x28..0x2C].copy_from_slice(&reason.to_le_bytes());
buf[0x34..0x38].copy_from_slice(&0x20u32.to_le_bytes());
buf[0x38..0x3A].copy_from_slice(&(name_bytes_len as u16).to_le_bytes());
buf[0x3A..0x3C].copy_from_slice(&0x3Cu16.to_le_bytes());
for (i, &ch) in name_utf16.iter().enumerate() {
let off = 0x3C + i * 2;
buf[off..off + 2].copy_from_slice(&ch.to_le_bytes());
}
buf
}
#[test]
fn test_streaming_reader_basic() {
let r = build_v2_record_bytes(100, 1, 5, 5, 0x100, "test.txt");
let cursor = Cursor::new(r);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "test.txt");
}
#[test]
fn test_streaming_reader_skips_zeros() {
let mut data = vec![0u8; 4096];
data.extend_from_slice(&build_v2_record_bytes(100, 1, 5, 5, 0x100, "found.txt"));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "found.txt");
}
#[test]
fn test_streaming_reader_multiple() {
let mut data = Vec::new();
data.extend_from_slice(&build_v2_record_bytes(100, 1, 5, 5, 0x100, "a.txt"));
data.extend_from_slice(&build_v2_record_bytes(200, 1, 100, 1, 0x200, "b.txt"));
data.extend_from_slice(&build_v2_record_bytes(300, 1, 100, 1, 0x100, "c.txt"));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 3);
}
#[test]
fn test_streaming_reader_empty_data() {
let cursor = Cursor::new(Vec::<u8>::new());
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 0);
}
#[test]
fn test_streaming_reader_all_zeros() {
let data = vec![0u8; 4096];
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 0);
}
#[test]
fn test_streaming_reader_includes_close_only() {
let data = build_v2_record_bytes(100, 1, 5, 5, 0x8000_0000, "closed.txt");
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].reason, UsnReason::CLOSE);
}
#[test]
fn test_streaming_reader_invalid_record_length() {
let mut data = vec![0u8; 64];
data[0..4].copy_from_slice(&3u32.to_le_bytes()); data[4..6].copy_from_slice(&2u16.to_le_bytes());
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 0);
}
#[test]
fn test_streaming_reader_invalid_then_valid() {
let mut data = vec![0u8; 16]; data[0..4].copy_from_slice(&5u32.to_le_bytes()); data[4..6].copy_from_slice(&99u16.to_le_bytes()); data.resize(16, 0);
data.extend_from_slice(&[0u8; 64]);
data.extend_from_slice(&build_v2_record_bytes(100, 1, 5, 5, 0x100, "valid.txt"));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "valid.txt");
}
#[test]
fn test_streaming_reader_unknown_version() {
let mut data = vec![0u8; 0x40];
data[0..4].copy_from_slice(&(0x40u32).to_le_bytes());
data[4..6].copy_from_slice(&99u16.to_le_bytes());
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 0);
}
fn build_v3_record_bytes(entry: u64, parent: u64, reason: u32, name: &str) -> Vec<u8> {
let name_utf16: Vec<u16> = name.encode_utf16().collect();
let name_bytes_len = name_utf16.len() * 2;
let record_len = 0x4C + name_bytes_len;
let aligned_len = (record_len + 7) & !7;
let mut buf = vec![0u8; aligned_len];
buf[0..4].copy_from_slice(&(record_len as u32).to_le_bytes());
buf[4..6].copy_from_slice(&3u16.to_le_bytes());
buf[6..8].copy_from_slice(&0u16.to_le_bytes());
buf[0x08..0x18].copy_from_slice(&u128::from(entry).to_le_bytes());
buf[0x18..0x28].copy_from_slice(&u128::from(parent).to_le_bytes());
buf[0x28..0x30].copy_from_slice(&200i64.to_le_bytes());
let ts: i64 = 133_500_480_000_000_000;
buf[0x30..0x38].copy_from_slice(&ts.to_le_bytes());
buf[0x38..0x3C].copy_from_slice(&reason.to_le_bytes());
buf[0x44..0x48].copy_from_slice(&0x20u32.to_le_bytes());
buf[0x48..0x4A].copy_from_slice(&(name_bytes_len as u16).to_le_bytes());
buf[0x4A..0x4C].copy_from_slice(&0x4Cu16.to_le_bytes());
for (i, &ch) in name_utf16.iter().enumerate() {
let off = 0x4C + i * 2;
buf[off..off + 2].copy_from_slice(&ch.to_le_bytes());
}
buf
}
#[test]
fn test_streaming_reader_v3_record() {
let data = build_v3_record_bytes(100, 5, 0x100, "v3file.txt");
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "v3file.txt");
assert_eq!(records[0].major_version, 3);
}
#[test]
fn test_streaming_reader_v3_close_only_included() {
let data = build_v3_record_bytes(100, 5, 0x8000_0000, "closed_v3.txt");
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].reason, UsnReason::CLOSE);
}
#[test]
fn test_streaming_reader_large_zero_gap() {
let mut data = vec![0u8; 128 * 1024]; data.extend_from_slice(&build_v2_record_bytes(100, 1, 5, 5, 0x100, "deep.txt"));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "deep.txt");
}
#[test]
fn test_streaming_reader_record_larger_than_initial_buffer_fill() {
let record = build_v2_record_bytes(42, 3, 5, 5, 0x100, "buffer_test.txt");
let cursor = Cursor::new(record);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].mft_entry, 42);
assert_eq!(records[0].mft_sequence, 3);
}
#[test]
fn test_streaming_reader_record_too_large() {
let mut data = vec![0u8; 128];
data[0..4].copy_from_slice(&(65537u32).to_le_bytes());
data[4..6].copy_from_slice(&2u16.to_le_bytes());
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 0);
}
#[test]
fn test_streaming_reader_mixed_v2_v3() {
let mut data = Vec::new();
data.extend_from_slice(&build_v2_record_bytes(100, 1, 5, 5, 0x100, "v2.txt"));
data.extend_from_slice(&build_v3_record_bytes(200, 5, 0x200, "v3.txt"));
data.extend_from_slice(&build_v2_record_bytes(300, 1, 5, 5, 0x100, "v2b.txt"));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 3);
assert_eq!(records[0].major_version, 2);
assert_eq!(records[1].major_version, 3);
assert_eq!(records[2].major_version, 2);
}
#[test]
fn test_streaming_reader_fill_buffer_with_unconsumed_data() {
let mut data = Vec::new();
let record_size;
{
let sample = build_v2_record_bytes(1, 1, 5, 5, 0x100, "sample.txt");
record_size = sample.len();
}
let num_records_to_fill = (BUF_SIZE - record_size) / record_size;
for i in 0..num_records_to_fill {
data.extend_from_slice(&build_v2_record_bytes(
(i + 1) as u64,
1,
5,
5,
0x100,
&format!("f{i:04}.txt"),
));
}
let remaining = BUF_SIZE - (num_records_to_fill * record_size);
if remaining > 0 && remaining < record_size {
data.extend_from_slice(&vec![0u8; remaining]); }
for i in 0..5 {
data.extend_from_slice(&build_v2_record_bytes(
(num_records_to_fill + i + 1) as u64,
1,
5,
5,
0x100,
&format!("after{i}.txt"),
));
}
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert!(records.len() >= num_records_to_fill + 5);
}
#[test]
fn test_streaming_reader_record_at_exact_buffer_boundary() {
let sample = build_v2_record_bytes(1, 1, 5, 5, 0x100, "sample.txt");
let record_size = sample.len();
let mut data = Vec::new();
let records_per_buffer = BUF_SIZE / record_size;
let exact_fill = records_per_buffer * record_size;
for i in 0..records_per_buffer {
data.extend_from_slice(&build_v2_record_bytes(
(i + 1) as u64,
1,
5,
5,
0x100,
"exact.txt",
));
}
if exact_fill < BUF_SIZE {
data.extend_from_slice(&vec![0u8; BUF_SIZE - exact_fill]);
}
data.extend_from_slice(&build_v2_record_bytes(
(records_per_buffer + 1) as u64,
1,
5,
5,
0x100,
"boundary.txt",
));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert!(records.iter().any(|r| r.filename == "boundary.txt"));
}
#[test]
fn test_streaming_reader_record_straddles_buffer() {
let sample = build_v2_record_bytes(1, 1, 5, 5, 0x100, "sample.txt");
let record_size = sample.len();
let mut data = Vec::new();
let records_to_fill = (BUF_SIZE / record_size) - 1;
for i in 0..records_to_fill {
data.extend_from_slice(&build_v2_record_bytes(
(i + 1) as u64,
1,
5,
5,
0x100,
"fill.txt",
));
}
let current_len = data.len();
let padding = BUF_SIZE - current_len - (record_size / 2);
if padding > 0 {
data.extend_from_slice(&vec![0u8; padding]);
}
data.extend_from_slice(&build_v2_record_bytes(999, 1, 5, 5, 0x100, "straddle.txt"));
data.extend_from_slice(&vec![0u8; 256]);
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert!(records.iter().any(|r| r.filename == "straddle.txt"));
}
#[test]
fn test_streaming_reader_data_larger_than_buffer() {
let mut data = Vec::new();
let total_records = 2000; for i in 0..total_records {
data.extend_from_slice(&build_v2_record_bytes(
(i + 1) as u64,
1,
5,
5,
0x100,
&format!("r{i:04}.txt"),
));
}
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), total_records);
}
struct ErrorAfterNReads {
data: Cursor<Vec<u8>>,
reads_remaining: usize,
}
impl ErrorAfterNReads {
fn new(data: Vec<u8>, successful_reads: usize) -> Self {
Self {
data: Cursor::new(data),
reads_remaining: successful_reads,
}
}
}
impl Read for ErrorAfterNReads {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if self.reads_remaining == 0 {
return Err(std::io::Error::other("simulated read error"));
}
self.reads_remaining -= 1;
self.data.read(buf)
}
}
impl Seek for ErrorAfterNReads {
fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
self.data.seek(pos)
}
}
struct TinyChunkReader {
data: Vec<u8>,
pos: u64,
chunk_size: usize,
}
impl TinyChunkReader {
fn new(data: Vec<u8>, chunk_size: usize) -> Self {
Self {
data,
pos: 0,
chunk_size,
}
}
}
impl Read for TinyChunkReader {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let remaining = self.data.len() - self.pos as usize;
if remaining == 0 {
return Ok(0); }
let to_read = buf.len().min(self.chunk_size).min(remaining);
let start = self.pos as usize;
buf[..to_read].copy_from_slice(&self.data[start..start + to_read]);
self.pos += to_read as u64;
Ok(to_read)
}
}
impl Seek for TinyChunkReader {
fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
match pos {
SeekFrom::Start(n) => self.pos = n,
SeekFrom::End(n) => self.pos = (self.data.len() as i64 + n) as u64,
SeekFrom::Current(n) => self.pos = (self.pos as i64 + n) as u64, }
Ok(self.pos)
}
}
#[test]
fn test_streaming_reader_done_flag_returns_none() {
let record = build_v2_record_bytes(100, 1, 5, 5, 0x100, "done.txt");
let cursor = Cursor::new(record);
let mut reader = UsnJournalReader::new(cursor).unwrap();
let first = reader.next();
assert!(first.is_some());
assert!(first.unwrap().is_ok());
let second = reader.next();
assert!(second.is_none());
let third = reader.next();
assert!(third.is_none());
}
#[test]
fn test_streaming_reader_fill_buffer_error_propagation() {
let record = build_v2_record_bytes(100, 1, 5, 5, 0x100, "err.txt");
let err_reader = ErrorAfterNReads::new(record, 0);
let mut reader = UsnJournalReader::new(err_reader).unwrap();
let result = reader.next();
assert!(result.is_some());
let err = result.unwrap();
assert!(err.is_err());
assert!(err
.unwrap_err()
.to_string()
.contains("simulated read error"));
}
#[test]
fn test_streaming_reader_skip_zeros_error_propagation() {
let mut data = vec![0u8; BUF_SIZE]; data.extend_from_slice(&[1, 2, 3, 4, 5, 6, 7, 8]);
let err_reader = ErrorAfterNReads::new(data, 1);
let mut reader = UsnJournalReader::new(err_reader).unwrap();
let result = reader.next();
assert!(result.is_some());
let err = result.unwrap();
assert!(err.is_err());
}
#[test]
fn test_streaming_reader_eof_mid_fill_with_remaining_data() {
let record = build_v2_record_bytes(42, 1, 5, 5, 0x100, "tiny.txt");
let data_len = record.len();
let tiny_reader = TinyChunkReader::new(record, data_len);
let mut reader = UsnJournalReader::new(tiny_reader).unwrap();
let result = reader.next();
assert!(result.is_some());
let rec = result.unwrap().unwrap();
assert_eq!(rec.filename, "tiny.txt");
}
#[test]
fn test_streaming_reader_eof_mid_fill_no_remaining_data() {
let tiny_reader = TinyChunkReader::new(Vec::new(), 1);
let mut reader = UsnJournalReader::new(tiny_reader).unwrap();
let result = reader.next();
assert!(result.is_none());
}
#[test]
fn test_streaming_reader_header_refill_insufficient() {
let mut data = vec![0u8; BUF_SIZE - 4]; data.extend_from_slice(&[0xAA, 0xBB, 0xCC, 0xDD]);
let cursor = Cursor::new(data);
let mut reader = UsnJournalReader::new(cursor).unwrap();
let result = reader.next();
assert!(result.is_none());
}
#[test]
fn test_streaming_reader_record_refill_insufficient() {
let mut data = vec![0u8; 16];
data[0..4].copy_from_slice(&(1024u32).to_le_bytes());
data[4..6].copy_from_slice(&2u16.to_le_bytes());
let cursor = Cursor::new(data);
let mut reader = UsnJournalReader::new(cursor).unwrap();
let result = reader.next();
assert!(result.is_none());
}
#[test]
fn test_streaming_reader_v2_parse_error_skips() {
let mut data = Vec::new();
let mut bad_v2 = vec![0u8; 0x20]; bad_v2[0..4].copy_from_slice(&(0x20u32).to_le_bytes()); bad_v2[4..6].copy_from_slice(&2u16.to_le_bytes()); data.extend_from_slice(&bad_v2);
data.extend_from_slice(&build_v2_record_bytes(
100,
1,
5,
5,
0x100,
"after_bad_v2.txt",
));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "after_bad_v2.txt");
}
#[test]
fn test_streaming_reader_v3_parse_error_skips() {
let mut data = Vec::new();
let mut bad_v3 = vec![0u8; 0x20];
bad_v3[0..4].copy_from_slice(&(0x20u32).to_le_bytes());
bad_v3[4..6].copy_from_slice(&3u16.to_le_bytes()); data.extend_from_slice(&bad_v3);
data.extend_from_slice(&build_v2_record_bytes(
200,
1,
5,
5,
0x200,
"after_bad_v3.txt",
));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "after_bad_v3.txt");
}
#[test]
fn test_streaming_reader_skip_zeros_refill_then_find_data() {
let mut data = vec![0u8; BUF_SIZE * 2 + 512]; data.extend_from_slice(&build_v2_record_bytes(
100,
1,
5,
5,
0x100,
"after_many_zeros.txt",
));
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 1);
assert_eq!(records[0].filename, "after_many_zeros.txt");
}
#[test]
fn test_streaming_reader_skip_zeros_all_zeros_eof() {
let data = vec![0u8; BUF_SIZE + 100];
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert_eq!(records.len(), 0);
}
#[test]
fn test_streaming_reader_record_straddles_buffer_refill_fails() {
let sample = build_v2_record_bytes(1, 1, 5, 5, 0x100, "fill.txt");
let record_size = sample.len();
let mut data = Vec::new();
let records_to_fill = (BUF_SIZE / record_size) - 1;
for i in 0..records_to_fill {
data.extend_from_slice(&build_v2_record_bytes(
(i + 1) as u64,
1,
5,
5,
0x100,
"fill.txt",
));
}
let current_len = data.len();
let remaining_in_buffer = BUF_SIZE - current_len;
if remaining_in_buffer > 16 {
data.extend_from_slice(&vec![0u8; remaining_in_buffer - 16]);
}
let mut truncated_header = vec![0u8; 16];
truncated_header[0..4].copy_from_slice(&(4096u32).to_le_bytes()); truncated_header[4..6].copy_from_slice(&2u16.to_le_bytes()); truncated_header[8] = 0xFF; data.extend_from_slice(&truncated_header);
let cursor = Cursor::new(data);
let reader = UsnJournalReader::new(cursor).unwrap();
let records: Vec<_> = reader.filter_map(std::result::Result::ok).collect();
assert!(records.len() >= records_to_fill);
}
}