bgpkit_parser/parser/iters/mod.rs
1/*!
2Iterator implementations for bgpkit-parser.
3
4This module contains different iterator implementations for parsing BGP data:
5- `default`: Standard iterators that skip errors (RecordIterator, ElemIterator)
6- `fallible`: Fallible iterators that return Results (FallibleRecordIterator, FallibleElemIterator)
7- `update`: Iterators for BGP UPDATE messages (UpdateIterator, FallibleUpdateIterator)
8
9It also contains the trait implementations that enable BgpkitParser to be used with
10Rust's iterator syntax.
11*/
12
13pub mod default;
14mod diagnostic;
15pub mod fallible;
16mod raw;
17mod recovery;
18mod route;
19mod update;
20
21// Re-export all iterator types for convenience
22pub use default::{ElemIterator, RecordIterator};
23pub use diagnostic::{
24 record_validation_warnings, span_record_warnings, DiagnosticEvent, DiagnosticIterator,
25 DissectedDiagnosticEvent, DissectingDiagnosticIterator,
26};
27pub use fallible::{FallibleElemIterator, FallibleRecordIterator};
28pub use raw::{FilteredRawRecordIterator, RawRecordIterator};
29pub use recovery::{
30 RecoveringElemIterator, RecoveringRecordIterator, RecoveryConfig, RecoveryError, RecoveryEvent,
31 RecoveryEvidence, RecoveryGap,
32};
33pub use route::{FallibleRouteIterator, RouteIterator};
34pub use update::{
35 Bgp4MpUpdate, FallibleUpdateIterator, LegacyBgpUpdate, MrtUpdate, TableDumpV2Entry,
36 UpdateIterator,
37};
38
39use crate::error::ParserError;
40use crate::models::BgpElem;
41use crate::models::{MrtMessage, MrtRecord, TableDumpV2Message};
42use crate::parser::BgpkitParser;
43use crate::RawMrtRecord;
44use crate::{Elementor, Filter, Filterable};
45use log::{debug, error, warn};
46use std::io::Read;
47use std::path::Path;
48
49#[inline]
50pub(crate) fn record_matches_filters(
51 record: &MrtRecord,
52 filters: &[Filter],
53 elementor: &mut Elementor,
54) -> bool {
55 if filters.is_empty() {
56 return true;
57 }
58 if matches!(
59 &record.message,
60 MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(_))
61 ) {
62 let _ = elementor.record_to_elems(record.clone());
63 return true;
64 }
65 // Filters match on the elem projection. Records that produce no elems
66 // (KEEPALIVE, OPEN, NOTIFICATION, state changes) can therefore never
67 // match and are dropped from record iteration while filters are active.
68 let elems = elementor.record_to_elems(record.clone());
69 if elems.is_empty() {
70 debug!(
71 "filters active: record of type {:?} yields no elems and is dropped",
72 record.common_header.entry_type
73 );
74 return false;
75 }
76 elems.iter().any(|element| element.match_filters(filters))
77}
78
79/// Shared body-parse error policy for record-producing iterators.
80///
81/// Mirrors the historical `RecordIterator` behavior: warnings honor
82/// `disable_warnings()`, core dumps are written for recoverable classes,
83/// and a fatal `ParseError` with core dumps enabled stops the iterator so
84/// a later failure cannot overwrite the dump. Returns `true` to continue
85/// iterating, `false` to stop.
86pub(crate) fn handle_record_parse_error<R>(
87 parser: &mut crate::parser::BgpkitParser<R>,
88 error: ParserError,
89 bytes: Option<Vec<u8>>,
90) -> bool {
91 match error {
92 ParserError::TruncatedMsg(err_str) | ParserError::Unsupported(err_str) => {
93 if parser.options.show_warnings {
94 warn!("parser warn: {}", err_str);
95 }
96 write_mrt_core_dump(parser.core_dump, bytes);
97 true
98 }
99 ParserError::ParseError(err_str) => {
100 error!("parser error: {}", err_str);
101 write_mrt_core_dump(parser.core_dump, bytes);
102 // stop after writing the dump so later failures don't overwrite it
103 !parser.core_dump
104 }
105 ParserError::EofExpected => {
106 // normal end of file
107 false
108 }
109 ParserError::IoError(err) | ParserError::EofError(err) => {
110 // when reaching IO error, stop iterating
111 error!("{:?}", err);
112 write_mrt_core_dump(parser.core_dump, bytes);
113 false
114 }
115 #[cfg(feature = "oneio")]
116 ParserError::OneIoError(_) => false,
117 ParserError::FilterError(_) => {
118 // this should not happen at this stage
119 false
120 }
121 // Labeled NLRI parsing errors - treat as malformed and skip
122 ParserError::InvalidLabeledNlriLength
123 | ParserError::TruncatedLabeledNlri
124 | ParserError::TruncatedPrefix
125 | ParserError::MaxLabelStackDepthExceeded
126 | ParserError::PeerMaxLabelsExceeded
127 | ParserError::InvalidPrefix => {
128 if parser.options.show_warnings {
129 warn!("parser warn: labeled NLRI parsing error: {:?}", error);
130 }
131 true
132 }
133 }
134}
135
136pub(crate) fn write_mrt_core_dump(enabled: bool, bytes: Option<Vec<u8>>) {
137 write_mrt_core_dump_to_path(enabled, bytes, "mrt_core_dump");
138}
139
140pub(crate) fn write_mrt_core_dump_to_path<P: AsRef<Path>>(
141 enabled: bool,
142 bytes: Option<Vec<u8>>,
143 path: P,
144) {
145 if enabled {
146 if let Some(bytes) = bytes {
147 std::fs::write(path, bytes).expect("Unable to write to mrt_core_dump");
148 }
149 }
150}
151
152/// Use [ElemIterator] as the default iterator to return [BgpElem]s instead of [MrtRecord]s.
153impl<R: Read> IntoIterator for BgpkitParser<R> {
154 type Item = BgpElem;
155 type IntoIter = ElemIterator<R>;
156
157 fn into_iter(self) -> Self::IntoIter {
158 ElemIterator::new(self)
159 }
160}
161
162impl<R> BgpkitParser<R> {
163 pub fn into_record_iter(self) -> RecordIterator<R> {
164 RecordIterator::new(self)
165 }
166
167 pub fn into_elem_iter(self) -> ElemIterator<R> {
168 ElemIterator::new(self)
169 }
170
171 pub fn into_raw_record_iter(self) -> RawRecordIterator<R> {
172 RawRecordIterator::new(self)
173 }
174
175 /// Creates an iterator over raw MRT records with record-level filter
176 /// semantics applied.
177 ///
178 /// Like [`into_raw_record_iter`](Self::into_raw_record_iter), but only
179 /// records passing the parser's filters are yielded (same semantics as
180 /// [`into_record_iter`](Self::into_record_iter): filters match on the
181 /// elem projection, so no-elem records such as KEEPALIVEs are dropped
182 /// while filters are active, and the `PeerIndexTable` always passes).
183 /// The yielded records carry their original wire bytes — no
184 /// re-encoding — which is what byte-exact consumers (hex output,
185 /// re-dissection) need. Every record body is parsed once inside the
186 /// iterator and the parsed record is yielded alongside, so consumers
187 /// do not parse the bytes twice; parse failures follow the same
188 /// variant-aware diagnostics as the record iterator.
189 ///
190 /// # Example
191 /// ```no_run
192 /// use bgpkit_parser::BgpkitParser;
193 ///
194 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
195 /// for (raw, _) in parser.into_filtered_raw_record_iter() {
196 /// println!("{}", raw.raw_bytes().len());
197 /// }
198 /// ```
199 pub fn into_filtered_raw_record_iter(self) -> FilteredRawRecordIterator<R> {
200 FilteredRawRecordIterator::new(self)
201 }
202
203 /// Creates an opt-in iterator that reports skipped byte ranges while recovering MRT framing.
204 ///
205 /// Recovery never reconstructs a damaged record. It scans for a structurally valid boundary,
206 /// confirms a chain of records, emits [`RecoveryEvent::Gap`], and then resumes normal parsing.
207 /// Damage extending to the end of the stream is reported as a terminal gap. Offsets in
208 /// recovery events refer to the decompressed MRT byte stream.
209 pub fn into_recovering_record_iter(
210 self,
211 config: RecoveryConfig,
212 ) -> RecoveringRecordIterator<R> {
213 RecoveringRecordIterator::new(self, config)
214 }
215
216 /// Creates an opt-in iterator over BGP elements that reports skipped byte ranges while
217 /// recovering MRT framing.
218 ///
219 /// Behaves like [`into_recovering_record_iter`](Self::into_recovering_record_iter) but
220 /// converts each recovered record to [`BgpElem`]s, applying the parser's filters per
221 /// element.
222 pub fn into_recovering_elem_iter(self, config: RecoveryConfig) -> RecoveringElemIterator<R> {
223 RecoveringElemIterator::new(self, config)
224 }
225
226 /// Creates an iterator over BGP announcements from MRT data.
227 ///
228 /// This iterator yields `MrtUpdate` items from both UPDATES files (BGP4MP messages)
229 /// and RIB dump files (TableDump/TableDumpV2 messages). It's a middle ground
230 /// between `into_record_iter()` and `into_elem_iter()`:
231 ///
232 /// - More focused than `into_record_iter()` as it only returns BGP announcements
233 /// - More efficient than `into_elem_iter()` as it doesn't duplicate attributes per prefix
234 ///
235 /// The iterator returns an `MrtUpdate` enum with variants:
236 /// - `Bgp4MpUpdate`: BGP UPDATE messages from UPDATES files
237 /// - `LegacyBgpUpdate`: Deprecated MRT Type 5 BGP UPDATE messages
238 /// - `TableDumpV2Entry`: RIB entries from TableDumpV2 RIB dumps
239 /// - `TableDumpMessage`: Legacy TableDump v1 messages
240 ///
241 /// # Example
242 /// ```no_run
243 /// use bgpkit_parser::{BgpkitParser, MrtUpdate};
244 ///
245 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
246 /// for update in parser.into_update_iter() {
247 /// match update {
248 /// MrtUpdate::Bgp4MpUpdate(u) => {
249 /// println!("Peer {} announced {} prefixes",
250 /// u.peer_ip,
251 /// u.message.announced_prefixes.len()
252 /// );
253 /// }
254 /// MrtUpdate::LegacyBgpUpdate(u) => {
255 /// println!("Legacy UPDATE from peer {}", u.peer_ip);
256 /// }
257 /// MrtUpdate::TableDumpV2Entry(e) => {
258 /// println!("RIB entry for {} with {} peers",
259 /// e.prefix,
260 /// e.rib_entries.len()
261 /// );
262 /// }
263 /// MrtUpdate::TableDumpMessage(m) => {
264 /// println!("Legacy table dump for {}", m.prefix);
265 /// }
266 /// }
267 /// }
268 /// ```
269 pub fn into_update_iter(self) -> UpdateIterator<R> {
270 UpdateIterator::new(self)
271 }
272
273 /// Creates an iterator over lightweight route elements from MRT data.
274 ///
275 /// This iterator yields [`BgpRouteElem`](crate::models::BgpRouteElem)
276 /// values and only parses route identity, peer metadata, timestamp, and
277 /// AS path. Use [`into_elem_iter`](Self::into_elem_iter) when you need
278 /// the full [`BgpElem`] attribute set. Filters that only depend on route
279 /// fields are supported; `community` filters do not match route elements.
280 pub fn into_route_iter(self) -> RouteIterator<R> {
281 RouteIterator::new(self)
282 }
283
284 /// Creates a fallible iterator over MRT records that returns parsing errors.
285 ///
286 /// # Example
287 /// ```no_run
288 /// use bgpkit_parser::BgpkitParser;
289 ///
290 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
291 /// for result in parser.into_fallible_record_iter() {
292 /// match result {
293 /// Ok(record) => {
294 /// // Process the record
295 /// }
296 /// Err(e) => {
297 /// // Handle the error
298 /// eprintln!("Error parsing record: {}", e);
299 /// }
300 /// }
301 /// }
302 /// ```
303 pub fn into_fallible_record_iter(self) -> FallibleRecordIterator<R> {
304 FallibleRecordIterator::new(self)
305 }
306
307 /// Creates a fallible iterator over BGP elements that returns parsing errors.
308 ///
309 /// # Example
310 /// ```no_run
311 /// use bgpkit_parser::BgpkitParser;
312 ///
313 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
314 /// for result in parser.into_fallible_elem_iter() {
315 /// match result {
316 /// Ok(elem) => {
317 /// // Process the element
318 /// }
319 /// Err(e) => {
320 /// // Handle the error
321 /// eprintln!("Error parsing element: {}", e);
322 /// }
323 /// }
324 /// }
325 /// ```
326 pub fn into_fallible_elem_iter(self) -> FallibleElemIterator<R> {
327 FallibleElemIterator::new(self)
328 }
329
330 /// Creates a fallible iterator over BGP announcements that returns parsing errors.
331 ///
332 /// Unlike the default `into_update_iter()`, this iterator returns
333 /// `Result<MrtUpdate, ParserErrorWithBytes>` allowing users to handle parsing
334 /// errors explicitly instead of having them logged and skipped.
335 ///
336 /// # Example
337 /// ```no_run
338 /// use bgpkit_parser::{BgpkitParser, MrtUpdate};
339 ///
340 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
341 /// for result in parser.into_fallible_update_iter() {
342 /// match result {
343 /// Ok(MrtUpdate::Bgp4MpUpdate(update)) => {
344 /// println!("Peer {} announced {} prefixes",
345 /// update.peer_ip,
346 /// update.message.announced_prefixes.len()
347 /// );
348 /// }
349 /// Ok(_) => { /* handle other variants */ }
350 /// Err(e) => {
351 /// eprintln!("Error parsing: {}", e);
352 /// }
353 /// }
354 /// }
355 /// ```
356 pub fn into_fallible_update_iter(self) -> FallibleUpdateIterator<R> {
357 FallibleUpdateIterator::new(self)
358 }
359
360 /// Creates a fallible iterator over lightweight route elements.
361 pub fn into_fallible_route_iter(self) -> FallibleRouteIterator<R> {
362 FallibleRouteIterator::new(self)
363 }
364
365 /// Creates an iterator that classifies each MRT record for malformed-data investigation.
366 ///
367 /// This iterator emits every record with its raw bytes attached: clean
368 /// records have empty warning lists, records with recoverable RFC 7606
369 /// validation findings carry them in `warnings`, and fatal parse errors
370 /// retain the consumed bytes plus a best-effort partial dissection tree
371 /// showing where parsing stopped. It ignores parser filters so that
372 /// malformed records cannot be hidden by element-oriented matching. Text-dump parsers yield
373 /// no diagnostic events because they have no MRT record representation.
374 ///
375 /// # Example
376 /// ```no_run
377 /// use bgpkit_parser::{BgpkitParser, DiagnosticEvent};
378 ///
379 /// for event in BgpkitParser::new("updates.mrt")?.into_diagnostic_iter() {
380 /// match event {
381 /// DiagnosticEvent::Record { record, raw, warnings } => {
382 /// if warnings.is_empty() {
383 /// println!("{record}");
384 /// } else {
385 /// eprintln!("validation findings: {warnings:?}");
386 /// raw.write_raw_bytes("malformed-record.mrt")?;
387 /// }
388 /// }
389 /// DiagnosticEvent::ParseError { error, raw_bytes, .. } => {
390 /// eprintln!("parse error: {error}");
391 /// if let Some(raw_bytes) = raw_bytes {
392 /// std::fs::write("malformed-record.mrt", raw_bytes)?;
393 /// }
394 /// }
395 /// _ => {}
396 /// }
397 /// }
398 /// # Ok::<(), Box<dyn std::error::Error>>(())
399 /// ```
400 pub fn into_diagnostic_iter(self) -> DiagnosticIterator<R> {
401 DiagnosticIterator::new(self)
402 }
403
404 /// Creates an Elementor pre-initialized with PeerIndexTable and an iterator over raw records.
405 ///
406 /// This is useful for parallel processing where the Elementor needs to be shared across threads.
407 /// The Elementor is created with the PeerIndexTable from the first record if present,
408 /// otherwise a new Elementor is created.
409 ///
410 /// # Example
411 /// See the `parallel_records_to_elem` example for full usage.
412 /// ```ignore
413 /// use bgpkit_parser::BgpkitParser;
414 ///
415 /// let parser = BgpkitParser::new_cached(url, "/tmp")?;
416 /// let (elementor, records) = parser.into_elementor_and_raw_record_iter();
417 /// ```
418 ///
419 pub fn into_elementor_and_raw_record_iter(
420 self,
421 ) -> (Elementor, impl Iterator<Item = RawMrtRecord>)
422 where
423 R: Read,
424 {
425 let mut raw_iter = RawRecordIterator::new(self).peekable();
426 let elementor = match raw_iter.peek().cloned().and_then(|r| r.parse().ok()) {
427 Some(MrtRecord {
428 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(pit)),
429 ..
430 }) => {
431 raw_iter.next();
432 Elementor::with_peer_table(pit)
433 }
434 _ => Elementor::new(),
435 };
436 (elementor, raw_iter)
437 }
438
439 /// Creates an Elementor pre-initialized with PeerIndexTable and an iterator over parsed records.
440 ///
441 /// This is useful for parallel processing where the Elementor needs to be shared across threads.
442 /// The Elementor is created with the PeerIndexTable from the first record if present,
443 /// otherwise a new Elementor is created.
444 ///
445 /// # Example
446 /// See the `parallel_records_to_elem` example for full usage.
447 pub fn into_elementor_and_record_iter(self) -> (Elementor, impl Iterator<Item = MrtRecord>)
448 where
449 R: Read,
450 {
451 let mut record_iter = RecordIterator::new(self).peekable();
452 let elementor = match record_iter.peek().cloned() {
453 Some(MrtRecord {
454 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(pit)),
455 ..
456 }) => {
457 record_iter.next();
458 Elementor::with_peer_table(pit)
459 }
460 _ => Elementor::new(),
461 };
462 (elementor, record_iter)
463 }
464}