feat: add RIS Live raw message parsing

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. 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.
 * **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.

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);
             }

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.

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}");

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

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);
             }

src/parser/mod.rs

@@ -29,7 +29,13 @@ pub use iters::*;
 pub use mrt::*;
 
 #[cfg(feature = "rislive")]
-pub use rislive::parse_ris_live_message;
+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,
+};
 
 pub struct BgpkitParser<R> {
     reader: R,

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;

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<bool>,
 
     /// Send a `ris_subscribe_ok` message for all succesful subscriptions
     ///
     /// *Default: false*
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub acknowledge: Option<bool>,
 }
 
@@ -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"}"#
+        );
+    }
 }

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;

src/parser/rislive/messages/server/raw_bytes.rs

@@ -1,58 +1,67 @@
 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 `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<Vec<BgpElem>, 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 `ris_message` envelope using the hex-encoded `data.raw` BGP message.
+pub fn parse_ris_live_message_raw(msg_str: &str) -> Result<Vec<BgpElem>, ParserRisliveError> {
+    let msg: RisLiveMessage = serde_json::from_str(msg_str)
+        .map_err(|e| ParserRisliveError::IncorrectJson(e.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<u8>,
+    timestamp: f64,
+    peer_ip: IpAddr,
+    peer_asn: Asn,
+) -> Result<Vec<BgpElem>, 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 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 peer_ip = peer_str.parse::<IpAddr>().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::<Asn>().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 +83,7 @@ pub fn parse_raw_bytes(msg_str: &str) -> Result<Vec<BgpElem>, ParserRisliveError
 }
 
 fn get_micro_seconds(sec: f64) -> u32 {
-    format!("{sec:.6}").split('.').collect::<Vec<&str>>()[1]
-        .to_owned()
-        .parse::<u32>()
-        .unwrap()
+    ((sec.fract().abs() * 1_000_000.0).round() as u32).min(999_999)
 }
 
 #[cfg(test)]
@@ -143,4 +149,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(_)));
+    }
 }