refactor: move binary msg parsing to protocol crate, improve examples, source token from env for easier testing

Author: tejasbadadare

lazer/Cargo.lock

@@ -12,7 +12,7 @@ tokio-tungstenite = { version = "0.20", features = ["native-tls"] }
 futures-util = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-base64 = "0.21"
+base64 = "0.22.1"
 anyhow = "1.0"
 tracing = "0.1"
 url = "2.4"

lazer/sdk/rust/client/Cargo.toml

@@ -6,49 +6,87 @@ use pyth_lazer_protocol::router::{
 };
 use pyth_lazer_protocol::subscription::{Request, SubscribeRequest, SubscriptionId};
 
+fn get_lazer_access_token() -> String {
+    // Place your access token in your env at LAZER_ACCESS_TOKEN or set it here
+    let token = "your token here";
+    std::env::var("LAZER_ACCESS_TOKEN").unwrap_or_else(|_| token.to_string())
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     // Create and start the client
     let mut client = LazerClient::new(
         "wss://pyth-lazer.dourolabs.app/v1/stream",
-        "YOUR_ACCESS_TOKEN",
+        &get_lazer_access_token(),
     )?;
     let mut stream = client.start().await?;
-    // Create subscription request
-    let subscription_id = SubscriptionId(1);
-    let subscription_request = SubscribeRequest {
-        subscription_id,
-        params: SubscriptionParams::new(SubscriptionParamsRepr {
-            price_feed_ids: vec![PriceFeedId(1), PriceFeedId(2), PriceFeedId(3)],
-            properties: vec![PriceFeedProperty::Price],
-            chains: vec![Chain::Solana],
-            delivery_format: DeliveryFormat::Binary,
-            json_binary_encoding: JsonBinaryEncoding::default(),
-            parsed: false,
-            channel: Channel::FixedRate(FixedRate::from_ms(200).expect("unsupported update rate")),
-        })
-        .expect("invalid subscription params"),
-    };
-
-    client
-        .subscribe(Request::Subscribe(subscription_request))
-        .await?;
-
-    println!("Subscribed to BTC/USD price feed. Waiting for updates...");
-
-    // Process the first 50 updates
+
+    let subscription_requests = vec![
+        // Example subscription: Parsed JSON feed targeting Solana
+        SubscribeRequest {
+            subscription_id: SubscriptionId(1),
+            params: SubscriptionParams::new(SubscriptionParamsRepr {
+                price_feed_ids: vec![PriceFeedId(1), PriceFeedId(2)],
+                properties: vec![
+                    PriceFeedProperty::Price,
+                    PriceFeedProperty::Exponent,
+                    PriceFeedProperty::BestAskPrice,
+                    PriceFeedProperty::BestBidPrice,
+                ],
+                chains: vec![Chain::Solana],
+                delivery_format: DeliveryFormat::Json,
+                json_binary_encoding: JsonBinaryEncoding::default(),
+                parsed: true,
+                channel: Channel::FixedRate(
+                    FixedRate::from_ms(200).expect("unsupported update rate"),
+                ),
+            })
+            .expect("invalid subscription params"),
+        },
+        // Example subscription: binary feed targeting Solana and EVM
+        SubscribeRequest {
+            subscription_id: SubscriptionId(2),
+            params: SubscriptionParams::new(SubscriptionParamsRepr {
+                price_feed_ids: vec![PriceFeedId(3), PriceFeedId(4)],
+                properties: vec![
+                    PriceFeedProperty::Price,
+                    PriceFeedProperty::Exponent,
+                    PriceFeedProperty::BestAskPrice,
+                    PriceFeedProperty::BestBidPrice,
+                ],
+                chains: vec![Chain::Evm, Chain::Solana],
+                delivery_format: DeliveryFormat::Binary,
+                json_binary_encoding: JsonBinaryEncoding::default(),
+                parsed: false,
+                channel: Channel::FixedRate(
+                    FixedRate::from_ms(50).expect("unsupported update rate"),
+                ),
+            })
+            .expect("invalid subscription params"),
+        },
+    ];
+
+    for req in subscription_requests {
+        client.subscribe(Request::Subscribe(req)).await?;
+    }
+
+    println!("Subscribed to price feeds. Waiting for updates...");
+
+    // Process the first few updates
     let mut count = 0;
     while let Some(msg) = stream.next().await {
         println!("Received update: {:?}", msg?);
         count += 1;
-        if count >= 10 {
+        if count >= 50 {
             break;
         }
     }
 
     // Unsubscribe before exiting
-    client.unsubscribe(subscription_id).await?;
-    println!("Unsubscribed from {:?}", subscription_id);
+    for sub_id in [SubscriptionId(1), SubscriptionId(2)] {
+        client.unsubscribe(sub_id).await?;
+        println!("Unsubscribed from {:?}", sub_id);
+    }
 
     tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
     client.close().await?;

lazer/sdk/rust/client/examples/subscribe_price_feeds.rs

@@ -1,30 +1,10 @@
 use anyhow::Result;
-use base64::Engine;
 use futures_util::{SinkExt, StreamExt};
-use pyth_lazer_protocol::{
-    message::{EvmMessage, SolanaMessage},
-    payload::{
-        BINARY_UPDATE_FORMAT_MAGIC, EVM_FORMAT_MAGIC, PARSED_FORMAT_MAGIC, SOLANA_FORMAT_MAGIC_BE,
-    },
-    router::{JsonBinaryData, JsonBinaryEncoding, JsonUpdate},
-    subscription::{
-        ErrorResponse, Request, Response, StreamUpdatedResponse, SubscriptionId, UnsubscribeRequest,
-    },
+use pyth_lazer_protocol::subscription::{
+    ErrorResponse, Request, Response, SubscriptionId, UnsubscribeRequest,
 };
 use tokio_tungstenite::{connect_async, tungstenite::Message};
 use url::Url;
-
-/// Response type for binary messages containing chain-specific data
-#[derive(Debug)]
-pub enum BinaryResponse {
-    /// EVM chain message with payload and signature
-    Evm(EvmMessage),
-    /// Solana chain message with payload and signature
-    Solana(SolanaMessage),
-    /// Parsed JSON payload for human-readable format
-    Parsed(serde_json::Value),
-}
-
 /// A WebSocket client for consuming Pyth Lazer price feed updates
 ///
 /// This client provides a simple interface to:
@@ -86,59 +66,7 @@ impl LazerClient {
             let msg = msg?;
             match msg {
                 Message::Text(text) => Ok(serde_json::from_str(&text)?),
-                Message::Binary(data) => {
-                    let mut pos = 0;
-                    let magic = u32::from_be_bytes(data[pos..pos + 4].try_into()?);
-                    pos += 4;
-
-                    if magic != BINARY_UPDATE_FORMAT_MAGIC {
-                        anyhow::bail!("binary update format magic mismatch");
-                    }
-
-                    let subscription_id =
-                        SubscriptionId(u64::from_be_bytes(data[pos..pos + 8].try_into()?));
-                    pos += 8;
-
-                    let mut evm = None;
-                    let mut solana = None;
-                    let mut parsed = None;
-
-                    while pos < data.len() {
-                        let len = u16::from_be_bytes(data[pos..pos + 2].try_into()?) as usize;
-                        pos += 2;
-                        let magic = u32::from_be_bytes(data[pos..pos + 4].try_into()?);
-
-                        match magic {
-                            EVM_FORMAT_MAGIC => {
-                                evm = Some(EvmMessage::deserialize_slice(&data[pos..pos + len])?);
-                            }
-                            SOLANA_FORMAT_MAGIC_BE => {
-                                solana =
-                                    Some(SolanaMessage::deserialize_slice(&data[pos..pos + len])?);
-                            }
-                            PARSED_FORMAT_MAGIC => {
-                                parsed = Some(serde_json::from_slice(&data[pos + 4..pos + len])?);
-                            }
-                            _ => anyhow::bail!("unknown magic: {}", magic),
-                        }
-                        pos += len;
-                    }
-
-                    Ok(Response::StreamUpdated(StreamUpdatedResponse {
-                        subscription_id,
-                        payload: JsonUpdate {
-                            evm: evm.map(|m| JsonBinaryData {
-                                encoding: JsonBinaryEncoding::Base64,
-                                data: base64::engine::general_purpose::STANDARD.encode(&m.payload),
-                            }),
-                            solana: solana.map(|m| JsonBinaryData {
-                                encoding: JsonBinaryEncoding::Base64,
-                                data: base64::engine::general_purpose::STANDARD.encode(&m.payload),
-                            }),
-                            parsed,
-                        },
-                    }))
-                }
+                Message::Binary(data) => Ok(Response::from_binary(&data)?),
                 Message::Close(_) => Ok(Response::Error(ErrorResponse {
                     error: "WebSocket connection closed".to_string(),
                 })),

lazer/sdk/rust/client/src/lib.rs

@@ -10,9 +10,11 @@ repository = "https://github.com/pyth-network/pyth-crosschain"
 byteorder = "1.5.0"
 anyhow = "1.0.89"
 serde = { version = "1.0.210", features = ["derive"] }
+serde_json = "1.0"
 derive_more = { version = "1.0.0", features = ["from"] }
 itertools = "0.13.0"
 rust_decimal = "1.36.0"
+base64 = "0.22.1"
 
 [dev-dependencies]
 bincode = "1.3.3"

lazer/sdk/rust/protocol/Cargo.toml

@@ -2,7 +2,16 @@
 //! used across publishers, agents and routers.
 
 use {
-    crate::router::{JsonUpdate, SubscriptionParams},
+    crate::{
+        message::{EvmMessage, SolanaMessage},
+        payload::{
+            BINARY_UPDATE_FORMAT_MAGIC, EVM_FORMAT_MAGIC, PARSED_FORMAT_MAGIC,
+            SOLANA_FORMAT_MAGIC_BE,
+        },
+        router::{JsonBinaryData, JsonBinaryEncoding, JsonUpdate, SubscriptionParams},
+    },
+    anyhow::bail,
+    base64::Engine,
     derive_more::From,
     serde::{Deserialize, Serialize},
 };
@@ -33,7 +42,7 @@ pub struct UnsubscribeRequest {
     pub subscription_id: SubscriptionId,
 }
 
-/// A response sent from the server to the client.
+/// A JSON response sent from the server to the client.
 #[derive(Debug, Clone, Serialize, Deserialize, From)]
 #[serde(tag = "type")]
 #[serde(rename_all = "camelCase")]
@@ -45,6 +54,60 @@ pub enum Response {
     StreamUpdated(StreamUpdatedResponse),
 }
 
+impl Response {
+    /// Parse a binary server message into a Response
+    pub fn from_binary(data: &[u8]) -> anyhow::Result<Self> {
+        let mut pos = 0;
+        let magic = u32::from_be_bytes(data[pos..pos + 4].try_into()?);
+        pos += 4;
+
+        if magic != BINARY_UPDATE_FORMAT_MAGIC {
+            bail!("binary update format magic mismatch");
+        }
+
+        let subscription_id = SubscriptionId(u64::from_be_bytes(data[pos..pos + 8].try_into()?));
+        pos += 8;
+
+        let mut evm = None;
+        let mut solana = None;
+        let mut parsed = None;
+
+        while pos < data.len() {
+            let len = u16::from_be_bytes(data[pos..pos + 2].try_into()?) as usize;
+            pos += 2;
+            let magic = u32::from_be_bytes(data[pos..pos + 4].try_into()?);
+
+            match magic {
+                EVM_FORMAT_MAGIC => {
+                    evm = Some(EvmMessage::deserialize_slice(&data[pos..pos + len])?);
+                }
+                SOLANA_FORMAT_MAGIC_BE => {
+                    solana = Some(SolanaMessage::deserialize_slice(&data[pos..pos + len])?);
+                }
+                PARSED_FORMAT_MAGIC => {
+                    parsed = Some(serde_json::from_slice(&data[pos + 4..pos + len])?);
+                }
+                _ => bail!("unknown magic: {}", magic),
+            }
+            pos += len;
+        }
+
+        Ok(Response::StreamUpdated(StreamUpdatedResponse {
+            subscription_id,
+            payload: JsonUpdate {
+                evm: evm.map(|m| JsonBinaryData {
+                    encoding: JsonBinaryEncoding::Base64,
+                    data: base64::engine::general_purpose::STANDARD.encode(&m.payload),
+                }),
+                solana: solana.map(|m| JsonBinaryData {
+                    encoding: JsonBinaryEncoding::Base64,
+                    data: base64::engine::general_purpose::STANDARD.encode(&m.payload),
+                }),
+                parsed,
+            },
+        }))
+    }
+}
 /// Sent from the server after a successul subscription.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]