feat(lazer): add state endpoint support to history client (#3175)

* feat(lazer): add state endpoint support to history client

* chore: bump pyth-lazer-client version to 9.0.0

Author: Riateche

Cargo.lock

@@ -1,13 +1,16 @@
 [package]
 name = "pyth-lazer-client"
-version = "8.6.1"
+version = "9.0.0"
 edition = "2021"
 description = "A Rust client for Pyth Lazer"
 license = "Apache-2.0"
 
 [dependencies]
 pyth-lazer-protocol = { path = "../protocol", version = "0.20.1" }
+pyth-lazer-publisher-sdk = { path = "../../../publisher_sdk/rust", version = "0.20.1" }
+
 tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1.17"
 tokio-tungstenite = { version = "0.20", features = ["native-tls"] }
 futures-util = "0.3"
 serde = { version = "1.0", features = ["derive"] }
@@ -25,7 +28,9 @@ futures = "0.3.31"
 humantime-serde = "1.1.1"
 fs-err = "3.1.1"
 atomicwrites = "0.4.4"
-
+protobuf-json-mapping = "3.7.2"
+serde_with = "3.15.1"
+async-trait = "0.1.89"
 
 [dev-dependencies]
 bincode = "1.3.3"

lazer/sdk/rust/client/Cargo.toml

@@ -0,0 +1,45 @@
+use {
+    pyth_lazer_client::arc_swap::StreamIntoAutoUpdatedHandle,
+    pyth_lazer_client::history_client::GetStateParams,
+    pyth_lazer_client::history_client::{PythLazerHistoryClient, PythLazerHistoryClientConfig},
+    std::{env, time::Duration},
+    tokio::time::sleep,
+    url::Url,
+};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    let urls = std::env::args()
+        .skip(1)
+        .map(|s| Url::parse(&s))
+        .collect::<Result<Vec<_>, _>>()?;
+
+    let client = PythLazerHistoryClient::new(PythLazerHistoryClientConfig {
+        urls,
+        update_interval: Duration::from_secs(5),
+        access_token: Some(env::var("ACCESS_TOKEN")?),
+        ..Default::default()
+    });
+    let state = client
+        .state_stream(GetStateParams {
+            publishers: true,
+            ..Default::default()
+        })
+        .await?
+        .into_auto_updated_handle()
+        .await?;
+
+    loop {
+        println!("publishers len: {}", state.load().publishers.len());
+        println!(
+            "publisher 1: {:?}",
+            state
+                .load()
+                .publishers
+                .iter()
+                .find(|p| p.publisher_id == Some(1))
+        );
+        sleep(Duration::from_secs(15)).await;
+    }
+}

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

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use {pyth_lazer_client::arc_swap::StreamIntoAutoUpdatedHandle, std::time::Duration};
 
 use pyth_lazer_client::history_client::{PythLazerHistoryClient, PythLazerHistoryClientConfig};
 use pyth_lazer_protocol::PriceFeedId;
@@ -18,11 +18,21 @@ async fn main() -> anyhow::Result<()> {
         update_interval: Duration::from_secs(5),
         ..Default::default()
     });
-    let symbols = client.all_symbols_metadata_handle().await?;
+    let symbols = client
+        .all_symbols_metadata_stream()
+        .await?
+        .into_auto_updated_handle()
+        .await?;
 
     loop {
-        println!("symbols len: {}", symbols.symbols().len());
-        println!("symbol 1: {:?}", symbols.symbols().get(&PriceFeedId(1)));
+        println!("symbols len: {}", symbols.load().len());
+        println!(
+            "symbol 1: {:?}",
+            symbols
+                .load()
+                .iter()
+                .find(|feed| feed.pyth_lazer_id == PriceFeedId(1))
+        );
         sleep(Duration::from_secs(15)).await;
     }
 }

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

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use {futures::StreamExt, std::time::Duration};
 
 use pyth_lazer_client::history_client::{PythLazerHistoryClient, PythLazerHistoryClientConfig};
 use pyth_lazer_protocol::PriceFeedId;
@@ -19,7 +19,7 @@ async fn main() -> anyhow::Result<()> {
     });
     let mut symbols_stream = client.all_symbols_metadata_stream().await?;
 
-    while let Some(symbols) = symbols_stream.recv().await {
+    while let Some(symbols) = symbols_stream.next().await {
         println!("symbols len: {}", symbols.len());
         println!(
             "symbol 1: {:?}",

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

@@ -0,0 +1,39 @@
+use {
+    anyhow::Context as _,
+    arc_swap::ArcSwap,
+    futures::Stream,
+    futures_util::StreamExt as _,
+    std::sync::Arc,
+    tracing::{info, Instrument as _},
+};
+
+#[async_trait::async_trait]
+pub trait StreamIntoAutoUpdatedHandle: Stream + Unpin + Sized + 'static
+where
+    Self::Item: Send + Sync,
+{
+    /// Create an `ArcSwap` that provides access to the most recent value produced by the stream.
+    async fn into_auto_updated_handle(mut self) -> anyhow::Result<Arc<ArcSwap<Self::Item>>> {
+        let first_value = self
+            .next()
+            .await
+            .context("cannot create auto updated handle from empty stream")?;
+        let handle = Arc::new(ArcSwap::new(Arc::new(first_value)));
+        let weak_handle = Arc::downgrade(&handle);
+        tokio::spawn(
+            async move {
+                while let Some(value) = self.next().await {
+                    let Some(handle) = weak_handle.upgrade() else {
+                        info!("handle dropped, stopping auto handle update task");
+                        return;
+                    };
+                    handle.store(Arc::new(value));
+                }
+            }
+            .in_current_span(),
+        );
+        Ok(handle)
+    }
+}
+
+impl<T: Stream + Unpin + 'static> StreamIntoAutoUpdatedHandle for T where T::Item: Send + Sync {}