From c07daf8068d123f90f3dc5ceff22fd4f4e279610 Mon Sep 17 00:00:00 2001 From: Mingwei Zhang Date: Thu, 2 Jul 2026 09:38:23 -0700 Subject: [PATCH 1/4] feat: add RIS Live raw message parsing --- CHANGELOG.md | 3 + README.md | 15 ++-- examples/README.md | 4 +- examples/real_time_ris_live_websocket.rs | 13 ++-- .../real_time_ris_live_websocket_async.rs | 8 +- src/lib.rs | 15 ++-- src/parser/mod.rs | 6 +- src/parser/rislive/messages/client/mod.rs | 2 +- .../rislive/messages/client/ris_subscribe.rs | 13 +++- src/parser/rislive/messages/server/mod.rs | 2 +- .../rislive/messages/server/raw_bytes.rs | 78 +++++++++---------- src/parser/rislive/mod.rs | 27 +++++-- 12 files changed, 113 insertions(+), 73 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bf9926e..bb635410 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,9 @@ All notable changes to this project will be documented in this file. ### New features +* **RIS Live raw message parsing**: Added `parse_ris_live_message_raw()` and `parse_ris_live_message_json()` so users can opt into RIS Live `socketOptions.includeRaw` and parse original BGP wire messages instead of only RIS Live's reduced JSON attribute projection. +* **RIS Live subscription helpers**: Re-exported `RisSubscribe`, `RisSubscribeType`, and `RisLiveClientMessage` from the crate root, making it easier to request `includeRaw` in examples and downstream code. +* **RIS Live examples**: Updated the synchronous WebSocket example to demonstrate `includeRaw` + raw BGP parsing; kept the async example on JSON-field parsing with comments showing how to opt into raw parsing. * **Raw attribute retention**: Known BGP path attributes without semantic parsing are no longer silently dropped. They are retained as `AttributeValue::Raw(AttrRaw)` with their original wire code and bytes, enabling faithful re-encoding. * **Deprecated code point handling**: Removed `AttrType::CLUSTER_ID = 13` (deprecated per IANA); code 13 is now retained as `Deprecated(AttrRaw)`. Added `is_deprecated_attr_type()` helper. * **`AttrType::BIER` added**: Code 41 (RFC 9793) added to the enum. diff --git a/README.md b/README.md index 5c075438..b2f1ac2a 100644 --- a/README.md +++ b/README.md @@ -245,11 +245,14 @@ and [BMP][bmp-rfc]/[OpenBMP][openbmp-url] messages. Here is an example of handling RIS-Live message streams. After connecting to the websocket server, we need to subscribe to a specific data stream. In this example, we subscribe to the data stream -from on collector (`rrc21`). We can then loop and read messages from the websocket. +from one collector (`rrc21`). We can then loop and read messages from the websocket. + +RIS Live's JSON fields expose only a subset of BGP attributes. To parse the original BGP wire +message instead, request `includeRaw` and use `parse_ris_live_message_raw`. The older +`parse_ris_live_message` function remains available for parsing RIS Live's JSON-projected fields. ```rust -use bgpkit_parser::parse_ris_live_message; -use serde_json::json; +use bgpkit_parser::{parse_ris_live_message_raw, RisLiveClientMessage, RisSubscribe}; use tungstenite::{connect, Message}; const RIS_LIVE_URL: &str = "ws://ris-live.ripe.net/v1/ws/?client=rust-bgpkit-parser"; @@ -263,13 +266,13 @@ fn main() { connect(RIS_LIVE_URL) .expect("Can't connect to RIS Live websocket server"); - // subscribe to messages from one collector - let msg = json!({"type": "ris_subscribe", "data": {"host": "rrc21"}}).to_string(); + // subscribe to messages from one collector and request hex-encoded raw BGP messages + let msg = RisSubscribe::new().host("rrc21").include_raw(true).to_json_string(); socket.send(Message::Text(msg.into())).unwrap(); loop { let msg = socket.read().expect("Error reading message").to_string(); - if let Ok(elems) = parse_ris_live_message(msg.as_str()) { + if let Ok(elems) = parse_ris_live_message_raw(msg.as_str()) { for elem in elems { println!("{}", elem); } diff --git a/examples/README.md b/examples/README.md index 97fd369b..6f0bc148 100644 --- a/examples/README.md +++ b/examples/README.md @@ -24,8 +24,8 @@ This directory contains runnable examples for bgpkit_parser. They demonstrate ba - [mrt_filter_archiver.rs](mrt_filter_archiver.rs) — Apply filters while reading updates and archive the filtered stream into a new MRT file; re_parse to verify the output. ## Real_time Streams (RIS Live, RouteViews Kafka, BMP) -- [real_time_ris_live_websocket.rs](real_time_ris_live_websocket.rs) — Connect to RIPE RIS Live over WebSocket (tungstenite), subscribe to a collector, parse, and print elements. -- [real_time_ris_live_websocket_async.rs](real_time_ris_live_websocket_async.rs) — Async tokio+tungstenite version of the RIS Live WebSocket subscriber. +- [real_time_ris_live_websocket.rs](real_time_ris_live_websocket.rs) — Connect to RIPE RIS Live over WebSocket (tungstenite), request `includeRaw`, parse raw BGP wire messages, and print elements. +- [real_time_ris_live_websocket_async.rs](real_time_ris_live_websocket_async.rs) — Async tokio+tungstenite version of the RIS Live WebSocket subscriber using RIS Live's JSON-projected fields, with comments showing how to switch to raw parsing. - [real_time_routeviews_kafka_openbmp.rs](real_time_routeviews_kafka_openbmp.rs) — Consume RouteViews Kafka topics (OpenBMP format), parse BMP messages, and print derived elements. - [real_time_routeviews_kafka_to_mrt.rs](real_time_routeviews_kafka_to_mrt.rs) — Consume RouteViews Kafka BMP messages and archive them into an MRT file for later analysis. - [bmp_listener.rs](bmp_listener.rs) — Minimal TCP BMP listener that parses incoming BMP Route Monitoring messages and logs them. diff --git a/examples/real_time_ris_live_websocket.rs b/examples/real_time_ris_live_websocket.rs index dfa9a6bc..27c0f9bf 100644 --- a/examples/real_time_ris_live_websocket.rs +++ b/examples/real_time_ris_live_websocket.rs @@ -1,19 +1,22 @@ -use bgpkit_parser::parse_ris_live_message; -use bgpkit_parser::rislive::messages::{RisLiveClientMessage, RisSubscribe}; +use bgpkit_parser::{parse_ris_live_message_raw, RisLiveClientMessage, RisSubscribe}; use tungstenite::{connect, Message}; const RIS_LIVE_URL: &str = "ws://ris-live.ripe.net/v1/ws/?client=bgpkit-parser-example"; /// This is an example of subscribing to RIS-Live's streaming data from one host (`rrc21`). /// +/// This example opts into RIS Live's `socketOptions.includeRaw` mode and parses the original +/// BGP wire message. Use `parse_ris_live_message` instead if you want to parse RIS Live's +/// JSON-projected fields without requesting raw bytes. +/// /// For more RIS-Live details, check out their documentation at https://ris-live.ripe.net/manual/ fn main() { // connect to RIPE RIS Live websocket server let (mut socket, _response) = connect(RIS_LIVE_URL).expect("Can't connect to RIS Live websocket server"); - // subscribe to messages from one collector - let msg = RisSubscribe::new().host("rrc21"); + // subscribe to messages from one collector and request hex-encoded raw BGP messages + let msg = RisSubscribe::new().host("rrc21").include_raw(true); socket .send(Message::Text(msg.to_json_string().into())) .unwrap(); @@ -23,7 +26,7 @@ fn main() { if msg.is_empty() { continue; } - match parse_ris_live_message(msg.as_str()) { + match parse_ris_live_message_raw(msg.as_str()) { Ok(elems) => { for elem in elems { println!("{elem}"); diff --git a/examples/real_time_ris_live_websocket_async.rs b/examples/real_time_ris_live_websocket_async.rs index ae1437e1..3a5d56c7 100644 --- a/examples/real_time_ris_live_websocket_async.rs +++ b/examples/real_time_ris_live_websocket_async.rs @@ -1,5 +1,4 @@ -use bgpkit_parser::parse_ris_live_message; -use bgpkit_parser::rislive::messages::{RisLiveClientMessage, RisSubscribe}; +use bgpkit_parser::{parse_ris_live_message, RisLiveClientMessage, RisSubscribe}; use futures_util::{SinkExt, StreamExt}; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message; @@ -11,7 +10,10 @@ async fn main() { // connect to websocket server let (ws_stream, _response) = connect_async(RIS_LIVE_URL).await.unwrap(); - // send a subscription message + // Send a subscription message. + // + // This async example keeps the default JSON-field parser. To parse the original BGP wire + // message instead, add `.include_raw(true)` here and call `parse_ris_live_message_raw` below. let msg = RisSubscribe::new().host("rrc21"); let (mut write, mut read) = ws_stream.split(); write diff --git a/src/lib.rs b/src/lib.rs index e7dcb016..4d06ca10 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -235,12 +235,15 @@ and [BMP][bmp-rfc]/[OpenBMP][openbmp-url] messages. Here is an example of handling RIS-Live message streams. After connecting to the websocket server, we need to subscribe to a specific data stream. In this example, we subscribe to the data stream -from on collector (`rrc21`). We can then loop and read messages from the websocket. +from one collector (`rrc21`). We can then loop and read messages from the websocket. + +RIS Live's JSON fields expose only a subset of BGP attributes. To parse the original BGP wire +message instead, request `includeRaw` and use `parse_ris_live_message_raw`. The older +`parse_ris_live_message` function remains available for parsing RIS Live's JSON-projected fields. ```no_run # #[cfg(feature = "rislive")] -use bgpkit_parser::parse_ris_live_message; -use serde_json::json; +use bgpkit_parser::{parse_ris_live_message_raw, RisLiveClientMessage, RisSubscribe}; use tungstenite::{connect, Message}; const RIS_LIVE_URL: &str = "ws://ris-live.ripe.net/v1/ws/?client=rust-bgpkit-parser"; @@ -254,14 +257,14 @@ fn main() { connect(RIS_LIVE_URL) .expect("Can't connect to RIS Live websocket server"); - // subscribe to messages from one collector - let msg = json!({"type": "ris_subscribe", "data": {"host": "rrc21"}}).to_string(); + // subscribe to messages from one collector and request hex-encoded raw BGP messages + let msg = RisSubscribe::new().host("rrc21").include_raw(true).to_json_string(); socket.send(Message::Text(msg.into())).unwrap(); loop { let msg = socket.read().expect("Error reading message").to_string(); # #[cfg(feature = "rislive")] - if let Ok(elems) = parse_ris_live_message(msg.as_str()) { + if let Ok(elems) = parse_ris_live_message_raw(msg.as_str()) { for elem in elems { println!("{}", elem); } diff --git a/src/parser/mod.rs b/src/parser/mod.rs index d407e8d8..cc69df3e 100644 --- a/src/parser/mod.rs +++ b/src/parser/mod.rs @@ -29,7 +29,11 @@ pub use iters::*; pub use mrt::*; #[cfg(feature = "rislive")] -pub use rislive::parse_ris_live_message; +pub use rislive::messages::{RisLiveClientMessage, RisSubscribe, RisSubscribeType}; +#[cfg(feature = "rislive")] +pub use rislive::{ + parse_ris_live_message, parse_ris_live_message_json, parse_ris_live_message_raw, +}; pub struct BgpkitParser { reader: R, diff --git a/src/parser/rislive/messages/client/mod.rs b/src/parser/rislive/messages/client/mod.rs index 89c90ad2..ee38f5a3 100644 --- a/src/parser/rislive/messages/client/mod.rs +++ b/src/parser/rislive/messages/client/mod.rs @@ -24,5 +24,5 @@ use serde_json::json; pub use ping::Ping; pub use request_rrc_list::RequestRrcList; -pub use ris_subscribe::RisSubscribe; +pub use ris_subscribe::{RisSubscribe, RisSubscribeSocketOptions, RisSubscribeType}; pub use ris_unsubscribe::RisUnsubscribe; diff --git a/src/parser/rislive/messages/client/ris_subscribe.rs b/src/parser/rislive/messages/client/ris_subscribe.rs index 6ce2e3eb..fe47d55a 100644 --- a/src/parser/rislive/messages/client/ris_subscribe.rs +++ b/src/parser/rislive/messages/client/ris_subscribe.rs @@ -15,15 +15,17 @@ pub enum RisSubscribeType { #[derive(Debug, Serialize)] pub struct RisSubscribeSocketOptions { - /// Include a Base64-encoded version of the original binary BGP message as `raw` for all subscriptions + /// Include a hex-encoded version of the original binary BGP message as `raw` for all subscriptions /// /// *Default: false* #[serde(rename = "includeRaw")] + #[serde(skip_serializing_if = "Option::is_none")] pub include_raw: Option, /// Send a `ris_subscribe_ok` message for all succesful subscriptions /// /// *Default: false* + #[serde(skip_serializing_if = "Option::is_none")] pub acknowledge: Option, } @@ -203,4 +205,13 @@ mod tests { println!("{}", ris_subscribe.to_json_string()); } + + #[test] + fn test_include_raw_serializes_socket_option() { + let ris_subscribe = RisSubscribe::new().host("rrc00").include_raw(true); + assert_eq!( + ris_subscribe.to_json_string(), + r#"{"data":{"host":"rrc00","socketOptions":{"includeRaw":true}},"type":"ris_subscribe"}"# + ); + } } diff --git a/src/parser/rislive/messages/server/mod.rs b/src/parser/rislive/messages/server/mod.rs index 4e4def33..4a72ccfd 100644 --- a/src/parser/rislive/messages/server/mod.rs +++ b/src/parser/rislive/messages/server/mod.rs @@ -9,7 +9,7 @@ pub mod ris_rrc_list; pub mod ris_subscribe_ok; pub use pong::Pong; -pub use raw_bytes::parse_raw_bytes; +pub use raw_bytes::{parse_raw_bytes, parse_ris_live_message_raw}; pub use ris_error::RisError; pub use ris_message::RisMessage; pub use ris_message::RisMessageEnum; diff --git a/src/parser/rislive/messages/server/raw_bytes.rs b/src/parser/rislive/messages/server/raw_bytes.rs index 88508234..d78ff2d8 100644 --- a/src/parser/rislive/messages/server/raw_bytes.rs +++ b/src/parser/rislive/messages/server/raw_bytes.rs @@ -1,58 +1,61 @@ use crate::models::*; use crate::parser::bgp::parse_bgp_message; use crate::parser::rislive::error::ParserRisliveError; +use crate::parser::rislive::messages::RisLiveMessage; use crate::Elementor; use bytes::Bytes; -use serde_json::Value; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; -// FIXME: This function will panic on any unexpected values +/// Parse a RIS Live JSON message using the `data.raw` BGP message bytes. +/// +/// RIS Live includes `raw` only when subscribing with +/// `socketOptions.includeRaw=true`. The field is hex-encoded BGP wire data. +/// Parsing from raw bytes exposes attributes that RIS Live's JSON projection +/// omits, while still returning the same [`BgpElem`] interface as the JSON +/// parser. pub fn parse_raw_bytes(msg_str: &str) -> Result, ParserRisliveError> { - let msg: Value = serde_json::from_str(msg_str)?; - let msg_type = match msg.get("type") { - None => return Err(ParserRisliveError::IrregularRisLiveFormat), - Some(t) => t.as_str().unwrap(), + parse_ris_live_message_raw(msg_str) +} + +/// Parse a RIS Live JSON message using the hex-encoded `data.raw` BGP message. +pub fn parse_ris_live_message_raw(msg_str: &str) -> Result, ParserRisliveError> { + let msg: RisLiveMessage = serde_json::from_str(msg_str) + .map_err(|_| ParserRisliveError::IncorrectJson(msg_str.to_string()))?; + + let ris_msg = match msg { + RisLiveMessage::RisMessage(ris_msg) => ris_msg, + RisLiveMessage::RisError(_) + | RisLiveMessage::RisRrcList(_) + | RisLiveMessage::RisSubscribeOk(_) + | RisLiveMessage::Pong(_) => return Err(ParserRisliveError::UnsupportedMessage), }; - match msg_type { - "ris_message" => {} - "ris_error" | "ris_rrc_list" | "ris_subscribe_ok" | "pong" => { - return Err(ParserRisliveError::UnsupportedMessage) - } - _ => return Err(ParserRisliveError::IrregularRisLiveFormat), - } + let raw = ris_msg.raw.ok_or(ParserRisliveError::IncorrectRawBytes)?; + let raw_bytes = hex::decode(raw).map_err(|_| ParserRisliveError::IncorrectRawBytes)?; - let data = msg.get("data").unwrap().as_object().unwrap(); + parse_ris_live_raw_bgp_message(raw_bytes, ris_msg.timestamp, ris_msg.peer, ris_msg.peer_asn) +} - let mut bytes = Bytes::from(hex::decode(data.get("raw").unwrap().as_str().unwrap()).unwrap()); +fn parse_ris_live_raw_bgp_message( + raw_bytes: Vec, + timestamp: f64, + peer_ip: IpAddr, + peer_asn: Asn, +) -> Result, ParserRisliveError> { + let bytes = Bytes::from(raw_bytes); - let timestamp = data.get("timestamp").unwrap().as_f64().unwrap(); - let peer_str = data.get("peer").unwrap().as_str().unwrap().to_owned(); + let bgp_msg = parse_bgp_message(&mut bytes.clone(), false, &AsnLength::Bits32) + .or_else(|_| parse_bgp_message(&mut bytes.clone(), false, &AsnLength::Bits16)) + .map_err(|_| ParserRisliveError::IncorrectRawBytes)?; - let peer_ip = peer_str.parse::().unwrap(); let local_ip = match peer_ip.is_ipv4() { true => IpAddr::V4(Ipv4Addr::UNSPECIFIED), false => IpAddr::V6(Ipv6Addr::UNSPECIFIED), }; - let peer_asn_str = data.get("peer_asn").unwrap().as_str().unwrap().to_owned(); - - let peer_asn = peer_asn_str.parse::().unwrap(); - - let bgp_msg = match parse_bgp_message(&mut bytes, false, &AsnLength::Bits32) { - Ok(m) => m, - Err(_) => match parse_bgp_message(&mut bytes, false, &AsnLength::Bits16) { - Ok(m) => m, - Err(_) => return Err(ParserRisliveError::IncorrectRawBytes), - }, - }; - - let t_sec = timestamp as u32; - let t_msec = get_micro_seconds(timestamp); - let header = CommonHeader { - timestamp: t_sec, - microsecond_timestamp: Some(t_msec), + timestamp: timestamp as u32, + microsecond_timestamp: Some(get_micro_seconds(timestamp)), entry_type: EntryType::BGP4MP, entry_subtype: Bgp4MpType::MessageAs4 as u16, length: 0, @@ -74,10 +77,7 @@ pub fn parse_raw_bytes(msg_str: &str) -> Result, ParserRisliveError } fn get_micro_seconds(sec: f64) -> u32 { - format!("{sec:.6}").split('.').collect::>()[1] - .to_owned() - .parse::() - .unwrap() + ((sec.fract().abs() * 1_000_000.0).round() as u32).min(999_999) } #[cfg(test)] diff --git a/src/parser/rislive/mod.rs b/src/parser/rislive/mod.rs index e0a8a033..3f6f67a1 100644 --- a/src/parser/rislive/mod.rs +++ b/src/parser/rislive/mod.rs @@ -2,13 +2,14 @@ Provides parsing functions for [RIS-Live](https://ris-live.ripe.net/manual/) real-time BGP message stream JSON data. -The main parsing function, [parse_ris_live_message] converts a JSON-formatted message string into a -vector of [BgpElem]s. +The main parsing function, [parse_ris_live_message] converts RIS Live's JSON-projected UPDATE +fields into a vector of [BgpElem]s. If you subscribe with `includeRaw`, use +[parse_ris_live_message_raw] to parse the original BGP wire message instead; this preserves BGP +attributes that RIS Live omits from its JSON fields. Here is an example parsing stream data from one collector: ```no_run -use bgpkit_parser::parse_ris_live_message; -use serde_json::json; +use bgpkit_parser::{parse_ris_live_message_raw, RisLiveClientMessage, RisSubscribe}; use tungstenite::{connect, Message}; const RIS_LIVE_URL: &str = "ws://ris-live.ripe.net/v1/ws/?client=rust-bgpkit-parser"; @@ -22,13 +23,13 @@ fn main() { connect(RIS_LIVE_URL) .expect("Can't connect to RIS Live websocket server"); - // subscribe to messages from one collector - let msg = json!({"type": "ris_subscribe", "data": {"host": "rrc21"}}).to_string(); + // subscribe to messages from one collector and request hex-encoded raw BGP messages + let msg = RisSubscribe::new().host("rrc21").include_raw(true).to_json_string(); socket.send(Message::Text(msg.into())).unwrap(); loop { let msg = socket.read().expect("Error reading message").to_string(); - if let Ok(elems) = parse_ris_live_message(msg.as_str()) { + if let Ok(elems) = parse_ris_live_message_raw(msg.as_str()) { for elem in elems { println!("{}", elem); } @@ -38,6 +39,7 @@ fn main() { ``` */ use crate::parser::rislive::error::ParserRisliveError; +pub use crate::parser::rislive::messages::parse_ris_live_message_raw; use crate::parser::rislive::messages::{RisLiveMessage, RisMessageEnum}; use crate::models::*; @@ -73,7 +75,11 @@ fn parse_prefix(prefix_str: &str) -> Result { Ok(p) } -/// This function parses one message and returns a result of a vector of [BgpElem]s or an error +/// Parse one RIS Live message using RIS Live's JSON-projected UPDATE fields. +/// +/// This parser is convenient and does not require `socketOptions.includeRaw`, but RIS Live's JSON +/// schema exposes only a subset of BGP path attributes. Use [`parse_ris_live_message_raw`] when you +/// need attributes that are only present in the raw BGP message. pub fn parse_ris_live_message(msg_str: &str) -> Result, ParserRisliveError> { let msg_string = msg_str.to_string(); @@ -229,6 +235,11 @@ pub fn parse_ris_live_message(msg_str: &str) -> Result, ParserRisli } } +/// Alias for [`parse_ris_live_message`] to make the JSON-vs-raw choice explicit. +pub fn parse_ris_live_message_json(msg_str: &str) -> Result, ParserRisliveError> { + parse_ris_live_message(msg_str) +} + #[cfg(test)] mod tests { use super::*; From ae3cda5b8468b2da7e9f26afdc49b5541c1ada12 Mon Sep 17 00:00:00 2001 From: Mingwei Zhang Date: Thu, 2 Jul 2026 10:06:50 -0700 Subject: [PATCH 2/4] test: cover RIS Live raw parsing errors --- CHANGELOG.md | 2 +- .../rislive/messages/server/raw_bytes.rs | 61 +++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bb635410..fee7facf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ All notable changes to this project will be documented in this file. ### New features -* **RIS Live raw message parsing**: Added `parse_ris_live_message_raw()` and `parse_ris_live_message_json()` so users can opt into RIS Live `socketOptions.includeRaw` and parse original BGP wire messages instead of only RIS Live's reduced JSON attribute projection. +* **RIS Live raw message parsing**: Added `parse_ris_live_message_raw()` and `parse_ris_live_message_json()` so users can opt into RIS Live `socketOptions.includeRaw` and parse original BGP wire messages instead of only RIS Live's reduced JSON attribute projection. Raw parsing now validates the full RIS Live `ris_message` envelope and expects standard `id`/`host` fields. * **RIS Live subscription helpers**: Re-exported `RisSubscribe`, `RisSubscribeType`, and `RisLiveClientMessage` from the crate root, making it easier to request `includeRaw` in examples and downstream code. * **RIS Live examples**: Updated the synchronous WebSocket example to demonstrate `includeRaw` + raw BGP parsing; kept the async example on JSON-field parsing with comments showing how to opt into raw parsing. * **Raw attribute retention**: Known BGP path attributes without semantic parsing are no longer silently dropped. They are retained as `AttributeValue::Raw(AttrRaw)` with their original wire code and bytes, enabling faithful re-encoding. diff --git a/src/parser/rislive/messages/server/raw_bytes.rs b/src/parser/rislive/messages/server/raw_bytes.rs index d78ff2d8..7f07101f 100644 --- a/src/parser/rislive/messages/server/raw_bytes.rs +++ b/src/parser/rislive/messages/server/raw_bytes.rs @@ -143,4 +143,65 @@ mod tests { println!("{elem}"); } } + + fn ris_message_with_raw(raw: Option<&str>) -> String { + let raw_field = raw + .map(|raw| format!(r#", "raw": "{raw}""#)) + .unwrap_or_default(); + format!( + r#"{{ + "type": "ris_message", + "data": {{ + "timestamp": 1636245154.8, + "peer": "192.0.2.1", + "peer_asn": "64496", + "id": "00-192-0-2-1-1", + "host": "rrc00", + "type": "UPDATE"{raw_field} + }} + }}"# + ) + } + + #[test] + fn ris_live_message_missing_raw_returns_incorrect_raw_bytes() { + let err = parse_raw_bytes(&ris_message_with_raw(None)).unwrap_err(); + assert!(matches!(err, ParserRisliveError::IncorrectRawBytes)); + } + + #[test] + fn ris_live_message_malformed_hex_returns_incorrect_raw_bytes() { + let err = parse_raw_bytes(&ris_message_with_raw(Some("not-hex"))).unwrap_err(); + assert!(matches!(err, ParserRisliveError::IncorrectRawBytes)); + } + + #[test] + fn ris_live_message_invalid_bgp_raw_returns_incorrect_raw_bytes() { + let err = parse_raw_bytes(&ris_message_with_raw(Some("00"))).unwrap_err(); + assert!(matches!(err, ParserRisliveError::IncorrectRawBytes)); + } + + #[test] + fn non_ris_message_returns_unsupported_message() { + let err = parse_raw_bytes(r#"{"type":"pong","data":null}"#).unwrap_err(); + assert!(matches!(err, ParserRisliveError::UnsupportedMessage)); + } + + #[test] + fn ris_live_message_missing_required_id_or_host_returns_incorrect_json() { + let err = parse_raw_bytes( + r#"{ + "type": "ris_message", + "data": { + "timestamp": 1636245154.8, + "peer": "192.0.2.1", + "peer_asn": "64496", + "type": "UPDATE", + "raw": "00" + } + }"#, + ) + .unwrap_err(); + assert!(matches!(err, ParserRisliveError::IncorrectJson(_))); + } } From a22889b1f1564193879e3c984b54d12d8c22139c Mon Sep 17 00:00:00 2001 From: Mingwei Zhang Date: Thu, 2 Jul 2026 10:10:40 -0700 Subject: [PATCH 3/4] docs: clarify RIS Live raw parser contract --- src/parser/rislive/messages/server/raw_bytes.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/src/parser/rislive/messages/server/raw_bytes.rs b/src/parser/rislive/messages/server/raw_bytes.rs index 7f07101f..ccd149cc 100644 --- a/src/parser/rislive/messages/server/raw_bytes.rs +++ b/src/parser/rislive/messages/server/raw_bytes.rs @@ -6,21 +6,25 @@ use crate::Elementor; use bytes::Bytes; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; -/// Parse a RIS Live JSON message using the `data.raw` BGP message bytes. +/// Parse a RIS Live JSON `ris_message` envelope using the `data.raw` BGP message bytes. /// /// RIS Live includes `raw` only when subscribing with /// `socketOptions.includeRaw=true`. The field is hex-encoded BGP wire data. /// Parsing from raw bytes exposes attributes that RIS Live's JSON projection /// omits, while still returning the same [`BgpElem`] interface as the JSON /// parser. +/// +/// This function expects the full RIS Live `ris_message` envelope, including +/// the standard required fields such as `id`, `host`, `timestamp`, `peer`, and +/// `peer_asn`, plus a present `raw` field. pub fn parse_raw_bytes(msg_str: &str) -> Result, ParserRisliveError> { parse_ris_live_message_raw(msg_str) } -/// Parse a RIS Live JSON message using the hex-encoded `data.raw` BGP message. +/// Parse a RIS Live JSON `ris_message` envelope using the hex-encoded `data.raw` BGP message. pub fn parse_ris_live_message_raw(msg_str: &str) -> Result, ParserRisliveError> { let msg: RisLiveMessage = serde_json::from_str(msg_str) - .map_err(|_| ParserRisliveError::IncorrectJson(msg_str.to_string()))?; + .map_err(|e| ParserRisliveError::IncorrectJson(e.to_string()))?; let ris_msg = match msg { RisLiveMessage::RisMessage(ris_msg) => ris_msg, @@ -44,8 +48,10 @@ fn parse_ris_live_raw_bgp_message( ) -> Result, ParserRisliveError> { let bytes = Bytes::from(raw_bytes); - let bgp_msg = parse_bgp_message(&mut bytes.clone(), false, &AsnLength::Bits32) - .or_else(|_| parse_bgp_message(&mut bytes.clone(), false, &AsnLength::Bits16)) + let mut bytes_32bit_asn = bytes.clone(); + let mut bytes_16bit_asn = bytes.clone(); + let bgp_msg = parse_bgp_message(&mut bytes_32bit_asn, false, &AsnLength::Bits32) + .or_else(|_| parse_bgp_message(&mut bytes_16bit_asn, false, &AsnLength::Bits16)) .map_err(|_| ParserRisliveError::IncorrectRawBytes)?; let local_ip = match peer_ip.is_ipv4() { From b5f40b72c116ed0ad70ce78bd2f532b60996fdb7 Mon Sep 17 00:00:00 2001 From: Mingwei Zhang Date: Thu, 2 Jul 2026 10:12:43 -0700 Subject: [PATCH 4/4] feat: re-export RIS Live socket options --- src/parser/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/parser/mod.rs b/src/parser/mod.rs index cc69df3e..0081e56b 100644 --- a/src/parser/mod.rs +++ b/src/parser/mod.rs @@ -29,7 +29,9 @@ pub use iters::*; pub use mrt::*; #[cfg(feature = "rislive")] -pub use rislive::messages::{RisLiveClientMessage, RisSubscribe, RisSubscribeType}; +pub use rislive::messages::{ + RisLiveClientMessage, RisSubscribe, RisSubscribeSocketOptions, RisSubscribeType, +}; #[cfg(feature = "rislive")] pub use rislive::{ parse_ris_live_message, parse_ris_live_message_json, parse_ris_live_message_raw,