ifc_lite_landxml/
stream.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2 * License, v. 2.0. If a copy of the MPL was not distributed with this
3 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
4
5mod decoder;
6mod derived;
7mod event;
8mod fragments;
9mod lifecycle;
10pub(crate) mod metadata;
11mod output;
12mod token;
13pub use lifecycle::LandXmlMetadataStreamAssembler;
14
15use crate::{
16    parser::Parser, preflight::max_markup_bytes, xml::error, LandXmlDiagnosticCode as Code,
17    LandXmlError, LandXmlLimits,
18};
19use decoder::Decoder;
20pub use event::{
21    LandXmlMetadataRecord, LandXmlMetadataRecordFragment, LandXmlMetadataStreamEnd,
22    LandXmlMetadataStreamEvent, LandXmlMetadataStreamHeader, LandXmlStreamEvent,
23    LandXmlStreamHeader, LandXmlStreamMetadata, LandXmlStreamSummary, LandXmlSurfaceComponent,
24    LandXmlSurfaceFragment,
25};
26use quick_xml::{events::Event, Reader};
27use std::{collections::VecDeque, io::BufReader};
28use token::{TokenFeed, TokenKind};
29
30pub const MAX_LANDXML_STREAM_INPUT_CHUNK_BYTES: usize = 1024 * 1024;
31pub const MAX_LANDXML_STREAM_DRAIN_BYTES: usize = 1024 * 1024;
32/// The host must return transport credit by draining before more XML is read.
33pub const MAX_LANDXML_STREAM_QUEUED_BYTES: usize = 512 * 1024;
34pub const MAX_LANDXML_STREAM_QUEUED_EVENTS: usize = 4;
35/// Maximum serialized event retained while reserving queue headroom.
36pub const MAX_LANDXML_STREAM_EVENT_BYTES: usize = 192 * 1024;
37/// A single semantic metadata value may legally exceed one transport event,
38/// but no serializer/reassembler state may exceed the host's one-credit cap.
39pub const MAX_LANDXML_STREAM_METADATA_RECORD_BYTES: usize = MAX_LANDXML_STREAM_QUEUED_BYTES;
40
41struct QueuedEvent {
42    event: LandXmlStreamEvent,
43    serialized_bytes: usize,
44}
45
46pub struct LandXmlTinStreamSession {
47    parser: Option<Parser<'static>>,
48    plan: Option<crate::plan::parser::Parser<'static>>,
49    alignment: Option<crate::alignment::parser::Parser<'static>>,
50    pipe: Option<crate::pipe_parser::PipeParser<'static>>,
51    reader: Reader<BufReader<TokenFeed>>,
52    feed: TokenFeed,
53    decoder: Decoder,
54    token: Vec<u8>,
55    kind: TokenKind,
56    max_text_bytes: usize,
57    max_markup_bytes: usize,
58    queue: VecDeque<QueuedEvent>,
59    queued_bytes: usize,
60    pending_surface: Option<fragments::SurfaceCursor>,
61    metadata_cursor: Option<metadata::MetadataCursor>,
62    pending_input: VecDeque<u8>,
63    header_emitted: bool,
64    surfaces_drained: usize,
65    renderable_surfaces: usize,
66    preserved_surfaces: usize,
67    closed: bool,
68}
69
70impl LandXmlTinStreamSession {
71    pub fn new(limits: LandXmlLimits) -> Result<Self, LandXmlError> {
72        // #5175: validate a caller-supplied assumed-unit override up front,
73        // the same as the non-streaming entry point, so a bad token refuses
74        // at session construction rather than partway through a stream.
75        let assumed_units = crate::semantics::resolve_assumed_units(&limits)?;
76        let feed = TokenFeed::new();
77        let mut reader = Reader::from_reader(BufReader::new(feed.clone()));
78        reader.config_mut().trim_text(false);
79        Ok(Self {
80            parser: Some(Parser::new(&limits, None, assumed_units)),
81            plan: Some(crate::plan::parser::Parser::new_stream(
82                crate::LandXmlPlanLimits {
83                    xml: limits.clone(),
84                    ..Default::default()
85                },
86            )),
87            alignment: Some(crate::alignment::parser::Parser::new_stream(
88                crate::alignment::LandXmlAlignmentLimits {
89                    xml: limits.clone(),
90                    ..Default::default()
91                },
92            )),
93            pipe: Some(crate::pipe_parser::PipeParser::new_stream(limits.clone())),
94            reader,
95            feed,
96            decoder: Decoder::new(&limits)?,
97            token: Vec::new(),
98            kind: TokenKind::Text,
99            max_text_bytes: limits.max_text_bytes,
100            max_markup_bytes: max_markup_bytes(&limits)?,
101            queue: VecDeque::new(),
102            queued_bytes: 0,
103            pending_surface: None,
104            metadata_cursor: None,
105            pending_input: VecDeque::new(),
106            header_emitted: false,
107            surfaces_drained: 0,
108            renderable_surfaces: 0,
109            preserved_surfaces: 0,
110            closed: false,
111        })
112    }
113
114    pub fn advance(&mut self, chunk: &[u8]) -> Result<(), LandXmlError> {
115        if self.closed {
116            return Err(error(
117                Code::InvalidSemantic,
118                "LandXML stream session is closed",
119            ));
120        }
121        if chunk.len() > MAX_LANDXML_STREAM_INPUT_CHUNK_BYTES {
122            return Err(error(Code::InputTooLarge, "input chunk exceeds byte limit"));
123        }
124        if self.output_pending() {
125            return Err(error(
126                Code::LimitExceeded,
127                "stream output must be drained before advancing input",
128            ));
129        }
130        self.pending_input.extend(self.decoder.push(chunk)?);
131        self.consume_pending_input()?;
132        Ok(())
133    }
134
135    pub fn drain(&mut self, max_bytes: usize) -> Result<Vec<LandXmlStreamEvent>, LandXmlError> {
136        if max_bytes == 0 || max_bytes > MAX_LANDXML_STREAM_DRAIN_BYTES {
137            return Err(error(Code::LimitExceeded, "invalid stream drain budget"));
138        }
139        let mut bytes = 0usize;
140        let mut events = Vec::new();
141        while let Some(queued) = self.queue.front() {
142            let size = queued.serialized_bytes;
143            if size > max_bytes {
144                return Err(error(
145                    Code::LimitExceeded,
146                    "stream event exceeds drain budget",
147                ));
148            }
149            if !events.is_empty() && bytes + size > max_bytes {
150                break;
151            }
152            bytes += size;
153            let queued = self.queue.pop_front().expect("front checked");
154            self.queued_bytes -= queued.serialized_bytes;
155            events.push(queued.event);
156        }
157        self.resume_after_drain()?;
158        Ok(events)
159    }
160
161    /// Finalize parsing and begin credited metadata delivery.
162    ///
163    /// Call [`Self::drain`] until [`Self::output_pending`] is false. Metadata
164    /// is emitted as bounded header, record, and end events without cloning
165    /// the finalized family documents.
166    pub fn finish_cursor(&mut self) -> Result<(), LandXmlError> {
167        if self.closed {
168            return Err(error(
169                Code::InvalidSemantic,
170                "LandXML stream session is closed",
171            ));
172        }
173        self.decoder.finish()?;
174        if self.output_pending() || !self.pending_input.is_empty() {
175            return Err(error(
176                Code::LimitExceeded,
177                "stream output must be drained before finalization",
178            ));
179        }
180        if !self.token.is_empty() {
181            self.emit_token()?;
182        }
183        if self.parser().has_open_frames() {
184            return Err(error(Code::InvalidXml, "unclosed XML element"));
185        }
186        let header = self
187            .header()
188            .ok_or_else(|| error(Code::InvalidXml, "LandXML document is empty"))?;
189        if !self.header_emitted {
190            self.push_event(LandXmlStreamEvent::Header(header.clone()))?;
191            self.header_emitted = true;
192        }
193        self.closed = true;
194        let terrain = self.parser.take().expect("open parser").finish()?;
195        let plan = self.plan.take().expect("open plan parser");
196        if plan.has_open_frames() {
197            return Err(error(Code::InvalidXml, "unclosed plan XML element"));
198        }
199        let plan = plan.document();
200        let alignment = self.alignment.take().expect("open alignment parser");
201        if alignment.has_open_frames() {
202            return Err(error(Code::InvalidXml, "unclosed alignment XML element"));
203        }
204        let alignment = alignment.finish()?;
205        let pipe = self.pipe.take().expect("open pipe parser");
206        if pipe.has_open_frames() {
207            return Err(error(Code::InvalidXml, "unclosed pipe XML element"));
208        }
209        let (pipe, preflight_refusal_batches, has_pipe_networks) =
210            pipe.finish_stream_with_preflight()?;
211        let pipe_networks = pipe.networks.len();
212        let pipe_structures = pipe
213            .networks
214            .iter()
215            .map(|network| network.structures.len())
216            .sum();
217        let pipes = pipe
218            .networks
219            .iter()
220            .map(|network| network.pipes.len())
221            .sum();
222        let pipe_refusals = pipe.refusals.len();
223        let end = LandXmlMetadataStreamEnd {
224            surfaces_drained: self.surfaces_drained,
225            renderable_surfaces: self.renderable_surfaces,
226            preserved_surfaces: self.preserved_surfaces,
227            plan_cogo_points: plan.cogo_points().len(),
228            plan_parcels: plan.parcels.len(),
229            horizontal_alignments: alignment.alignments.len(),
230            has_pipe_networks,
231            pipe_networks,
232            pipe_structures,
233            pipes,
234            pipe_refusals,
235        };
236        self.metadata_cursor = Some(metadata::MetadataCursor::new(
237            header,
238            terrain.into_stream_parts(),
239            plan,
240            alignment,
241            pipe.into_stream_parts(preflight_refusal_batches),
242            end,
243        ));
244        self.flush_pending_metadata()
245    }
246
247    fn parser(&mut self) -> &mut Parser<'static> {
248        self.parser.as_mut().expect("open stream parser")
249    }
250
251    fn consume_pending_input(&mut self) -> Result<(), LandXmlError> {
252        while !self.output_pending() {
253            let Some(byte) = self.pending_input.pop_front() else {
254                return Ok(());
255            };
256            self.push_normalized(byte)?;
257        }
258        Ok(())
259    }
260
261    fn resume_after_drain(&mut self) -> Result<(), LandXmlError> {
262        self.flush_pending_surface()?;
263        self.flush_pending_metadata()?;
264        self.consume_pending_input()
265    }
266
267    fn push_normalized(&mut self, byte: u8) -> Result<(), LandXmlError> {
268        match &mut self.kind {
269            TokenKind::Text if byte == b'<' => {
270                self.emit_token()?;
271                self.token.push(byte);
272                self.kind = TokenKind::Markup { quote: None };
273            }
274            TokenKind::Text => self.token.push(byte),
275            TokenKind::Markup { quote } => {
276                self.token.push(byte);
277                if self.token.starts_with(b"<!DOCTYPE") {
278                    return Err(error(Code::DtdForbidden, "DOCTYPE is not allowed"));
279                }
280                let complete = if self.token.starts_with(b"<!--") {
281                    self.token.ends_with(b"-->")
282                } else if self.token.starts_with(b"<![CDATA[") {
283                    self.token.ends_with(b"]]>")
284                } else if self.token.starts_with(b"<?") {
285                    self.token.ends_with(b"?>")
286                } else {
287                    match *quote {
288                        Some(value) if byte == value => {
289                            *quote = None;
290                            false
291                        }
292                        Some(_) => false,
293                        None if matches!(byte, b'\'' | b'\"') => {
294                            *quote = Some(byte);
295                            false
296                        }
297                        None => byte == b'>',
298                    }
299                };
300                if complete {
301                    self.emit_token()?;
302                    self.kind = TokenKind::Text;
303                }
304            }
305        }
306        let (limit, message) = self.token_limit();
307        if self.token.len() > limit {
308            return Err(error(Code::LimitExceeded, message));
309        }
310        Ok(())
311    }
312
313    fn token_limit(&self) -> (usize, &'static str) {
314        match &self.kind {
315            TokenKind::Text => (self.max_text_bytes, "text limit exceeded"),
316            TokenKind::Markup { .. }
317                if self.token.starts_with(b"<!--") || self.token.starts_with(b"<![CDATA[") =>
318            {
319                (self.max_text_bytes, "XML token limit exceeded")
320            }
321            TokenKind::Markup { .. } => (self.max_markup_bytes, "markup limit exceeded"),
322        }
323    }
324
325    fn emit_token(&mut self) -> Result<(), LandXmlError> {
326        if self.token.is_empty() {
327            return Ok(());
328        }
329        let token = std::mem::take(&mut self.token);
330        let token_debug = String::from_utf8_lossy(&token).into_owned();
331        let text_token = !token.starts_with(b"<");
332        let comment_token = token.starts_with(b"<!--");
333        self.feed.push(token);
334        if text_token {
335            self.feed.push(b"<!--ifc-lite-stream-pad-->".to_vec());
336        } else if comment_token {
337            // Keep quick-xml from observing EOF for a standalone comment.
338            // The processing instruction is deliberately not exposed to the
339            // semantic parsers below.
340            self.feed.push(b"<?ifc-lite-stream-pad?>".to_vec());
341        }
342        let mut buffer = Vec::new();
343        let mut consumed_comment = false;
344        loop {
345            let event = self.reader.read_event_into(&mut buffer).map_err(|value| {
346                error(Code::InvalidXml, format!("malformed XML token: {value}"))
347            })?;
348            if matches!(event, Event::Comment(_)) {
349                // Text tokens leave a padding comment in the reader to stop
350                // quick-xml waiting for the next chunk. A real XML comment
351                // is also a complete legal token. Keep consuming in case an
352                // earlier padding comment precedes this token; EOF after only
353                // comments is the legitimate standalone-comment case.
354                consumed_comment = true;
355                buffer.clear();
356                continue;
357            }
358            if comment_token && consumed_comment && matches!(event, Event::PI(_)) {
359                return self.emit_header_and_surfaces();
360            }
361            if matches!(event, Event::Eof) {
362                if consumed_comment {
363                    return self.emit_header_and_surfaces();
364                }
365                return Err(error(
366                    Code::InvalidXml,
367                    format!("XML reader ended while consuming token {token_debug:?}"),
368                ));
369            }
370            self.parser().consume_event(event.clone())?;
371            self.plan
372                .as_mut()
373                .expect("open plan parser")
374                .consume_event(event.clone())?;
375            self.alignment
376                .as_mut()
377                .expect("open alignment parser")
378                .consume_event(event.clone())?;
379            self.pipe
380                .as_mut()
381                .expect("open pipe parser")
382                .consume_event(event)?;
383            break;
384        }
385        self.emit_header_and_surfaces()
386    }
387}
388
389impl Drop for LandXmlTinStreamSession {
390    fn drop(&mut self) {
391        self.abort();
392    }
393}
ifc_lite_landxml/stream.rs

ifc_lite_landxml/
stream.rs
