use std::io::Read;
use crate::error::{ErrorDomain, ErrorLevel, Result, XmlError};
use crate::options::ParseOptions;
use crate::xml_bytes_reader::{BytesEvent, XmlBytesReader, XmlDeclInfo};
pub const DEFAULT_BUFFER_SIZE: usize = 10 * 1024 * 1024;
pub const HUGE_BUFFER_SIZE: usize = 1024 * 1024 * 1024;
const INITIAL_CAPACITY_WITHOUT_HINT: usize = 64 * 1024;
const BUF_CAPACITY_MULTIPLE: usize = 2;
pub struct XmlByteStreamReader<R: Read> {
inner: R,
buf: Vec<u8>,
buffer_size: usize,
eof: bool,
reader: XmlBytesReader<'static>,
_utf8_only_marker: (),
}
impl<R: Read> XmlByteStreamReader<R> {
pub fn new(inner: R, buffer_size: usize) -> Result<Self> {
Self::with_size_hint(inner, None, buffer_size)
}
pub fn with_size_hint(
mut inner: R,
size_hint: Option<usize>,
buffer_size: usize,
) -> Result<Self> {
let internal_cap = buffer_size.saturating_mul(BUF_CAPACITY_MULTIPLE);
let initial = size_hint
.map(|n| n.min(internal_cap))
.unwrap_or(INITIAL_CAPACITY_WITHOUT_HINT.min(internal_cap));
let mut buf = Vec::with_capacity(initial);
read_into_vec(&mut inner, &mut buf, initial)?;
let eof = buf.len() < initial;
if let Some(bom) = sniff_non_utf8_bom(&buf) {
return Err(XmlError::new(
ErrorDomain::Encoding,
ErrorLevel::Fatal,
format!(
"streaming reader: detected {bom} BOM, but streaming \
v1 supports UTF-8 input only. Use the slurped reader \
for non-UTF-8 encodings."
),
));
}
if buf.starts_with(&[0xEF, 0xBB, 0xBF]) {
buf.drain(..3);
}
let mut opts = ParseOptions::default();
opts.stream_owned_names = true;
const EMPTY: &[u8] = &[];
let mut reader = XmlBytesReader::from_bytes(EMPTY)?.with_options(opts);
unsafe {
reader.rebind_scanner(buf.as_ptr(), buf.len(), 0);
}
Ok(Self {
inner,
buf,
buffer_size,
eof,
reader,
_utf8_only_marker: (),
})
}
pub fn with_options(mut self, mut opts: ParseOptions) -> Self {
opts.stream_owned_names = true;
self.reader = self.reader.with_options(opts);
self
}
pub fn xml_decl(&self) -> Option<&XmlDeclInfo> {
self.reader.xml_decl()
}
pub fn recovered_errors(&self) -> &[XmlError] {
self.reader.recovered_errors()
}
pub fn next_event(&mut self) -> Result<BytesEvent<'_, '_>> {
self.ensure_runway()?;
self.reader.next()
}
pub fn validate(mut self) -> Result<()> {
while !matches!(self.next_event()?, BytesEvent::Eof) {}
Ok(())
}
fn ensure_runway(&mut self) -> Result<()> {
let cur_pos = self.reader.src_offset();
let runway = self.buf.len() - cur_pos;
if runway >= self.buffer_size || self.eof {
return Ok(());
}
if cur_pos > 0 {
self.buf.drain(..cur_pos);
}
let target_cap = self.buffer_size.saturating_mul(BUF_CAPACITY_MULTIPLE);
if self.buf.capacity() < target_cap {
let additional = target_cap - self.buf.len();
self.buf.reserve(additional);
}
while self.buf.len() < target_cap && !self.eof {
let space = target_cap - self.buf.len();
let n = read_chunk(&mut self.inner, &mut self.buf, space)?;
if n == 0 {
self.eof = true;
break;
}
}
unsafe {
self.reader.rebind_scanner(self.buf.as_ptr(), self.buf.len(), 0);
}
Ok(())
}
}
fn read_into_vec<R: Read>(reader: &mut R, buf: &mut Vec<u8>, wanted: usize) -> Result<usize> {
let start = buf.len();
let target = start + wanted;
buf.resize(target, 0);
let mut filled = start;
while filled < target {
let n = reader.read(&mut buf[filled..target]).map_err(io_to_xml_err)?;
if n == 0 { break; }
filled += n;
}
buf.truncate(filled);
Ok(filled - start)
}
fn read_chunk<R: Read>(reader: &mut R, buf: &mut Vec<u8>, space: usize) -> Result<usize> {
let start = buf.len();
buf.resize(start + space, 0);
let n = reader.read(&mut buf[start..]).map_err(io_to_xml_err)?;
buf.truncate(start + n);
Ok(n)
}
fn io_to_xml_err(e: std::io::Error) -> XmlError {
XmlError::new(
ErrorDomain::Parser,
ErrorLevel::Fatal,
format!("streaming reader I/O error: {e}"),
)
}
fn sniff_non_utf8_bom(buf: &[u8]) -> Option<&'static str> {
if buf.starts_with(&[0xFF, 0xFE, 0x00, 0x00]) { return Some("UTF-32 LE"); }
if buf.starts_with(&[0x00, 0x00, 0xFE, 0xFF]) { return Some("UTF-32 BE"); }
if buf.starts_with(&[0xFF, 0xFE]) { return Some("UTF-16 LE"); }
if buf.starts_with(&[0xFE, 0xFF]) { return Some("UTF-16 BE"); }
None
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Cursor;
fn validate(bytes: &[u8]) -> Result<()> {
XmlByteStreamReader::new(Cursor::new(bytes.to_vec()), DEFAULT_BUFFER_SIZE)?
.validate()
}
fn validate_with_buffer(bytes: &[u8], buffer_size: usize) -> Result<()> {
XmlByteStreamReader::new(Cursor::new(bytes.to_vec()), buffer_size)?
.validate()
}
#[test]
fn accepts_well_formed_xml() {
assert!(validate(b"<r/>").is_ok());
assert!(validate(b"<?xml version=\"1.0\"?><r><a/><b>text</b></r>").is_ok());
}
#[test]
fn rejects_malformed_xml() {
assert!(validate(b"<a><b></a>").is_err());
assert!(validate(b"<unclosed>").is_err());
assert!(validate(b"text without root").is_err());
}
#[test]
fn handles_tiny_buffer_with_many_refills() {
let mut doc = String::from("<root>");
for i in 0..200 {
doc.push_str(&format!("<item id=\"{i}\">value{i}</item>"));
}
doc.push_str("</root>");
assert!(validate_with_buffer(doc.as_bytes(), 1024).is_ok());
}
#[test]
fn text_content_larger_than_buffer_splits_across_events() {
let big_text = "x".repeat(10_000);
let doc = format!("<r>{big_text}</r>");
assert!(validate_with_buffer(doc.as_bytes(), 1024).is_ok());
}
#[test]
fn errors_on_element_name_larger_than_buffer() {
let big_name = "a".repeat(4000);
let doc = format!("<{big_name}/>");
let result = validate_with_buffer(doc.as_bytes(), 1024);
assert!(result.is_err(), "expected error on huge name, got Ok");
}
#[test]
fn accepts_text_at_buffer_size_boundary() {
let text = "x".repeat(8000);
let doc = format!("<r>{text}</r>");
assert!(validate_with_buffer(doc.as_bytes(), 16 * 1024).is_ok());
}
#[test]
fn rejects_utf16_le_bom_with_clear_error() {
let doc = vec![0xFF, 0xFE, 0x3C, 0x00, 0x72, 0x00, 0x2F, 0x00, 0x3E, 0x00];
let err = validate(&doc).unwrap_err();
assert!(err.message.contains("UTF-16 LE"), "got: {}", err.message);
assert!(err.message.contains("streaming"), "got: {}", err.message);
}
#[test]
fn rejects_utf16_be_bom_with_clear_error() {
let doc = vec![0xFE, 0xFF, 0x00, 0x3C, 0x00, 0x72, 0x00, 0x2F, 0x00, 0x3E];
let err = validate(&doc).unwrap_err();
assert!(err.message.contains("UTF-16 BE"), "got: {}", err.message);
}
#[test]
fn strips_utf8_bom_silently() {
let mut doc = vec![0xEF, 0xBB, 0xBF];
doc.extend_from_slice(b"<?xml version=\"1.0\" encoding=\"UTF-8\"?><r/>");
assert!(validate(&doc).is_ok());
}
#[test]
fn size_hint_pre_allocates_to_match() {
let doc = b"<r><a>hi</a></r>".to_vec();
let r = XmlByteStreamReader::with_size_hint(
Cursor::new(doc.clone()),
Some(doc.len()),
1024 * 1024,
).unwrap();
assert!(r.buf.capacity() < 1024 * 1024 / 4,
"expected capacity ~{} bytes, got {}", doc.len(), r.buf.capacity());
assert!(r.validate().is_ok());
}
#[test]
fn empty_input_errors_cleanly() {
let result = validate(b"");
assert!(result.is_err());
}
#[test]
fn handles_text_split_across_refills() {
let text = "x".repeat(1000);
let doc = format!("<r>{text}</r>");
assert!(validate_with_buffer(doc.as_bytes(), 2048).is_ok());
}
#[test]
fn many_small_events_with_small_buffer() {
let mut doc = String::from("<r>");
for _ in 0..1000 {
doc.push_str("<x/>");
}
doc.push_str("</r>");
let r = validate_with_buffer(doc.as_bytes(), 512);
if let Err(e) = &r {
panic!("expected ok; got error: {} (line={:?}, col={:?})",
e.message, e.line, e.column);
}
}
#[test]
fn next_event_extracts_data_with_small_buffer() {
const N: usize = 500;
let mut doc = String::from("<root>");
for i in 0..N {
doc.push_str(&format!("<item>value{i}</item>"));
}
doc.push_str("</root>");
let mut r = XmlByteStreamReader::new(Cursor::new(doc.into_bytes()), 512).unwrap();
let mut values = Vec::new();
loop {
let eof = {
match r.next_event().unwrap() {
BytesEvent::Eof => true,
BytesEvent::Text(t) => {
let s = std::str::from_utf8(t.as_bytes()).unwrap();
if !s.trim().is_empty() {
values.push(s.to_string());
}
false
}
_ => false,
}
};
if eof { break; }
}
assert_eq!(values.len(), N, "should recover every item's text");
assert_eq!(values[0], "value0");
assert_eq!(values[N - 1], format!("value{}", N - 1));
}
#[test]
fn next_event_keeps_memory_bounded() {
let buffer_size = 1024;
let cap = buffer_size * BUF_CAPACITY_MULTIPLE;
let mut doc = String::from("<root>");
for i in 0..5000 {
doc.push_str(&format!("<item id=\"{i}\">value{i}</item>"));
}
doc.push_str("</root>");
assert!(doc.len() > cap * 10, "doc must dwarf the buffer for the test to mean anything");
let mut r = XmlByteStreamReader::new(Cursor::new(doc.into_bytes()), buffer_size).unwrap();
loop {
let eof = matches!(r.next_event().unwrap(), BytesEvent::Eof);
assert!(r.buf.len() <= cap,
"buffer grew to {} bytes, exceeding cap {cap}", r.buf.len());
if eof { break; }
}
}
#[test]
fn next_event_surfaces_wellformedness_errors() {
let mut r = XmlByteStreamReader::new(Cursor::new(b"<a></b>".to_vec()), 1024).unwrap();
let mut got_err = false;
for _ in 0..10 {
match r.next_event() {
Ok(BytesEvent::Eof) => break,
Ok(_) => {}
Err(_) => { got_err = true; break; }
}
}
assert!(got_err, "mismatched end tag should surface as Err from next_event");
}
}