pyth_hermes_client_rust/client.rs
1//! # Hermes Client
2//!
3//! This module provides a high-level client for connecting to Hermes data streams.
4//! The client maintains multiple WebSocket connections for redundancy and provides
5//! automatic deduplication of messages.
6//!
7//! ## Features
8//!
9//! - Multiple redundant WebSocket connections
10//! - Automatic message deduplication
11//! - Exponential backoff for reconnections
12//! - Configurable timeouts and channel capacities
13//! - Builder pattern for easy configuration
14//!
15//! ## Basic Usage
16//!
17//! ```rust,ignore
18//! use hermes_client::{HermesClientBuilder, HermesClientMessageSubscribe};
19//!
20//! #[tokio::main]
21//! async fn main() -> anyhow::Result<()> {
22//! let mut client = HermesClientBuilder::default()
23//! .with_num_connections(2)
24//! .build()?;
25//!
26//! let mut receiver = client.start().await?;
27//!
28//! // Subscribe to price feeds
29//! let subscribe_request = HermesClientMessageSubscribe {
30//! // ... configure subscription
31//! };
32//! client.subscribe(subscribe_request).await?;
33//!
34//! // Process incoming messages
35//! while let Some(response) = receiver.recv().await {
36//! println!("Received: {:?}", response);
37//! }
38//!
39//! Ok(())
40//! }
41//! ```
42
43use std::time::Duration;
44
45use crate::{
46 backoff::{HermesExponentialBackoff, HermesExponentialBackoffBuilder},
47 resilient_ws_connection::HermesResilientWSConnection,
48 ws_connection::{
49 HermesClientMessage, HermesClientMessageSubscribe, HermesClientMessageUnsubscribe,
50 HermesServerMessage,
51 },
52 CHANNEL_CAPACITY,
53};
54use anyhow::{bail, Result};
55use backoff::ExponentialBackoff;
56use tokio::sync::mpsc::{self, error::TrySendError};
57use tracing::{error, warn};
58use ttl_cache::TtlCache;
59use url::Url;
60
61const DEDUP_CACHE_SIZE: usize = 100_000;
62const DEDUP_TTL: Duration = Duration::from_secs(10);
63
64const DEFAULT_ENDPOINTS: [&str; 1] = ["wss://hermes.pyth.network/ws"];
65const DEFAULT_NUM_CONNECTIONS: usize = 3;
66const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
67
68/// A high-performance client for connecting to Hermes data streams.
69///
70/// The `HermesClient` maintains multiple WebSocket connections to Hermes endpoints
71/// for redundancy. It automatically handles connection management,
72/// message deduplication, and provides a unified stream of price updates.
73///
74/// ## Architecture
75///
76/// - Maintains multiple WebSocket connections to different endpoints
77/// - Uses a TTL cache for deduplicating messages across connections
78/// - Provides a single channel for consuming deduplicated messages
79/// - Handles connection failures with exponential backoff
80pub struct HermesClient {
81 endpoints: Vec<Url>,
82 num_connections: usize,
83 ws_connections: Vec<HermesResilientWSConnection>,
84 backoff: ExponentialBackoff,
85 timeout: Duration,
86 channel_capacity: usize,
87 access_token: Option<String>,
88}
89
90impl HermesClient {
91 /// Creates a new Hermes client instance.
92 ///
93 /// This is a low-level constructor. Consider using [`HermesClientBuilder`] for a more
94 /// convenient way to create clients with sensible defaults.
95 ///
96 /// # Arguments
97 ///
98 /// * `endpoints` - A vector of WebSocket endpoint URLs to connect to. Must not be empty.
99 /// * `num_connections` - The number of WebSocket connections to maintain for redundancy
100 /// * `backoff` - The exponential backoff configuration for connection retries
101 /// * `timeout` - The timeout duration for WebSocket operations
102 /// * `channel_capacity` - The capacity of the message channel
103 ///
104 /// # Returns
105 ///
106 /// Returns `Ok(HermesClient)` on success, or an error if the configuration is invalid.
107 ///
108 /// # Errors
109 ///
110 /// Returns an error if:
111 /// - The `endpoints` vector is empty
112 ///
113 pub fn new(
114 endpoints: Vec<Url>,
115 num_connections: usize,
116 backoff: HermesExponentialBackoff,
117 timeout: Duration,
118 channel_capacity: usize,
119 access_token: Option<String>,
120 ) -> Result<Self> {
121 if endpoints.is_empty() {
122 bail!("At least one endpoint must be provided");
123 }
124 Ok(Self {
125 endpoints,
126 num_connections,
127 ws_connections: Vec::with_capacity(num_connections),
128 backoff: backoff.into(),
129 timeout,
130 channel_capacity,
131 access_token,
132 })
133 }
134
135 /// Starts the client and begins establishing WebSocket connections.
136 ///
137 /// This method initializes all WebSocket connections and starts the message processing
138 /// loop. It returns a receiver channel that will yield deduplicated messages from
139 /// all connections.
140 ///
141 /// # Returns
142 ///
143 /// Returns a `Receiver<HermesServerMessage>` that yields deduplicated messages from all
144 /// WebSocket connections. The receiver will continue to yield messages until
145 /// all connections are closed or the client is dropped.
146 ///
147 /// # Errors
148 ///
149 /// This method itself doesn't return errors, but individual connection failures
150 /// are handled internally with automatic reconnection using the configured backoff
151 /// strategy.
152 ///
153 /// # Message Deduplication
154 ///
155 /// Messages are deduplicated using a TTL cache with a 10-second window. This ensures
156 /// that identical messages received from multiple connections are only delivered once.
157 ///
158 pub async fn start(&mut self) -> Result<mpsc::Receiver<HermesServerMessage>> {
159 let (sender, receiver) = mpsc::channel::<HermesServerMessage>(self.channel_capacity);
160 let (ws_connection_sender, mut ws_connection_receiver) =
161 mpsc::channel::<HermesServerMessage>(CHANNEL_CAPACITY);
162
163 for i in 0..self.num_connections {
164 let mut endpoint = self.endpoints[i % self.endpoints.len()].clone();
165 // Add access token as query parameter if provided
166 if let Some(ref token) = self.access_token {
167 endpoint
168 .query_pairs_mut()
169 .append_pair("ACCESS_TOKEN", token);
170 }
171 let connection = HermesResilientWSConnection::new(
172 endpoint,
173 self.backoff.clone(),
174 self.timeout,
175 ws_connection_sender.clone(),
176 );
177 self.ws_connections.push(connection);
178 }
179
180 let mut seen_updates = TtlCache::new(DEDUP_CACHE_SIZE);
181
182 tokio::spawn(async move {
183 while let Some(response) = ws_connection_receiver.recv().await {
184 let cache_key = response.cache_key();
185 if seen_updates.contains_key(&cache_key) {
186 continue;
187 }
188 seen_updates.insert(cache_key, response.clone(), DEDUP_TTL);
189
190 match sender.try_send(response) {
191 Ok(_) => (),
192 Err(TrySendError::Full(r)) => {
193 warn!("Sender channel is full, responses will be delayed");
194 if sender.send(r).await.is_err() {
195 error!("Sender channel is closed, stopping client");
196 }
197 }
198 Err(TrySendError::Closed(_)) => {
199 error!("Sender channel is closed, stopping client");
200 }
201 }
202 }
203 });
204
205 Ok(receiver)
206 }
207
208 /// Subscribes to data streams across all WebSocket connections.
209 ///
210 /// This method sends the subscription request to all active WebSocket connections,
211 /// ensuring redundancy. If any connection fails to subscribe,
212 /// an error is returned, but other connections may still be subscribed.
213 ///
214 /// # Arguments
215 ///
216 /// * `subscribe_request` - The subscription request specifying which data streams to subscribe to
217 ///
218 /// # Returns
219 ///
220 /// Returns `Ok(())` if the subscription was successfully sent to all connections,
221 /// or an error if any connection failed to process the subscription.
222 ///
223 pub async fn subscribe(
224 &mut self,
225 subscribe_request: HermesClientMessageSubscribe,
226 ) -> Result<()> {
227 for connection in &mut self.ws_connections {
228 connection
229 .send_request(HermesClientMessage::Subscribe(subscribe_request.clone()))
230 .await?;
231 }
232 Ok(())
233 }
234
235 /// Unsubscribes from a specific data stream across all WebSocket connections.
236 ///
237 /// This method sends an unsubscribe request for the specified subscription ID
238 /// to all active WebSocket connections.
239 ///
240 /// # Arguments
241 ///
242 /// * `unsubscribe_request` - The unsubscribe request specifying which data streams to unsubscribe from
243 ///
244 /// # Returns
245 ///
246 /// Returns `Ok(())` if the unsubscribe request was successfully sent to all connections,
247 /// or an error if any connection failed to process the request.
248 ///
249 pub async fn unsubscribe(
250 &mut self,
251 unsubscribe_request: HermesClientMessageUnsubscribe,
252 ) -> Result<()> {
253 for connection in &mut self.ws_connections {
254 connection
255 .send_request(HermesClientMessage::Unsubscribe(
256 unsubscribe_request.clone(),
257 ))
258 .await?;
259 }
260 Ok(())
261 }
262}
263
264/// A builder for creating [`HermesClient`] instances with customizable configuration.
265///
266/// The builder provides a convenient way to configure a Hermes client with sensible
267/// defaults while allowing customization of all parameters. It follows the builder pattern
268/// for a fluent API.
269///
270/// ## Default Configuration
271///
272/// - **Endpoints**: Uses Hermes's default production endpoints
273/// - **Connections**: 3 concurrent WebSocket connections
274/// - **Timeout**: 5 seconds for WebSocket operations
275/// - **Backoff**: Exponential backoff with default settings
276/// - **Channel Capacity**: Uses the default 1000
277/// - **Access Token**: None (no authentication)
278///
279pub struct HermesClientBuilder {
280 endpoints: Vec<Url>,
281 num_connections: usize,
282 backoff: HermesExponentialBackoff,
283 timeout: Duration,
284 channel_capacity: usize,
285 access_token: Option<String>,
286}
287
288impl Default for HermesClientBuilder {
289 fn default() -> Self {
290 Self {
291 endpoints: DEFAULT_ENDPOINTS
292 .iter()
293 .map(|&s| s.parse().unwrap())
294 .collect(),
295 num_connections: DEFAULT_NUM_CONNECTIONS,
296 backoff: HermesExponentialBackoffBuilder::default().build(),
297 timeout: DEFAULT_TIMEOUT,
298 channel_capacity: CHANNEL_CAPACITY,
299 access_token: None,
300 }
301 }
302}
303
304impl HermesClientBuilder {
305 /// Sets custom WebSocket endpoints for the client.
306 ///
307 /// By default, the client uses Hermes's production endpoints. Use this method
308 /// to connect to different environments (staging, local development) or to use
309 /// custom endpoint configurations.
310 ///
311 /// # Arguments
312 ///
313 /// * `endpoints` - A vector of WebSocket endpoint URLs. Must not be empty.
314 ///
315 pub fn with_endpoints(mut self, endpoints: Vec<Url>) -> Self {
316 self.endpoints = endpoints;
317 self
318 }
319
320 /// Sets the number of concurrent WebSocket connections to maintain.
321 ///
322 /// More connections provide better redundancy and can improve throughput,
323 /// but also consume more resources.
324 ///
325 /// # Arguments
326 ///
327 /// * `num_connections` - The number of WebSocket connections (must be > 0)
328 ///
329 pub fn with_num_connections(mut self, num_connections: usize) -> Self {
330 self.num_connections = num_connections;
331 self
332 }
333
334 /// Sets the exponential backoff configuration for connection retries.
335 ///
336 /// The backoff strategy determines how the client handles connection failures
337 /// and retries.
338 ///
339 /// # Arguments
340 ///
341 /// * `backoff` - The exponential backoff configuration
342 ///
343 pub fn with_backoff(mut self, backoff: HermesExponentialBackoff) -> Self {
344 self.backoff = backoff;
345 self
346 }
347
348 /// Sets the timeout duration for WebSocket operations.
349 ///
350 /// This timeout applies to each WebSocket connection,
351 /// if no response is received within this duration,
352 /// the connection will be considered failed and retried.
353 ///
354 /// # Arguments
355 ///
356 /// * `timeout` - The timeout duration for each WebSocket
357 ///
358 pub fn with_timeout(mut self, timeout: Duration) -> Self {
359 self.timeout = timeout;
360 self
361 }
362
363 /// Sets the capacity of the internal message channel.
364 ///
365 /// This determines how many messages can be buffered internally before
366 /// the client starts applying backpressure.
367 ///
368 /// # Arguments
369 ///
370 /// * `channel_capacity` - The channel capacity (number of messages)
371 ///
372 pub fn with_channel_capacity(mut self, channel_capacity: usize) -> Self {
373 self.channel_capacity = channel_capacity;
374 self
375 }
376
377 /// Sets the API access token for authentication.
378 ///
379 /// When provided, this token will be included in all WebSocket connections
380 /// as an `ACCESS_TOKEN` query parameter.
381 ///
382 /// # Arguments
383 ///
384 /// * `access_token` - The API access token for authentication
385 ///
386 pub fn with_access_token(mut self, access_token: String) -> Self {
387 self.access_token = Some(access_token);
388 self
389 }
390
391 /// Builds the configured [`HermesClient`] instance.
392 ///
393 /// This consumes the builder and creates a new client with the specified
394 /// configuration. The client is ready to use but connections are not
395 /// established until [`HermesClient::start`] is called.
396 ///
397 /// # Returns
398 ///
399 /// Returns `Ok(HermesClient)` on success, or an error if the configuration
400 /// is invalid.
401 ///
402 /// # Errors
403 ///
404 /// Returns an error if:
405 /// - No endpoints are configured
406 /// - Any configuration parameter is invalid
407 ///
408 pub fn build(self) -> Result<HermesClient> {
409 HermesClient::new(
410 self.endpoints,
411 self.num_connections,
412 self.backoff,
413 self.timeout,
414 self.channel_capacity,
415 self.access_token,
416 )
417 }
418}