Currently, I am building a custom StreamRecorder struct that records bytes read from a stream. While implementing it, I experimented with two different wrapping orders: putting the BufReader inside my struct versus wrapping my input reader in a BufReader inside my iterator function.
Intuitively, I expected that changing where the buffering happens would cause my recorder to log large prefetched blocks (like 8KiB) instead of clean atom boundaries when calling get_segment(). However, in my tests, both approaches behave identically in respect to getting the correct stream segments.
I would greatly appreciate it if someone could help clarify how BufReader works under the hood in these scenarios and why both approaches yield the same result based on my test.
The Two Approaches I Tried
1. What I am doing now (BufReader inside the struct):
pub struct StreamRecorder<R: Read> {
recording: BytesMut,
stream: BufReader<R>,
}
pub fn mp4_stream_to_atoms(mut input: impl Read) -> impl Iterator<Item anyhow::Result<Any>> {
std::iter::from_fn(move || match Option::<Any>::read_from(&mut input) {
Ok(Some(atom)) => Some(Ok(atom)),
Ok(None) => None,
Err(e) => {
eprintln!("\nIt returned an error: {e}");
None
}
})
}
2. What I initially did (BufReader wrapping the input inside the iterator function):
pub struct StreamRecorder<R: Read> {
recording: BytesMut,
stream: R,
}
pub fn mp4_stream_to_atoms(input: impl Read) -> impl Iterator<Item anyhow::Result<Any>> {
let mut buffered_stream = BufReader::new(input);
std::iter::from_fn(move || match Option::<Any>::read_from(&mut buffered_stream) {
Ok(Some(atom)) => Some(Ok(atom)),
Ok(None) => None,
Err(e) => {
eprintln!("\nIt returned an error: {e}");
None
}
})
}
The Struct Interface
impl<R: Read> StreamRecorder<R> {
// Supposed to retrieve stream segments alone (e.g., ftyp/moov or moof/mdat)
pub fn get_segment(&mut self) -> BytesMut {
self.recording.split()
}
}
impl<R: Read> Read for StreamRecorder<R> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let size = self.stream.read(buf)?;
self.recording.extend_from_slice(&buf[..size]);
Ok(size)
}
}
The Test
#[test]
fn test_stream_recorder_roundtrip() -> anyhow::Result<()> {
// Stream creation code...
let mut recorder = StreamRecorder::new(stdout);
let mut live_atoms = Vec::new();
{
let mut atom_iterator = mp4_stream_to_atoms(&mut recorder);
for _ in 0..3 {
if let Some(result) = atom_iterator.next() {
live_atoms.push(result?);
} else {
break;
}
}
}
assert!(
!live_atoms.is_empty(),
"Should have parsed at least one atom live"
);
let segment_bytes = recorder.get_segment();
// Re-parse the bytes from the recorder buffer using a Cursor
let cursor = std::io::Cursor::new(segment_bytes);
let recorded_atom_iterator = mp4_stream_to_atoms(cursor);
let mut recorded_atoms = Vec::new();
for result in recorded_atom_iterator {
recorded_atoms.push(result?);
}
// Verify count matches
assert_eq!(
live_atoms.len(),
recorded_atoms.len(),
"Atom count mismatch between live run and recorded segment"
);
// Verify round-trip serialization matches byte-for-byte
for (live_atom, recorded_atom) in live_atoms.into_iter().zip(recorded_atoms) {
let mut live_buf = Vec::new();
let mut recorded_buf = Vec::new();
atoms_to_mp4_stream(live_atom, &mut live_buf)?;
atoms_to_mp4_stream(recorded_atom, &mut recorded_buf)?;
assert_eq!(
live_buf, recorded_buf,
"Serialized atom bytes differ between live parse and recorded re-parse"
);
}
// Clean up...
Ok(())
}