Switch over error types to use thiserror

Cargo.toml

@@ -28,6 +28,7 @@ serde={version="1.0", features=["derive"], optional = true}
 itertools = {version = "0.11.0", optional = true}
 ipnet = {version="2.7", optional = true}
 bitflags = {version="2.3.3", features = ["serde"], optional = true}
+thiserror = {version = "1.0.44", optional = true}
 
 #######################
 # Parser dependencies #
@@ -54,6 +55,7 @@ models = [
     "ipnet",
     "itertools",
     "bitflags",
+    "thiserror",
 ]
 parser = [
     "bytes",

examples/cache_reading.rs

@@ -15,7 +15,10 @@ fn main() {
         let parser =
             BgpkitParser::new_cached(item.url.as_str(), "/tmp/bgpkit-cache-example/").unwrap();
         // iterating through the parser. the iterator returns `BgpElem` one at a time.
-        let elems = parser.into_elem_iter().collect::<Vec<BgpElem>>();
+        let elems = parser
+            .into_elem_iter()
+            .filter_map(Result::ok)
+            .collect::<Vec<BgpElem>>();
         log::info!("{} {} {}", item.collector_id, item.url, elems.len());
     }
 }

examples/deprecated_attributes.rs

@@ -9,6 +9,7 @@ fn main() {
     )
     .unwrap()
     {
+        let elem = elem.unwrap();
         if elem.deprecated.is_some() {
             println!(
                 "{}",

examples/display_elems.rs

@@ -4,6 +4,7 @@ fn main() {
     let url = "http://archive.routeviews.org/bgpdata/\
         2021.10/UPDATES/updates.20211001.0000.bz2";
     for elem in BgpkitParser::new(url).unwrap() {
+        let elem = elem.unwrap();
         println!(
             "{:?}|{:?}|{:?}|{:?}|{:?}",
             elem.elem_type, elem.timestamp, elem.prefix, elem.as_path, elem.next_hop,

examples/extended_communities.rs

@@ -16,6 +16,7 @@ fn main() {
     log::info!("parsing updates file");
     // iterating through the parser. the iterator returns `BgpElem` one at a time.
     for elem in parser {
+        let elem = elem.unwrap();
         if let Some(cs) = &elem.communities {
             for c in cs {
                 match c {

examples/filters.rs

@@ -1,3 +1,4 @@
+use bgpkit_parser::filter::Filter;
 use bgpkit_parser::BgpkitParser;
 
 /// This example shows how to parse a MRT file and filter by prefix.
@@ -27,13 +28,12 @@ fn main() {
         "http://archive.routeviews.org/bgpdata/2021.10/UPDATES/updates.20211001.0000.bz2",
     )
     .unwrap()
-    .add_filter("prefix", "211.98.251.0/24")
-    .unwrap();
+    .add_filter(Filter::prefix("211.98.251.0/24").unwrap());
 
     log::info!("parsing updates file");
     // iterating through the parser. the iterator returns `BgpElem` one at a time.
     for elem in parser {
-        log::info!("{}", &elem);
+        log::info!("{}", elem.unwrap());
     }
     log::info!("done");
 }

examples/find_as_set_messages.rs

@@ -19,6 +19,7 @@ fn main() {
         let collector = item.collector_id.clone();
         let mut origins: HashSet<Asn> = HashSet::new();
         for elem in parser {
+            let elem = elem.unwrap();
             if !elem.elem_type.is_announce() {
                 continue;
             }

examples/only-to-customer.rs

@@ -6,6 +6,7 @@ fn main() {
     )
     .unwrap()
     {
+        let elem = elem.unwrap();
         if let Some(otc) = elem.only_to_customer {
             println!(
                 "OTC found: {} for path {}\n{}\n",

examples/parse-files-from-broker.rs

@@ -19,6 +19,7 @@ fn main() {
         // iterating through the parser. the iterator returns `BgpElem` one at a time.
         let elems = parser
             .into_elem_iter()
+            .map(Result::unwrap)
             .filter_map(|elem| {
                 if let Some(origins) = &elem.origin_asns {
                     if origins.contains(&13335.into()) {

examples/parse-single-file.rs

@@ -15,7 +15,7 @@ fn main() {
     log::info!("parsing updates file");
     // iterating through the parser. the iterator returns `BgpElem` one at a time.
     for elem in parser {
-        log::info!("{}", &elem);
+        log::info!("{}", elem.unwrap());
     }
     log::info!("done");
 }

examples/peer_index_table.rs

@@ -8,6 +8,6 @@ fn main() {
     let url = "https://data.ris.ripe.net/rrc03/2021.11/bview.20211128.1600.gz";
     let parser = bgpkit_parser::BgpkitParser::new(url).unwrap();
     for record in parser.into_record_iter().take(1) {
-        println!("{}", to_string_pretty(&json!(record)).unwrap());
+        println!("{}", to_string_pretty(&json!(record.unwrap())).unwrap());
     }
 }

examples/records_iter.rs

@@ -6,7 +6,7 @@ fn main() {
     let url = "http://archive.routeviews.org/route-views.amsix/bgpdata/2023.02/UPDATES/updates.20230222.0430.bz2";
     let parser = BgpkitParser::new(url).unwrap();
     for record in parser.into_record_iter() {
-        match record.message {
+        match record.unwrap().message {
             MrtMessage::TableDumpMessage(_) => {}
             MrtMessage::TableDumpV2Message(_) => {}
             MrtMessage::Bgp4Mp(msg) => match msg {

src/bin/main.rs

@@ -1,10 +1,12 @@
-use itertools::Itertools;
-use serde_json::json;
-use std::io::Write;
+use std::fmt::Display;
+use std::io;
+use std::io::{stdout, BufWriter, Write};
 use std::net::IpAddr;
 use std::path::PathBuf;
 
-use bgpkit_parser::{BgpkitParser, Elementor};
+use bgpkit_parser::filter::Filter;
+use bgpkit_parser::models::ElemType;
+use bgpkit_parser::{BgpkitParser, Elementor, PrefixMatchType};
 use clap::Parser;
 use ipnet::IpNet;
 
@@ -104,52 +106,53 @@ fn main() {
     };
 
     if let Some(v) = opts.filters.as_path {
-        parser = parser.add_filter("as_path", v.as_str()).unwrap();
+        parser = parser.add_filter(Filter::as_path(v.as_str()).unwrap());
     }
     if let Some(v) = opts.filters.origin_asn {
-        parser = parser
-            .add_filter("origin_asn", v.to_string().as_str())
-            .unwrap();
+        parser = parser.add_filter(Filter::OriginAsn(v));
     }
     if let Some(v) = opts.filters.prefix {
         let filter_type = match (opts.filters.include_super, opts.filters.include_sub) {
-            (false, false) => "prefix",
-            (true, false) => "prefix_super",
-            (false, true) => "prefix_sub",
-            (true, true) => "prefix_super_sub",
+            (false, false) => PrefixMatchType::Exact,
+            (true, false) => PrefixMatchType::IncludeSuper,
+            (false, true) => PrefixMatchType::IncludeSub,
+            (true, true) => PrefixMatchType::IncludeSuperSub,
         };
-        parser = parser
-            .add_filter(filter_type, v.to_string().as_str())
-            .unwrap();
+        parser = parser.add_filter(Filter::Prefix(v, filter_type));
     }
     if !opts.filters.peer_ip.is_empty() {
-        let v = opts.filters.peer_ip.iter().map(|p| p.to_string()).join(",");
-        parser = parser.add_filter("peer_ips", v.as_str()).unwrap();
+        parser = parser.add_filter(Filter::PeerIps(opts.filters.peer_ip.to_owned()));
     }
     if let Some(v) = opts.filters.peer_asn {
-        parser = parser
-            .add_filter("peer_asn", v.to_string().as_str())
-            .unwrap();
+        parser = parser.add_filter(Filter::PeerAsn(v));
     }
     if let Some(v) = opts.filters.elem_type {
-        parser = parser.add_filter("type", v.as_str()).unwrap();
+        let filter_type = match v.as_str() {
+            "w" | "withdraw" | "withdrawal" => ElemType::WITHDRAW,
+            "a" | "announce" | "announcement" => ElemType::ANNOUNCE,
+            x => panic!("cannot parse elem type from {}", x),
+        };
+        parser = parser.add_filter(Filter::Type(filter_type));
     }
     if let Some(v) = opts.filters.start_ts {
-        parser = parser
-            .add_filter("start_ts", v.to_string().as_str())
-            .unwrap();
+        parser = parser.add_filter(Filter::TsStart(v));
     }
     if let Some(v) = opts.filters.end_ts {
-        parser = parser.add_filter("end_ts", v.to_string().as_str()).unwrap();
+        parser = parser.add_filter(Filter::TsEnd(v));
     }
 
     match (opts.elems_count, opts.records_count) {
         (true, true) => {
             let mut elementor = Elementor::new();
             let (mut records_count, mut elems_count) = (0, 0);
             for record in parser.into_record_iter() {
-                records_count += 1;
-                elems_count += elementor.record_to_elems(record).len();
+                match record {
+                    Ok(record) => {
+                        records_count += 1;
+                        elems_count += elementor.record_to_elems(record).len();
+                    }
+                    Err(err) => handle_non_fatal_error(&mut stdout(), err),
+                }
             }
             println!("total records: {}", records_count);
             println!("total elems:   {}", elems_count);
@@ -158,28 +161,71 @@ fn main() {
             println!("total records: {}", parser.into_record_iter().count());
         }
         (true, false) => {
-            println!("total records: {}", parser.into_elem_iter().count());
+            println!("total elems: {}", parser.into_elem_iter().count());
         }
         (false, false) => {
-            let mut stdout = std::io::stdout();
+            let mut stdout = BufWriter::new(stdout().lock());
+
             for elem in parser {
-                let output_str = if opts.json {
-                    let val = json!(elem);
-                    if opts.pretty {
-                        serde_json::to_string_pretty(&val).unwrap()
-                    } else {
-                        val.to_string()
+                match elem {
+                    Ok(elem) => {
+                        if opts.json {
+                            let res = if opts.pretty {
+                                serde_json::to_writer_pretty(&mut stdout, &elem)
+                            } else {
+                                serde_json::to_writer(&mut stdout, &elem)
+                            };
+
+                            handle_serde_json_result(&mut stdout, res);
+                        } else {
+                            let res = writeln!(stdout, "{}", elem);
+                            handle_io_result(&mut stdout, res);
+                        }
                     }
-                } else {
-                    elem.to_string()
-                };
-                if let Err(e) = writeln!(stdout, "{}", &output_str) {
-                    if e.kind() != std::io::ErrorKind::BrokenPipe {
-                        eprintln!("{}", e);
+                    Err(err) => {
+                        let res = stdout.flush();
+                        handle_io_result(&mut stdout, res);
+                        eprintln!("{}", err);
                     }
-                    std::process::exit(1);
                 }
             }
         }
     }
 }
+
+fn handle_serde_json_result<W: Write>(stdout: &mut W, res: serde_json::Result<()>) {
+    if let Err(err) = res {
+        if err.is_io() {
+            // If it was an IO error, we likely wont be able to flush stdout
+            eprintln!("{}", err);
+            std::process::exit(1);
+        }
+
+        handle_non_fatal_error(stdout, err);
+    }
+}
+
+fn handle_non_fatal_error<W: Write, E: Display>(stdout: &mut W, err: E) {
+    // Attempt to flush stdout before printing the error to avoid mangling combined CLI output
+    if let Err(flush_err) = stdout.flush() {
+        eprintln!("{}", err);
+        eprintln!("{}", flush_err);
+        std::process::exit(1);
+    }
+
+    // Write the error to stderr then flush stderr to avoid mangling combined CLI output
+    eprintln!("{}", err);
+    if io::stderr().flush().is_err() {
+        // If this fails, then we are out of options for logging errors
+        std::process::exit(1);
+    }
+}
+
+fn handle_io_result<W: Write>(stdout: &mut W, res: io::Result<()>) {
+    if let Err(err) = res {
+        // We can try flushing stdout, but it will almost certainly fail
+        let _ = stdout.flush();
+        eprintln!("{}", err);
+        std::process::exit(1);
+    }
+}