impl dedup

Author: keyvankhademi

Cargo.lock

@@ -19,6 +19,8 @@ tracing = "0.1"
 url = "2.4"
 derive_more = { version = "1.0.0", features = ["from"] }
 backoff = { version = "0.4.0", features = ["futures", "tokio"] }
+ttl_cache = "0.5.1"
+
 
 [dev-dependencies]
 bincode = "1.3.3"

lazer/sdk/rust/client/Cargo.toml

@@ -36,7 +36,7 @@ async fn main() -> anyhow::Result<()> {
     let mut client = PythLazerClient::new(
         vec!["wss://pyth-lazer.dourolabs.app/v1/stream".to_string()],
         get_lazer_access_token(),
-        1,
+        2,
         ExponentialBackoffBuilder::default()
             .with_max_elapsed_time(None) // max_elapsed_time is not supported in Pyth Lazer client
             .build(),

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

@@ -1,3 +1,8 @@
+use std::{
+    hash::{DefaultHasher, Hash},
+    time::Duration,
+};
+
 use crate::{
     resilient_ws_connection::PythLazerResilientWSConnection, ws_connection::AnyResponse,
     CHANNEL_CAPACITY,
@@ -9,6 +14,10 @@ use pyth_lazer_protocol::subscription::{SubscribeRequest, SubscriptionId};
 use tokio::sync::mpsc::{self, error::TrySendError};
 use tokio_stream::{wrappers::ReceiverStream, StreamExt};
 use tracing::{error, warn};
+use ttl_cache::TtlCache;
+
+const DEDUP_CACHE_SIZE: usize = 100_000;
+const DEDUP_TTL: Duration = Duration::from_secs(10);
 
 pub struct PythLazerClient {
     endpoints: Vec<String>,
@@ -63,9 +72,16 @@ impl PythLazerClient {
 
         let streams: Vec<_> = self.receivers.drain(..).map(ReceiverStream::new).collect();
         let mut merged_stream = stream::select_all(streams);
+        let mut seen_updates = TtlCache::new(DEDUP_CACHE_SIZE);
 
         tokio::spawn(async move {
             while let Some(response) = merged_stream.next().await {
+                let cache_key = response.cache_key();
+                if seen_updates.contains_key(&cache_key) {
+                    continue;
+                }
+                seen_updates.insert(cache_key, response.clone(), DEDUP_TTL);
+
                 match sender.try_send(response) {
                     Ok(_) => (),
                     Err(TrySendError::Full(r)) => {

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

@@ -1,3 +1,5 @@
+use std::hash::{DefaultHasher, Hash, Hasher};
+
 use anyhow::Result;
 use derive_more::From;
 use futures_util::{SinkExt, StreamExt, TryStreamExt};
@@ -34,6 +36,13 @@ pub enum AnyResponse {
     Binary(BinaryWsUpdate),
 }
 
+impl AnyResponse {
+    pub fn cache_key(&self) -> u64 {
+        let mut hasher = DefaultHasher::new();
+        self.hash(&mut hasher);
+        hasher.finish()
+    }
+}
 impl PythLazerWSConnection {
     /// Creates a new Lazer client instance
     ///