1use crate::models::*;
48use crate::parser::iters::handle_record_parse_error;
49use crate::parser::BgpkitParser;
50use crate::Elementor;
51use std::io::Read;
52use std::net::IpAddr;
53
54#[derive(Debug, Clone, PartialEq)]
60#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
61pub struct Bgp4MpUpdate {
62 pub timestamp: f64,
64 pub peer_ip: IpAddr,
66 pub peer_asn: Asn,
68 pub message: BgpUpdateMessage,
70}
71
72#[derive(Debug, Clone, PartialEq)]
80#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
81pub struct LegacyBgpUpdate {
82 pub timestamp: f64,
83 pub peer_ip: IpAddr,
84 pub peer_asn: Asn,
85 pub local_ip: IpAddr,
86 pub local_asn: Asn,
87 pub message: BgpUpdateMessage,
88}
89
90#[derive(Debug, Clone, PartialEq)]
95#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
96pub struct TableDumpV2Entry {
97 pub timestamp: f64,
99 pub rib_type: TableDumpV2Type,
101 pub sequence_number: u32,
103 pub prefix: NetworkPrefix,
105 pub rib_entries: Vec<RibEntry>,
108}
109
110#[derive(Debug, Clone, PartialEq)]
117#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
118pub enum MrtUpdate {
119 Bgp4MpUpdate(Bgp4MpUpdate),
121 LegacyBgpUpdate(LegacyBgpUpdate),
123 TableDumpV2Entry(TableDumpV2Entry),
125 TableDumpMessage(TableDumpMessage),
127}
128
129impl MrtUpdate {
130 pub fn timestamp(&self) -> f64 {
132 match self {
133 MrtUpdate::Bgp4MpUpdate(u) => u.timestamp,
134 MrtUpdate::LegacyBgpUpdate(u) => u.timestamp,
135 MrtUpdate::TableDumpV2Entry(e) => e.timestamp,
136 MrtUpdate::TableDumpMessage(m) => m.originated_time as f64,
137 }
138 }
139}
140
141pub struct UpdateIterator<R> {
150 parser: BgpkitParser<R>,
151 elementor: Elementor,
152 pending_table_dump: Vec<TableDumpMessage>,
153}
154
155impl<R> UpdateIterator<R> {
156 pub(crate) fn new(parser: BgpkitParser<R>) -> Self {
157 UpdateIterator {
158 parser,
159 elementor: Elementor::new(),
160 pending_table_dump: Vec::new(),
161 }
162 }
163}
164
165impl<R: Read> Iterator for UpdateIterator<R> {
166 type Item = MrtUpdate;
167
168 fn next(&mut self) -> Option<MrtUpdate> {
169 if self.parser.text_dump_iter.is_some() {
172 return None;
173 }
174 loop {
175 if let Some(message) = self.pending_table_dump.pop() {
176 return Some(MrtUpdate::TableDumpMessage(message));
177 }
178 let record = match self.parser.next_record() {
179 Ok(record) => record,
180 Err(e) => {
181 if !handle_record_parse_error(&mut self.parser, e.error, e.bytes) {
182 return None;
183 }
184 continue;
185 }
186 };
187
188 let t = record.common_header.timestamp;
189 let timestamp: f64 = if let Some(micro) = &record.common_header.microsecond_timestamp {
190 let m = (*micro as f64) / 1_000_000.0;
191 t as f64 + m
192 } else {
193 f64::from(t)
194 };
195
196 match record.message {
197 MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(msg)) => {
198 if let BgpMessage::Update(update) = msg.bgp_message {
199 return Some(MrtUpdate::Bgp4MpUpdate(Bgp4MpUpdate {
200 timestamp,
201 peer_ip: msg.peer_ip,
202 peer_asn: msg.peer_asn,
203 message: update,
204 }));
205 }
206 continue;
208 }
209 MrtMessage::Bgp4Mp(Bgp4MpEnum::StateChange(_)) => {
210 continue;
212 }
213 MrtMessage::LegacyBgp(LegacyBgp::Message(msg)) => {
214 if let BgpMessage::Update(update) = msg.bgp_message {
215 return Some(MrtUpdate::LegacyBgpUpdate(LegacyBgpUpdate {
216 timestamp,
217 peer_ip: msg.peer_ip,
218 peer_asn: msg.peer_asn,
219 local_ip: msg.local_ip,
220 local_asn: msg.local_asn,
221 message: update,
222 }));
223 }
224 continue;
225 }
226 MrtMessage::LegacyBgp(LegacyBgp::StateChange(_)) => continue,
227 MrtMessage::TableDumpV2Message(msg) => {
228 match msg {
229 TableDumpV2Message::PeerIndexTable(p) => {
230 self.elementor.peer_table = Some(p);
232 continue;
233 }
234 TableDumpV2Message::RibAfi(entries) => {
235 return Some(MrtUpdate::TableDumpV2Entry(TableDumpV2Entry {
236 timestamp,
237 rib_type: entries.rib_type,
238 sequence_number: entries.sequence_number,
239 prefix: entries.prefix,
240 rib_entries: entries.rib_entries,
241 }));
242 }
243 TableDumpV2Message::RibGeneric(_) => {
244 continue;
246 }
247 TableDumpV2Message::GeoPeerTable(_) => {
248 continue;
250 }
251 }
252 }
253 MrtMessage::TableDumpMessage(msg) => {
254 return Some(MrtUpdate::TableDumpMessage(msg));
255 }
256 MrtMessage::TableDumpMessageBatch(mut messages) => {
257 messages.reverse();
258 self.pending_table_dump = messages;
259 continue;
260 }
261 }
262 }
263 }
264}
265
266pub struct FallibleUpdateIterator<R> {
271 parser: BgpkitParser<R>,
272 elementor: Elementor,
273 pending_table_dump: Vec<TableDumpMessage>,
274 finished: bool,
275}
276
277impl<R> FallibleUpdateIterator<R> {
278 pub(crate) fn new(parser: BgpkitParser<R>) -> Self {
279 FallibleUpdateIterator {
280 parser,
281 elementor: Elementor::new(),
282 pending_table_dump: Vec::new(),
283 finished: false,
284 }
285 }
286}
287
288impl<R: Read> Iterator for FallibleUpdateIterator<R> {
289 type Item = Result<MrtUpdate, crate::error::ParserErrorWithBytes>;
290
291 fn next(&mut self) -> Option<Self::Item> {
292 if self.parser.text_dump_iter.is_some() {
294 return None;
295 }
296 loop {
297 if let Some(message) = self.pending_table_dump.pop() {
298 return Some(Ok(MrtUpdate::TableDumpMessage(message)));
299 }
300 match super::next_fallible_record(&mut self.parser, &mut self.finished) {
301 Some(Ok(record)) => {
302 let t = record.common_header.timestamp;
303 let timestamp: f64 =
304 if let Some(micro) = &record.common_header.microsecond_timestamp {
305 let m = (*micro as f64) / 1_000_000.0;
306 t as f64 + m
307 } else {
308 f64::from(t)
309 };
310
311 match record.message {
312 MrtMessage::Bgp4Mp(Bgp4MpEnum::Message(msg)) => {
313 if let BgpMessage::Update(update) = msg.bgp_message {
314 return Some(Ok(MrtUpdate::Bgp4MpUpdate(Bgp4MpUpdate {
315 timestamp,
316 peer_ip: msg.peer_ip,
317 peer_asn: msg.peer_asn,
318 message: update,
319 })));
320 }
321 continue;
322 }
323 MrtMessage::Bgp4Mp(Bgp4MpEnum::StateChange(_)) => {
324 continue;
325 }
326 MrtMessage::LegacyBgp(LegacyBgp::Message(msg)) => {
327 if let BgpMessage::Update(update) = msg.bgp_message {
328 return Some(Ok(MrtUpdate::LegacyBgpUpdate(LegacyBgpUpdate {
329 timestamp,
330 peer_ip: msg.peer_ip,
331 peer_asn: msg.peer_asn,
332 local_ip: msg.local_ip,
333 local_asn: msg.local_asn,
334 message: update,
335 })));
336 }
337 continue;
338 }
339 MrtMessage::LegacyBgp(LegacyBgp::StateChange(_)) => continue,
340 MrtMessage::TableDumpV2Message(msg) => match msg {
341 TableDumpV2Message::PeerIndexTable(p) => {
342 self.elementor.peer_table = Some(p);
343 continue;
344 }
345 TableDumpV2Message::RibAfi(entries) => {
346 return Some(Ok(MrtUpdate::TableDumpV2Entry(TableDumpV2Entry {
347 timestamp,
348 rib_type: entries.rib_type,
349 sequence_number: entries.sequence_number,
350 prefix: entries.prefix,
351 rib_entries: entries.rib_entries,
352 })));
353 }
354 TableDumpV2Message::RibGeneric(_) => {
355 continue;
356 }
357 TableDumpV2Message::GeoPeerTable(_) => {
358 continue;
359 }
360 },
361 MrtMessage::TableDumpMessage(msg) => {
362 return Some(Ok(MrtUpdate::TableDumpMessage(msg)));
363 }
364 MrtMessage::TableDumpMessageBatch(mut messages) => {
365 messages.reverse();
366 self.pending_table_dump = messages;
367 continue;
368 }
369 }
370 }
371 None => return None,
372 Some(Err(e)) => {
373 return Some(Err(e));
374 }
375 }
376 }
377 }
378}
379
380#[cfg(test)]
381mod tests {
382 use super::*;
383 use std::io::Cursor;
384
385 #[test]
386 fn test_bgp4mp_update_struct() {
387 let update = Bgp4MpUpdate {
388 timestamp: 1234567890.123456,
389 peer_ip: "192.0.2.1".parse().unwrap(),
390 peer_asn: Asn::new_32bit(65000),
391 message: BgpUpdateMessage::default(),
392 };
393
394 assert_eq!(update.timestamp, 1234567890.123456);
395 assert_eq!(update.peer_ip.to_string(), "192.0.2.1");
396 assert_eq!(update.peer_asn, Asn::new_32bit(65000));
397 }
398
399 #[test]
400 fn test_table_dump_v2_entry_struct() {
401 let entry = TableDumpV2Entry {
402 timestamp: 1234567890.0,
403 rib_type: TableDumpV2Type::RibIpv4Unicast,
404 sequence_number: 42,
405 prefix: "10.0.0.0/8".parse().unwrap(),
406 rib_entries: vec![],
407 };
408
409 assert_eq!(entry.timestamp, 1234567890.0);
410 assert_eq!(entry.rib_type, TableDumpV2Type::RibIpv4Unicast);
411 assert_eq!(entry.sequence_number, 42);
412 assert_eq!(entry.prefix.to_string(), "10.0.0.0/8");
413 assert!(entry.rib_entries.is_empty());
414 }
415
416 #[test]
417 fn test_mrt_update_timestamp() {
418 let bgp4mp = MrtUpdate::Bgp4MpUpdate(Bgp4MpUpdate {
420 timestamp: 1234567890.5,
421 peer_ip: "192.0.2.1".parse().unwrap(),
422 peer_asn: Asn::new_32bit(65000),
423 message: BgpUpdateMessage::default(),
424 });
425 assert_eq!(bgp4mp.timestamp(), 1234567890.5);
426
427 let table_dump_v2 = MrtUpdate::TableDumpV2Entry(TableDumpV2Entry {
429 timestamp: 1234567891.5,
430 rib_type: TableDumpV2Type::RibIpv4Unicast,
431 sequence_number: 1,
432 prefix: "10.0.0.0/8".parse().unwrap(),
433 rib_entries: vec![],
434 });
435 assert_eq!(table_dump_v2.timestamp(), 1234567891.5);
436
437 let table_dump_v1 = MrtUpdate::TableDumpMessage(TableDumpMessage {
439 view_number: 0,
440 sequence_number: 1,
441 prefix: "192.168.0.0/16".parse().unwrap(),
442 status: 1,
443 originated_time: 1234567892,
444 peer_ip: "10.0.0.1".parse().unwrap(),
445 peer_asn: Asn::new_32bit(65001),
446 attributes: Attributes::default(),
447 });
448 assert_eq!(table_dump_v1.timestamp(), 1234567892.0);
449 }
450
451 #[test]
452 fn test_update_iterator_empty() {
453 let cursor = Cursor::new(vec![]);
454 let parser = BgpkitParser::from_reader(cursor);
455 let mut iter = UpdateIterator::new(parser);
456
457 assert!(iter.next().is_none());
458 }
459
460 #[test]
461 fn test_fallible_update_iterator_empty() {
462 let cursor = Cursor::new(vec![]);
463 let parser = BgpkitParser::from_reader(cursor);
464 let mut iter = FallibleUpdateIterator::new(parser);
465
466 assert!(iter.next().is_none());
467 }
468
469 #[test]
470 fn test_bgp4mp_update_clone_and_debug() {
471 let update = Bgp4MpUpdate {
472 timestamp: 1234567890.123456,
473 peer_ip: "192.0.2.1".parse().unwrap(),
474 peer_asn: Asn::new_32bit(65000),
475 message: BgpUpdateMessage::default(),
476 };
477
478 let cloned = update.clone();
480 assert_eq!(update, cloned);
481
482 let debug_str = format!("{:?}", update);
484 assert!(debug_str.contains("Bgp4MpUpdate"));
485 assert!(debug_str.contains("192.0.2.1"));
486 }
487
488 #[test]
489 fn test_table_dump_v2_entry_clone_and_debug() {
490 let entry = TableDumpV2Entry {
491 timestamp: 1234567890.0,
492 rib_type: TableDumpV2Type::RibIpv4Unicast,
493 sequence_number: 42,
494 prefix: "10.0.0.0/8".parse().unwrap(),
495 rib_entries: vec![],
496 };
497
498 let cloned = entry.clone();
500 assert_eq!(entry, cloned);
501
502 let debug_str = format!("{:?}", entry);
504 assert!(debug_str.contains("TableDumpV2Entry"));
505 assert!(debug_str.contains("10.0.0.0/8"));
506 }
507
508 #[test]
509 fn test_mrt_update_clone_and_debug() {
510 let update = MrtUpdate::Bgp4MpUpdate(Bgp4MpUpdate {
511 timestamp: 1234567890.5,
512 peer_ip: "192.0.2.1".parse().unwrap(),
513 peer_asn: Asn::new_32bit(65000),
514 message: BgpUpdateMessage::default(),
515 });
516
517 let cloned = update.clone();
519 assert_eq!(update, cloned);
520
521 let debug_str = format!("{:?}", update);
523 assert!(debug_str.contains("Bgp4MpUpdate"));
524 }
525
526 #[test]
527 fn test_fallible_update_iterator_with_invalid_data() {
528 let invalid_data = vec![
530 0x00, 0x00, 0x00, 0x00, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, ];
536
537 let cursor = Cursor::new(invalid_data);
538 let parser = BgpkitParser::from_reader(cursor);
539 let mut iter = FallibleUpdateIterator::new(parser);
540
541 let result = iter.next();
543 assert!(result.is_some());
544 assert!(result.unwrap().is_err());
545 }
546
547 #[test]
548 fn test_mrt_update_enum_variants() {
549 let updates: Vec<MrtUpdate> = vec![
551 MrtUpdate::Bgp4MpUpdate(Bgp4MpUpdate {
552 timestamp: 1.0,
553 peer_ip: "192.0.2.1".parse().unwrap(),
554 peer_asn: Asn::new_32bit(65000),
555 message: BgpUpdateMessage::default(),
556 }),
557 MrtUpdate::TableDumpV2Entry(TableDumpV2Entry {
558 timestamp: 2.0,
559 rib_type: TableDumpV2Type::RibIpv6Unicast,
560 sequence_number: 1,
561 prefix: "2001:db8::/32".parse().unwrap(),
562 rib_entries: vec![],
563 }),
564 MrtUpdate::TableDumpMessage(TableDumpMessage {
565 view_number: 0,
566 sequence_number: 1,
567 prefix: "10.0.0.0/8".parse().unwrap(),
568 status: 1,
569 originated_time: 3,
570 peer_ip: "10.0.0.1".parse().unwrap(),
571 peer_asn: Asn::new_32bit(65001),
572 attributes: Attributes::default(),
573 }),
574 ];
575
576 for (i, update) in updates.iter().enumerate() {
577 match update {
578 MrtUpdate::Bgp4MpUpdate(_) => assert_eq!(i, 0),
579 MrtUpdate::LegacyBgpUpdate(_) => panic!("not part of this fixture"),
580 MrtUpdate::TableDumpV2Entry(_) => assert_eq!(i, 1),
581 MrtUpdate::TableDumpMessage(_) => assert_eq!(i, 2),
582 }
583 }
584 }
585
586 #[test]
587 #[cfg(feature = "serde")]
588 fn test_bgp4mp_update_serde() {
589 let update = Bgp4MpUpdate {
590 timestamp: 1234567890.123456,
591 peer_ip: "192.0.2.1".parse().unwrap(),
592 peer_asn: Asn::new_32bit(65000),
593 message: BgpUpdateMessage::default(),
594 };
595
596 let serialized = serde_json::to_string(&update).unwrap();
597 let deserialized: Bgp4MpUpdate = serde_json::from_str(&serialized).unwrap();
598 assert_eq!(update, deserialized);
599 }
600
601 #[test]
602 #[cfg(feature = "serde")]
603 fn test_table_dump_v2_entry_serde() {
604 let entry = TableDumpV2Entry {
605 timestamp: 1234567890.0,
606 rib_type: TableDumpV2Type::RibIpv4Unicast,
607 sequence_number: 42,
608 prefix: "10.0.0.0/8".parse().unwrap(),
609 rib_entries: vec![],
610 };
611
612 let serialized = serde_json::to_string(&entry).unwrap();
613 let deserialized: TableDumpV2Entry = serde_json::from_str(&serialized).unwrap();
614 assert_eq!(entry, deserialized);
615 }
616
617 #[test]
618 #[cfg(feature = "serde")]
619 fn test_mrt_update_serde() {
620 let update = MrtUpdate::Bgp4MpUpdate(Bgp4MpUpdate {
621 timestamp: 1234567890.5,
622 peer_ip: "192.0.2.1".parse().unwrap(),
623 peer_asn: Asn::new_32bit(65000),
624 message: BgpUpdateMessage::default(),
625 });
626
627 let serialized = serde_json::to_string(&update).unwrap();
628 let deserialized: MrtUpdate = serde_json::from_str(&serialized).unwrap();
629 assert_eq!(update, deserialized);
630 }
631
632 #[test]
634 fn test_update_iterator_with_updates_file() {
635 let url = "https://spaces.bgpkit.org/parser/update-example";
636 let parser = BgpkitParser::new(url).unwrap();
637
638 let mut bgp4mp_count = 0;
639 let mut total_announced = 0;
640 let mut total_withdrawn = 0;
641
642 for update in parser.into_update_iter() {
643 match update {
644 MrtUpdate::Bgp4MpUpdate(u) => {
645 bgp4mp_count += 1;
646 total_announced += u.message.announced_prefixes.len();
647 total_withdrawn += u.message.withdrawn_prefixes.len();
648 for attr in &u.message.attributes {
650 match attr {
651 AttributeValue::MpReachNlri(nlri) => {
652 total_announced += nlri.prefixes.len();
653 }
654 AttributeValue::MpUnreachNlri(nlri) => {
655 total_withdrawn += nlri.prefixes.len();
656 }
657 _ => {}
658 }
659 }
660 }
661 MrtUpdate::TableDumpV2Entry(_) => {
662 panic!("Should not see TableDumpV2Entry in UPDATES file");
663 }
664 MrtUpdate::TableDumpMessage(_) => {
665 panic!("Should not see TableDumpMessage in UPDATES file");
666 }
667 MrtUpdate::LegacyBgpUpdate(_) => {
668 panic!("Should not see a legacy BGP update in this modern fixture");
669 }
670 }
671 }
672
673 assert!(bgp4mp_count > 0, "Should have parsed some BGP4MP updates");
675 assert!(
676 total_announced + total_withdrawn > 0,
677 "Should have some prefixes"
678 );
679 }
680
681 #[test]
683 fn test_update_iterator_with_rib_file() {
684 let url = "https://spaces.bgpkit.org/parser/rib-example-small.bz2";
685 let parser = BgpkitParser::new(url).unwrap();
686
687 let mut rib_entry_count = 0;
688 let mut total_rib_entries = 0;
689
690 for update in parser.into_update_iter().take(100) {
691 match update {
692 MrtUpdate::Bgp4MpUpdate(_) => {
693 panic!("Should not see Bgp4MpUpdate in RIB file");
694 }
695 MrtUpdate::LegacyBgpUpdate(_) => {
696 panic!("Should not see LegacyBgpUpdate in RIB file");
697 }
698 MrtUpdate::TableDumpV2Entry(e) => {
699 rib_entry_count += 1;
700 total_rib_entries += e.rib_entries.len();
701 assert!(e.sequence_number > 0 || rib_entry_count == 1);
703 }
704 MrtUpdate::TableDumpMessage(_) => {
705 }
707 }
708 }
709
710 assert!(rib_entry_count > 0, "Should have parsed some RIB entries");
712 assert!(
713 total_rib_entries > 0,
714 "Should have some RIB entries per prefix"
715 );
716 }
717
718 #[test]
720 fn test_fallible_update_iterator_with_updates_file() {
721 let url = "https://spaces.bgpkit.org/parser/update-example";
722 let parser = BgpkitParser::new(url).unwrap();
723
724 let mut success_count = 0;
725 let mut error_count = 0;
726
727 for result in parser.into_fallible_update_iter() {
728 match result {
729 Ok(_) => success_count += 1,
730 Err(_) => error_count += 1,
731 }
732 }
733
734 assert!(
735 success_count > 0,
736 "Should have parsed some updates successfully"
737 );
738 assert_eq!(
740 error_count, 0,
741 "Should have no parsing errors in valid file"
742 );
743 }
744
745 #[test]
747 fn test_update_iter_vs_elem_iter_consistency() {
748 let url = "https://spaces.bgpkit.org/parser/update-example";
749
750 let parser1 = BgpkitParser::new(url).unwrap();
752 let mut update_iter_announced = 0;
753 let mut update_iter_withdrawn = 0;
754
755 for update in parser1.into_update_iter() {
756 if let MrtUpdate::Bgp4MpUpdate(u) = update {
757 update_iter_announced += u.message.announced_prefixes.len();
758 update_iter_withdrawn += u.message.withdrawn_prefixes.len();
759 for attr in &u.message.attributes {
760 match attr {
761 AttributeValue::MpReachNlri(nlri) => {
762 update_iter_announced += nlri.prefixes.len();
763 }
764 AttributeValue::MpUnreachNlri(nlri) => {
765 update_iter_withdrawn += nlri.prefixes.len();
766 }
767 _ => {}
768 }
769 }
770 }
771 }
772
773 let parser2 = BgpkitParser::new(url).unwrap();
775 let mut elem_iter_announced = 0;
776 let mut elem_iter_withdrawn = 0;
777
778 for elem in parser2.into_elem_iter() {
779 match elem.elem_type {
780 ElemType::ANNOUNCE => elem_iter_announced += 1,
781 ElemType::WITHDRAW => elem_iter_withdrawn += 1,
782 }
783 }
784
785 assert_eq!(
787 update_iter_announced, elem_iter_announced,
788 "Announced prefix counts should match"
789 );
790 assert_eq!(
791 update_iter_withdrawn, elem_iter_withdrawn,
792 "Withdrawn prefix counts should match"
793 );
794 }
795}