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::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::models::BgpElem;
40use crate::models::{MrtMessage, MrtRecord, TableDumpV2Message};
41use crate::parser::BgpkitParser;
42use crate::RawMrtRecord;
43use crate::{Elementor, Filter, Filterable};
44use std::io::Read;
45use std::path::Path;
46
47#[inline]
48pub(crate) fn record_matches_filters(
49 record: &MrtRecord,
50 filters: &[Filter],
51 elementor: &mut Elementor,
52) -> bool {
53 if filters.is_empty() {
54 return true;
55 }
56 if matches!(
57 &record.message,
58 MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(_))
59 ) {
60 let _ = elementor.record_to_elems(record.clone());
61 return true;
62 }
63 elementor
64 .record_to_elems(record.clone())
65 .iter()
66 .any(|element| element.match_filters(filters))
67}
68
69pub(crate) fn write_mrt_core_dump(enabled: bool, bytes: Option<Vec<u8>>) {
70 write_mrt_core_dump_to_path(enabled, bytes, "mrt_core_dump");
71}
72
73pub(crate) fn write_mrt_core_dump_to_path<P: AsRef<Path>>(
74 enabled: bool,
75 bytes: Option<Vec<u8>>,
76 path: P,
77) {
78 if enabled {
79 if let Some(bytes) = bytes {
80 std::fs::write(path, bytes).expect("Unable to write to mrt_core_dump");
81 }
82 }
83}
84
85/// Use [ElemIterator] as the default iterator to return [BgpElem]s instead of [MrtRecord]s.
86impl<R: Read> IntoIterator for BgpkitParser<R> {
87 type Item = BgpElem;
88 type IntoIter = ElemIterator<R>;
89
90 fn into_iter(self) -> Self::IntoIter {
91 ElemIterator::new(self)
92 }
93}
94
95impl<R> BgpkitParser<R> {
96 pub fn into_record_iter(self) -> RecordIterator<R> {
97 RecordIterator::new(self)
98 }
99
100 pub fn into_elem_iter(self) -> ElemIterator<R> {
101 ElemIterator::new(self)
102 }
103
104 pub fn into_raw_record_iter(self) -> RawRecordIterator<R> {
105 RawRecordIterator::new(self)
106 }
107
108 /// Creates an opt-in iterator that reports skipped byte ranges while recovering MRT framing.
109 ///
110 /// Recovery never reconstructs a damaged record. It scans for a structurally valid boundary,
111 /// confirms a chain of records, emits [`RecoveryEvent::Gap`], and then resumes normal parsing.
112 /// Damage extending to the end of the stream is reported as a terminal gap. Offsets in
113 /// recovery events refer to the decompressed MRT byte stream.
114 pub fn into_recovering_record_iter(
115 self,
116 config: RecoveryConfig,
117 ) -> RecoveringRecordIterator<R> {
118 RecoveringRecordIterator::new(self, config)
119 }
120
121 /// Creates an opt-in iterator over BGP elements that reports skipped byte ranges while
122 /// recovering MRT framing.
123 ///
124 /// Behaves like [`into_recovering_record_iter`](Self::into_recovering_record_iter) but
125 /// converts each recovered record to [`BgpElem`]s, applying the parser's filters per
126 /// element.
127 pub fn into_recovering_elem_iter(self, config: RecoveryConfig) -> RecoveringElemIterator<R> {
128 RecoveringElemIterator::new(self, config)
129 }
130
131 /// Creates an iterator over BGP announcements from MRT data.
132 ///
133 /// This iterator yields `MrtUpdate` items from both UPDATES files (BGP4MP messages)
134 /// and RIB dump files (TableDump/TableDumpV2 messages). It's a middle ground
135 /// between `into_record_iter()` and `into_elem_iter()`:
136 ///
137 /// - More focused than `into_record_iter()` as it only returns BGP announcements
138 /// - More efficient than `into_elem_iter()` as it doesn't duplicate attributes per prefix
139 ///
140 /// The iterator returns an `MrtUpdate` enum with variants:
141 /// - `Bgp4MpUpdate`: BGP UPDATE messages from UPDATES files
142 /// - `LegacyBgpUpdate`: Deprecated MRT Type 5 BGP UPDATE messages
143 /// - `TableDumpV2Entry`: RIB entries from TableDumpV2 RIB dumps
144 /// - `TableDumpMessage`: Legacy TableDump v1 messages
145 ///
146 /// # Example
147 /// ```no_run
148 /// use bgpkit_parser::{BgpkitParser, MrtUpdate};
149 ///
150 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
151 /// for update in parser.into_update_iter() {
152 /// match update {
153 /// MrtUpdate::Bgp4MpUpdate(u) => {
154 /// println!("Peer {} announced {} prefixes",
155 /// u.peer_ip,
156 /// u.message.announced_prefixes.len()
157 /// );
158 /// }
159 /// MrtUpdate::LegacyBgpUpdate(u) => {
160 /// println!("Legacy UPDATE from peer {}", u.peer_ip);
161 /// }
162 /// MrtUpdate::TableDumpV2Entry(e) => {
163 /// println!("RIB entry for {} with {} peers",
164 /// e.prefix,
165 /// e.rib_entries.len()
166 /// );
167 /// }
168 /// MrtUpdate::TableDumpMessage(m) => {
169 /// println!("Legacy table dump for {}", m.prefix);
170 /// }
171 /// }
172 /// }
173 /// ```
174 pub fn into_update_iter(self) -> UpdateIterator<R> {
175 UpdateIterator::new(self)
176 }
177
178 /// Creates an iterator over lightweight route elements from MRT data.
179 ///
180 /// This iterator yields [`BgpRouteElem`](crate::models::BgpRouteElem)
181 /// values and only parses route identity, peer metadata, timestamp, and
182 /// AS path. Use [`into_elem_iter`](Self::into_elem_iter) when you need
183 /// the full [`BgpElem`] attribute set. Filters that only depend on route
184 /// fields are supported; `community` filters do not match route elements.
185 pub fn into_route_iter(self) -> RouteIterator<R> {
186 RouteIterator::new(self)
187 }
188
189 /// Creates a fallible iterator over MRT records that returns parsing errors.
190 ///
191 /// # Example
192 /// ```no_run
193 /// use bgpkit_parser::BgpkitParser;
194 ///
195 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
196 /// for result in parser.into_fallible_record_iter() {
197 /// match result {
198 /// Ok(record) => {
199 /// // Process the record
200 /// }
201 /// Err(e) => {
202 /// // Handle the error
203 /// eprintln!("Error parsing record: {}", e);
204 /// }
205 /// }
206 /// }
207 /// ```
208 pub fn into_fallible_record_iter(self) -> FallibleRecordIterator<R> {
209 FallibleRecordIterator::new(self)
210 }
211
212 /// Creates a fallible iterator over BGP elements that returns parsing errors.
213 ///
214 /// # Example
215 /// ```no_run
216 /// use bgpkit_parser::BgpkitParser;
217 ///
218 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
219 /// for result in parser.into_fallible_elem_iter() {
220 /// match result {
221 /// Ok(elem) => {
222 /// // Process the element
223 /// }
224 /// Err(e) => {
225 /// // Handle the error
226 /// eprintln!("Error parsing element: {}", e);
227 /// }
228 /// }
229 /// }
230 /// ```
231 pub fn into_fallible_elem_iter(self) -> FallibleElemIterator<R> {
232 FallibleElemIterator::new(self)
233 }
234
235 /// Creates a fallible iterator over BGP announcements that returns parsing errors.
236 ///
237 /// Unlike the default `into_update_iter()`, this iterator returns
238 /// `Result<MrtUpdate, ParserErrorWithBytes>` allowing users to handle parsing
239 /// errors explicitly instead of having them logged and skipped.
240 ///
241 /// # Example
242 /// ```no_run
243 /// use bgpkit_parser::{BgpkitParser, MrtUpdate};
244 ///
245 /// let parser = BgpkitParser::new("updates.mrt").unwrap();
246 /// for result in parser.into_fallible_update_iter() {
247 /// match result {
248 /// Ok(MrtUpdate::Bgp4MpUpdate(update)) => {
249 /// println!("Peer {} announced {} prefixes",
250 /// update.peer_ip,
251 /// update.message.announced_prefixes.len()
252 /// );
253 /// }
254 /// Ok(_) => { /* handle other variants */ }
255 /// Err(e) => {
256 /// eprintln!("Error parsing: {}", e);
257 /// }
258 /// }
259 /// }
260 /// ```
261 pub fn into_fallible_update_iter(self) -> FallibleUpdateIterator<R> {
262 FallibleUpdateIterator::new(self)
263 }
264
265 /// Creates a fallible iterator over lightweight route elements.
266 pub fn into_fallible_route_iter(self) -> FallibleRouteIterator<R> {
267 FallibleRouteIterator::new(self)
268 }
269
270 /// Creates an iterator that classifies each MRT record for malformed-data investigation.
271 ///
272 /// This iterator emits every record with its raw bytes attached: clean
273 /// records have empty warning lists, records with recoverable RFC 7606
274 /// validation findings carry them in `warnings`, and fatal parse errors
275 /// retain the consumed bytes plus a best-effort partial dissection tree
276 /// showing where parsing stopped. It ignores parser filters so that
277 /// malformed records cannot be hidden by element-oriented matching. Text-dump parsers yield
278 /// no diagnostic events because they have no MRT record representation.
279 ///
280 /// # Example
281 /// ```no_run
282 /// use bgpkit_parser::{BgpkitParser, DiagnosticEvent};
283 ///
284 /// for event in BgpkitParser::new("updates.mrt")?.into_diagnostic_iter() {
285 /// match event {
286 /// DiagnosticEvent::Record { record, raw, warnings } => {
287 /// if warnings.is_empty() {
288 /// println!("{record}");
289 /// } else {
290 /// eprintln!("validation findings: {warnings:?}");
291 /// raw.write_raw_bytes("malformed-record.mrt")?;
292 /// }
293 /// }
294 /// DiagnosticEvent::ParseError { error, raw_bytes, .. } => {
295 /// eprintln!("parse error: {error}");
296 /// if let Some(raw_bytes) = raw_bytes {
297 /// std::fs::write("malformed-record.mrt", raw_bytes)?;
298 /// }
299 /// }
300 /// _ => {}
301 /// }
302 /// }
303 /// # Ok::<(), Box<dyn std::error::Error>>(())
304 /// ```
305 pub fn into_diagnostic_iter(self) -> DiagnosticIterator<R> {
306 DiagnosticIterator::new(self)
307 }
308
309 /// Creates an Elementor pre-initialized with PeerIndexTable and an iterator over raw records.
310 ///
311 /// This is useful for parallel processing where the Elementor needs to be shared across threads.
312 /// The Elementor is created with the PeerIndexTable from the first record if present,
313 /// otherwise a new Elementor is created.
314 ///
315 /// # Example
316 /// See the `parallel_records_to_elem` example for full usage.
317 /// ```ignore
318 /// use bgpkit_parser::BgpkitParser;
319 ///
320 /// let parser = BgpkitParser::new_cached(url, "/tmp")?;
321 /// let (elementor, records) = parser.into_elementor_and_raw_record_iter();
322 /// ```
323 ///
324 pub fn into_elementor_and_raw_record_iter(
325 self,
326 ) -> (Elementor, impl Iterator<Item = RawMrtRecord>)
327 where
328 R: Read,
329 {
330 let mut raw_iter = RawRecordIterator::new(self).peekable();
331 let elementor = match raw_iter.peek().cloned().and_then(|r| r.parse().ok()) {
332 Some(MrtRecord {
333 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(pit)),
334 ..
335 }) => {
336 raw_iter.next();
337 Elementor::with_peer_table(pit)
338 }
339 _ => Elementor::new(),
340 };
341 (elementor, raw_iter)
342 }
343
344 /// Creates an Elementor pre-initialized with PeerIndexTable and an iterator over parsed records.
345 ///
346 /// This is useful for parallel processing where the Elementor needs to be shared across threads.
347 /// The Elementor is created with the PeerIndexTable from the first record if present,
348 /// otherwise a new Elementor is created.
349 ///
350 /// # Example
351 /// See the `parallel_records_to_elem` example for full usage.
352 pub fn into_elementor_and_record_iter(self) -> (Elementor, impl Iterator<Item = MrtRecord>)
353 where
354 R: Read,
355 {
356 let mut record_iter = RecordIterator::new(self).peekable();
357 let elementor = match record_iter.peek().cloned() {
358 Some(MrtRecord {
359 message: MrtMessage::TableDumpV2Message(TableDumpV2Message::PeerIndexTable(pit)),
360 ..
361 }) => {
362 record_iter.next();
363 Elementor::with_peer_table(pit)
364 }
365 _ => Elementor::new(),
366 };
367 (elementor, record_iter)
368 }
369}