fixed stop and error detection

src/parsers/datetime.rs

@@ -1,5 +1,5 @@
 use anyhow::Result;
-use chrono::{DateTime, TimeZone, Utc};
+use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -8,24 +8,27 @@ pub enum Error {
     InvalidDateTimeString(String),
 }
 
-/// Parse date time string into a duration. A timezone can be
-/// specified as a parameter. If no timezone is specified, UTC is
-/// assumed. The result is the duration in seconds.
+/// Parse date time string.
+/// TODO: A timezone should be specified as a parameter.
 pub fn parse(s: &str) -> Result<DateTime<Utc>> {
     let parts: Vec<&str> = s.split_whitespace().collect();
     let now = Utc::now();
     let date = format!("{}", now.format("%Y-%m-%d"));
+
     let (date, time) = match parts.len() {
         1 => (date.as_ref(), parts[0]),
         2 => (parts[0], parts[1]),
         _ => return Err(Error::InvalidDateTimeString(s.to_string()).into()),
     };
 
+    let parts = time.split('.').collect::<Vec<&str>>();
+    let time = parts[0];
+
     // Parse date time string
     let datetime = format!("{} {}", date, time);
-    let datetime = Utc
-        .datetime_from_str(datetime.as_ref(), "%Y-%m-%d %H:%M:%S")
-        .map_err(|_| Error::InvalidDateTimeString(s.to_string()))?;
+    let datetime =
+        NaiveDateTime::parse_from_str(datetime.as_ref(), "%Y-%m-%d %H:%M:%S")?;
+    let datetime = Utc.from_utc_datetime(&datetime);
 
     Ok(datetime)
 }
@@ -64,6 +67,9 @@ mod tests {
         assert_eq!(result.hour(), 10);
         assert_eq!(result.minute(), 42);
         assert_eq!(result.second(), 11);
+
+        let result = parse("10:42:11.123").unwrap();
+        assert_eq!(result.hour(), 10);
     }
 
     #[test]
@@ -78,9 +84,10 @@ mod tests {
     #[test]
     fn test_parse_duration_sec() {
         let fiveminutesago = Utc::now() - Duration::minutes(5);
-        let result =
-            parse_duration_sec(format!("{}", fiveminutesago.format("%Y-%m-%d %H:%M:%S")).as_ref())
-                .unwrap();
+        let result = parse_duration_sec(
+            format!("{}", fiveminutesago.format("%Y-%m-%d %H:%M:%S")).as_ref(),
+        )
+        .unwrap();
         assert_eq!(result, 300.0);
     }
 }

src/parsers/parser.rs

@@ -111,16 +111,27 @@ impl<R: BufRead> Iterator for BlockIterator<R> {
         // Create a peekable line iterator
         loop {
             // Read next line
-            let line = self.lines.next()?;
-            // Add line to block
-            let line = line.ok()?;
+            let line = self.lines.next()?.ok()?;
+
+            // Check stop marker
+            if line.starts_with("0000") {
+                return None;
+            }
+            if line.starts_with("9001") {
+                println!("ERROR: {}", line);
+                return None;
+            }
+
             block.push(line.clone());
 
             // Check next line in iterator
             if let Some(Ok(next)) = self.lines.peek() {
                 if self.start.is_match(next) {
                     break;
                 }
+                if next.starts_with("0000") {
+                    break;
+                }
             } else {
                 break; // EOF
             }

src/parsers/routes.rs

@@ -18,7 +18,7 @@ lazy_static! {
           (?P<prefix>[0-9a-f:\./]+)?\s+   # Network
           (?P<type>\w+)\s+ 
           \[
-            (?P<from_protocol>.*?)\s+(?P<age>[\d\-:\s]+)
+            (?P<from_protocol>.*?)\s+(?P<age>[\d\-:\.\s]+)
             (\s+from\s+(?P<learnt_from>.+))?
           \]\s+
           ((?P<primary>\*)\s+)?
@@ -87,6 +87,10 @@ impl Parse for PrefixGroup {
             } else {
                 prefix = route.network.clone();
             }
+
+            if route.neighbor_id.is_none() {
+                continue; // ??
+            }
             routes.push(route);
         }
 
@@ -143,6 +147,9 @@ fn parse_route_header(route: &mut Route, line: &str) -> Result<State> {
         if let Some(from) = caps.name("learnt_from") {
             route.learnt_from = Some(from.as_str().to_string());
         }
+        if let Some(proto) = caps.name("from_protocol") {
+            route.neighbor_id = Some(proto.as_str().to_string());
+        }
 
         return Ok(State::Meta);
     }
@@ -327,6 +334,10 @@ mod tests {
             "1007-203.17.254.0/24      unicast [R192_172 2023-04-19 09:28:42] * (100) [AS7545i]";
         let caps = RE_ROUTE_HEADER.captures(line).unwrap();
         println!("{:?}", caps);
+
+        let line = "1007-219.0.0.0/9          unicast [ebgp_rs4 10:38:20.602 from 10.255.253.250] * (100) [AS64967i]";
+        let caps = RE_ROUTE_HEADER.captures(line).unwrap();
+        println!("{:?}", caps);
     }
 
     #[test]

src/parsers/routes_worker.rs

@@ -1,10 +1,12 @@
+use std::{
+    num::NonZeroUsize,
+    sync::{Arc, Mutex},
+    thread,
+};
+
 use anyhow::Result;
-use std::num::NonZeroUsize;
-use std::sync::{Arc, Mutex};
-use std::thread;
-use tokio::sync::mpsc::error::TryRecvError;
 use tokio::sync::mpsc::{
-    unbounded_channel, UnboundedReceiver, UnboundedSender,
+    error::TryRecvError, unbounded_channel, UnboundedReceiver, UnboundedSender,
 };
 
 use crate::parsers::{
@@ -28,8 +30,7 @@ impl RoutesWorker {
     /// Spawn a new routes worker and create a response and
     /// request channel.
     pub fn spawn(&self, block_queue: BlockQueue, results_queue: ResultsQueue) {
-        println!("routes worker {} started.", self.id);
-
+        tracing::debug!("routes worker {} started.", self.id);
         thread::spawn(move || loop {
             let block = {
                 let mut queue = block_queue.lock().unwrap();
@@ -71,7 +72,10 @@ impl RoutesWorkerPool {
         let parallelism = std::thread::available_parallelism()
             .unwrap_or(NonZeroUsize::new(1).unwrap());
         let num_workers = parallelism.get();
-        println!("Starting routes worker pool with {} workers.", num_workers);
+        tracing::info!(
+            "Starting routes worker pool with {} workers.",
+            num_workers
+        );
 
         // Start workers
         for id in 0..num_workers {
@@ -97,7 +101,7 @@ mod tests {
     async fn test_routes_worker() {
         let file =
             File::open("tests/birdc/show-route-all-protocol-R1").unwrap();
-        // let file: File = File::open("tests/birdc/show-route-all-table-master4").unwrap();
+        // let file: File =             File::open("tests/birdc/show-route-all-table-master4").unwrap();
         let reader = BufReader::new(file);
         let re_routes_start = Regex::new(r"1007-\S").unwrap();