solana_pubsub_client/
pubsub_client.rs

1//! A client for subscribing to messages from the RPC server.
2//!
3//! The [`PubsubClient`] implements [Solana WebSocket event
4//! subscriptions][spec].
5//!
6//! [spec]: https://solana.com/docs/rpc/websocket
7//!
8//! This is a blocking API. For a non-blocking API use the asynchronous client
9//! in [`crate::nonblocking::pubsub_client`].
10//!
11//! `PubsubClient` contains static methods to subscribe to events, like
12//! [`PubsubClient::account_subscribe`]. These methods each return their own
13//! subscription type, like [`AccountSubscription`], that are typedefs of
14//! tuples, the first element being a handle to the subscription, like
15//! [`AccountSubscription`], the second a [`Receiver`] of [`RpcResponse`] of
16//! whichever type is appropriate for the subscription. The subscription handle
17//! is a typedef of [`PubsubClientSubscription`], and it must remain live for
18//! the receiver to continue receiving messages.
19//!
20//! Because this is a blocking API, with blocking receivers, a reasonable
21//! pattern for using this API is to move each event receiver to its own thread
22//! to block on messages, while holding all subscription handles on a single
23//! primary thread.
24//!
25//! While `PubsubClientSubscription` contains methods for shutting down,
26//! [`PubsubClientSubscription::send_unsubscribe`], and
27//! [`PubsubClientSubscription::shutdown`], because its internal receivers block
28//! on events from the server, these subscriptions cannot actually be shutdown
29//! reliably. For a non-blocking, cancelable API, use the asynchronous client
30//! in [`crate::nonblocking::pubsub_client`].
31//!
32//! By default the [`block_subscribe`] and [`vote_subscribe`] events are
33//! disabled on RPC nodes. They can be enabled by passing
34//! `--rpc-pubsub-enable-block-subscription` and
35//! `--rpc-pubsub-enable-vote-subscription` to `agave-validator`. When these
36//! methods are disabled, the RPC server will return a "Method not found" error
37//! message.
38//!
39//! [`block_subscribe`]: https://docs.rs/solana-rpc/latest/solana_rpc/rpc_pubsub/trait.RpcSolPubSub.html#tymethod.block_subscribe
40//! [`vote_subscribe`]: https://docs.rs/solana-rpc/latest/solana_rpc/rpc_pubsub/trait.RpcSolPubSub.html#tymethod.vote_subscribe
41//!
42//! # Examples
43//!
44//! This example subscribes to account events and then loops forever receiving
45//! them.
46//!
47//! ```
48//! use anyhow::Result;
49//! use solana_commitment_config::CommitmentConfig;
50//! use solana_pubkey::Pubkey;
51//! use solana_pubsub_client::pubsub_client::PubsubClient;
52//! use solana_rpc_client_types::config::RpcAccountInfoConfig;
53//! use std::thread;
54//!
55//! fn get_account_updates(account_pubkey: Pubkey) -> Result<()> {
56//!     let url = "wss://api.devnet.solana.com/";
57//!
58//!     let (mut account_subscription_client, account_subscription_receiver) =
59//!         PubsubClient::account_subscribe(
60//!             url,
61//!             &account_pubkey,
62//!             Some(RpcAccountInfoConfig {
63//!                 encoding: None,
64//!                 data_slice: None,
65//!                 commitment: Some(CommitmentConfig::confirmed()),
66//!                 min_context_slot: None,
67//!             }),
68//!         )?;
69//!
70//!     loop {
71//!         match account_subscription_receiver.recv() {
72//!             Ok(response) => {
73//!                 println!("account subscription response: {:?}", response);
74//!             }
75//!             Err(e) => {
76//!                 println!("account subscription error: {:?}", e);
77//!                 break;
78//!             }
79//!         }
80//!     }
81//!
82//!     Ok(())
83//! }
84//! #
85//! # get_account_updates(solana_pubkey::new_rand());
86//! # Ok::<(), anyhow::Error>(())
87//! ```
88
89pub use crate::nonblocking::pubsub_client::PubsubClientError;
90use {
91    crossbeam_channel::{Receiver, Sender, unbounded},
92    log::*,
93    serde::de::DeserializeOwned,
94    serde_json::{
95        Map, Value, json,
96        value::Value::{Number, Object},
97    },
98    solana_account_decoder_client_types::UiAccount,
99    solana_clock::Slot,
100    solana_pubkey::Pubkey,
101    solana_rpc_client_types::{
102        config::{
103            RpcAccountInfoConfig, RpcBlockSubscribeConfig, RpcBlockSubscribeFilter,
104            RpcProgramAccountsConfig, RpcSignatureSubscribeConfig, RpcTransactionLogsConfig,
105            RpcTransactionLogsFilter,
106        },
107        response::{
108            Response as RpcResponse, RpcBlockUpdate, RpcKeyedAccount, RpcLogsResponse,
109            RpcSignatureResult, RpcVote, SlotInfo, SlotUpdate,
110        },
111    },
112    solana_signature::Signature,
113    std::{
114        marker::PhantomData,
115        net::TcpStream,
116        sync::{
117            Arc, RwLock,
118            atomic::{AtomicBool, Ordering},
119        },
120        thread::{JoinHandle, sleep},
121        time::Duration,
122    },
123    tungstenite::{
124        Message, WebSocket,
125        client::IntoClientRequest,
126        connect,
127        http::{StatusCode, header},
128        stream::MaybeTlsStream,
129    },
130};
131
132/// A subscription.
133///
134/// The subscription is unsubscribed on drop, and note that unsubscription (and
135/// thus drop) time is unbounded. See
136/// [`PubsubClientSubscription::send_unsubscribe`].
137pub struct PubsubClientSubscription<T>
138where
139    T: DeserializeOwned,
140{
141    message_type: PhantomData<T>,
142    operation: &'static str,
143    socket: Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
144    subscription_id: u64,
145    t_cleanup: Option<JoinHandle<()>>,
146    exit: Arc<AtomicBool>,
147}
148
149impl<T> Drop for PubsubClientSubscription<T>
150where
151    T: DeserializeOwned,
152{
153    fn drop(&mut self) {
154        self.send_unsubscribe()
155            .unwrap_or_else(|_| warn!("unable to unsubscribe from websocket"));
156        self.socket
157            .write()
158            .unwrap()
159            .close(None)
160            .unwrap_or_else(|_| warn!("unable to close websocket"));
161    }
162}
163
164impl<T> PubsubClientSubscription<T>
165where
166    T: DeserializeOwned,
167{
168    fn send_subscribe(
169        writable_socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
170        body: String,
171    ) -> Result<u64, PubsubClientError> {
172        writable_socket
173            .write()
174            .unwrap()
175            .send(Message::Text(body.into()))
176            .map_err(Box::new)?;
177        let message = writable_socket.write().unwrap().read().map_err(Box::new)?;
178        Self::extract_subscription_id(message)
179    }
180
181    fn extract_subscription_id(message: Message) -> Result<u64, PubsubClientError> {
182        let message_text = &message.into_text().map_err(Box::new)?;
183
184        if let Ok(json_msg) = serde_json::from_str::<Map<String, Value>>(message_text) {
185            if let Some(Number(x)) = json_msg.get("result") {
186                if let Some(x) = x.as_u64() {
187                    return Ok(x);
188                }
189            }
190        }
191
192        Err(PubsubClientError::UnexpectedSubscriptionResponse(format!(
193            "msg={message_text}"
194        )))
195    }
196
197    /// Send an unsubscribe message to the server.
198    ///
199    /// Note that this will block as long as the internal subscription receiver
200    /// is waiting on messages from the server, and this can take an unbounded
201    /// amount of time if the server does not send any messages.
202    ///
203    /// If a pubsub client needs to shutdown reliably it should use
204    /// the async client in [`crate::nonblocking::pubsub_client`].
205    pub fn send_unsubscribe(&self) -> Result<(), PubsubClientError> {
206        let method = format!("{}Unsubscribe", self.operation);
207        self.socket
208            .write()
209            .unwrap()
210            .send(Message::Text(
211                json!({
212                "jsonrpc":"2.0","id":1,"method":method,"params":[self.subscription_id]
213                })
214                .to_string()
215                .into(),
216            ))
217            .map_err(Box::new)
218            .map_err(|err| err.into())
219    }
220
221    fn read_message(
222        writable_socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
223    ) -> Result<Option<T>, PubsubClientError> {
224        let message = writable_socket.write().unwrap().read().map_err(Box::new)?;
225        if message.is_ping() {
226            return Ok(None);
227        }
228        let message_text = &message.into_text().map_err(Box::new)?;
229        if let Ok(json_msg) = serde_json::from_str::<Map<String, Value>>(message_text) {
230            if let Some(Object(params)) = json_msg.get("params") {
231                if let Some(result) = params.get("result") {
232                    if let Ok(x) = T::deserialize(result) {
233                        return Ok(Some(x));
234                    }
235                }
236            }
237        }
238
239        Err(PubsubClientError::UnexpectedMessageError(format!(
240            "msg={message_text}"
241        )))
242    }
243
244    /// Shutdown the internel message receiver and wait for its thread to exit.
245    ///
246    /// Note that this will block as long as the subscription receiver is
247    /// waiting on messages from the server, and this can take an unbounded
248    /// amount of time if the server does not send any messages.
249    ///
250    /// If a pubsub client needs to shutdown reliably it should use
251    /// the async client in [`crate::nonblocking::pubsub_client`].
252    pub fn shutdown(&mut self) -> std::thread::Result<()> {
253        if self.t_cleanup.is_some() {
254            info!("websocket thread - shutting down");
255            self.exit.store(true, Ordering::Relaxed);
256            let x = self.t_cleanup.take().unwrap().join();
257            info!("websocket thread - shut down.");
258            x
259        } else {
260            warn!("websocket thread - already shut down.");
261            Ok(())
262        }
263    }
264}
265
266pub type PubsubLogsClientSubscription = PubsubClientSubscription<RpcResponse<RpcLogsResponse>>;
267pub type LogsSubscription = (
268    PubsubLogsClientSubscription,
269    Receiver<RpcResponse<RpcLogsResponse>>,
270);
271
272pub type PubsubSlotClientSubscription = PubsubClientSubscription<SlotInfo>;
273pub type SlotsSubscription = (PubsubSlotClientSubscription, Receiver<SlotInfo>);
274
275pub type PubsubSignatureClientSubscription =
276    PubsubClientSubscription<RpcResponse<RpcSignatureResult>>;
277pub type SignatureSubscription = (
278    PubsubSignatureClientSubscription,
279    Receiver<RpcResponse<RpcSignatureResult>>,
280);
281
282pub type PubsubBlockClientSubscription = PubsubClientSubscription<RpcResponse<RpcBlockUpdate>>;
283pub type BlockSubscription = (
284    PubsubBlockClientSubscription,
285    Receiver<RpcResponse<RpcBlockUpdate>>,
286);
287
288pub type PubsubProgramClientSubscription = PubsubClientSubscription<RpcResponse<RpcKeyedAccount>>;
289pub type ProgramSubscription = (
290    PubsubProgramClientSubscription,
291    Receiver<RpcResponse<RpcKeyedAccount>>,
292);
293
294pub type PubsubAccountClientSubscription = PubsubClientSubscription<RpcResponse<UiAccount>>;
295pub type AccountSubscription = (
296    PubsubAccountClientSubscription,
297    Receiver<RpcResponse<UiAccount>>,
298);
299
300pub type PubsubVoteClientSubscription = PubsubClientSubscription<RpcVote>;
301pub type VoteSubscription = (PubsubVoteClientSubscription, Receiver<RpcVote>);
302
303pub type PubsubRootClientSubscription = PubsubClientSubscription<Slot>;
304pub type RootSubscription = (PubsubRootClientSubscription, Receiver<Slot>);
305
306/// A client for subscribing to messages from the RPC server.
307///
308/// See the [module documentation][self].
309pub struct PubsubClient {}
310
311fn connect_with_retry<R: IntoClientRequest>(
312    request: R,
313) -> Result<WebSocket<MaybeTlsStream<TcpStream>>, Box<tungstenite::Error>> {
314    let mut connection_retries = 5;
315    let client_request = request.into_client_request().map_err(Box::new)?;
316    loop {
317        let result = connect(client_request.clone()).map(|(socket, _)| socket);
318        if let Err(tungstenite::Error::Http(response)) = &result {
319            if response.status() == StatusCode::TOO_MANY_REQUESTS && connection_retries > 0 {
320                let mut duration = Duration::from_millis(500);
321                if let Some(retry_after) = response.headers().get(header::RETRY_AFTER) {
322                    if let Ok(retry_after) = retry_after.to_str() {
323                        if let Ok(retry_after) = retry_after.parse::<u64>() {
324                            if retry_after < 120 {
325                                duration = Duration::from_secs(retry_after);
326                            }
327                        }
328                    }
329                }
330
331                connection_retries -= 1;
332                debug!(
333                    "Too many requests: server responded with {response:?}, {connection_retries} \
334                     retries left, pausing for {duration:?}"
335                );
336
337                sleep(duration);
338                continue;
339            }
340        }
341        return result.map_err(Box::new);
342    }
343}
344
345impl PubsubClient {
346    /// Subscribe to account events.
347    ///
348    /// Receives messages of type [`UiAccount`] when an account's lamports or data changes.
349    ///
350    /// # RPC Reference
351    ///
352    /// This method corresponds directly to the [`accountSubscribe`] RPC method.
353    ///
354    /// [`accountSubscribe`]: https://solana.com/docs/rpc/websocket/accountsubscribe
355    pub fn account_subscribe<R: IntoClientRequest>(
356        request: R,
357        pubkey: &Pubkey,
358        config: Option<RpcAccountInfoConfig>,
359    ) -> Result<AccountSubscription, PubsubClientError> {
360        let client_request = request.into_client_request().map_err(Box::new)?;
361        let socket = connect_with_retry(client_request)?;
362        let (sender, receiver) = unbounded();
363
364        let socket = Arc::new(RwLock::new(socket));
365        let socket_clone = socket.clone();
366        let exit = Arc::new(AtomicBool::new(false));
367        let exit_clone = exit.clone();
368        let body = json!({
369            "jsonrpc":"2.0",
370            "id":1,
371            "method":"accountSubscribe",
372            "params":[
373                pubkey.to_string(),
374                config
375            ]
376        })
377        .to_string();
378        let subscription_id = PubsubAccountClientSubscription::send_subscribe(&socket_clone, body)?;
379
380        let t_cleanup = std::thread::spawn(move || {
381            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
382        });
383
384        let result = PubsubClientSubscription {
385            message_type: PhantomData,
386            operation: "account",
387            socket,
388            subscription_id,
389            t_cleanup: Some(t_cleanup),
390            exit,
391        };
392
393        Ok((result, receiver))
394    }
395
396    /// Subscribe to block events.
397    ///
398    /// Receives messages of type [`RpcBlockUpdate`] when a block is confirmed or finalized.
399    ///
400    /// This method is disabled by default. It can be enabled by passing
401    /// `--rpc-pubsub-enable-block-subscription` to `agave-validator`.
402    ///
403    /// # RPC Reference
404    ///
405    /// This method corresponds directly to the [`blockSubscribe`] RPC method.
406    ///
407    /// [`blockSubscribe`]: https://solana.com/docs/rpc/websocket/blocksubscribe
408    pub fn block_subscribe<R: IntoClientRequest>(
409        request: R,
410        filter: RpcBlockSubscribeFilter,
411        config: Option<RpcBlockSubscribeConfig>,
412    ) -> Result<BlockSubscription, PubsubClientError> {
413        let client_request = request.into_client_request().map_err(Box::new)?;
414        let socket = connect_with_retry(client_request)?;
415        let (sender, receiver) = unbounded();
416
417        let socket = Arc::new(RwLock::new(socket));
418        let socket_clone = socket.clone();
419        let exit = Arc::new(AtomicBool::new(false));
420        let exit_clone = exit.clone();
421        let body = json!({
422            "jsonrpc":"2.0",
423            "id":1,
424            "method":"blockSubscribe",
425            "params":[filter, config]
426        })
427        .to_string();
428
429        let subscription_id = PubsubBlockClientSubscription::send_subscribe(&socket_clone, body)?;
430
431        let t_cleanup = std::thread::spawn(move || {
432            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
433        });
434
435        let result = PubsubClientSubscription {
436            message_type: PhantomData,
437            operation: "block",
438            socket,
439            subscription_id,
440            t_cleanup: Some(t_cleanup),
441            exit,
442        };
443
444        Ok((result, receiver))
445    }
446
447    /// Subscribe to transaction log events.
448    ///
449    /// Receives messages of type [`RpcLogsResponse`] when a transaction is committed.
450    ///
451    /// # RPC Reference
452    ///
453    /// This method corresponds directly to the [`logsSubscribe`] RPC method.
454    ///
455    /// [`logsSubscribe`]: https://solana.com/docs/rpc/websocket/logssubscribe
456    pub fn logs_subscribe<R: IntoClientRequest>(
457        request: R,
458        filter: RpcTransactionLogsFilter,
459        config: RpcTransactionLogsConfig,
460    ) -> Result<LogsSubscription, PubsubClientError> {
461        let client_request = request.into_client_request().map_err(Box::new)?;
462        let socket = connect_with_retry(client_request)?;
463        let (sender, receiver) = unbounded();
464
465        let socket = Arc::new(RwLock::new(socket));
466        let socket_clone = socket.clone();
467        let exit = Arc::new(AtomicBool::new(false));
468        let exit_clone = exit.clone();
469        let body = json!({
470            "jsonrpc":"2.0",
471            "id":1,
472            "method":"logsSubscribe",
473            "params":[filter, config]
474        })
475        .to_string();
476
477        let subscription_id = PubsubLogsClientSubscription::send_subscribe(&socket_clone, body)?;
478
479        let t_cleanup = std::thread::spawn(move || {
480            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
481        });
482
483        let result = PubsubClientSubscription {
484            message_type: PhantomData,
485            operation: "logs",
486            socket,
487            subscription_id,
488            t_cleanup: Some(t_cleanup),
489            exit,
490        };
491
492        Ok((result, receiver))
493    }
494
495    /// Subscribe to program account events.
496    ///
497    /// Receives messages of type [`RpcKeyedAccount`] when an account owned
498    /// by the given program changes.
499    ///
500    /// # RPC Reference
501    ///
502    /// This method corresponds directly to the [`programSubscribe`] RPC method.
503    ///
504    /// [`programSubscribe`]: https://solana.com/docs/rpc/websocket/programsubscribe
505    pub fn program_subscribe<R: IntoClientRequest>(
506        request: R,
507        pubkey: &Pubkey,
508        config: Option<RpcProgramAccountsConfig>,
509    ) -> Result<ProgramSubscription, PubsubClientError> {
510        let client_request = request.into_client_request().map_err(Box::new)?;
511        let socket = connect_with_retry(client_request)?;
512        let (sender, receiver) = unbounded();
513
514        let socket = Arc::new(RwLock::new(socket));
515        let socket_clone = socket.clone();
516        let exit = Arc::new(AtomicBool::new(false));
517        let exit_clone = exit.clone();
518
519        let body = json!({
520            "jsonrpc":"2.0",
521            "id":1,
522            "method":"programSubscribe",
523            "params":[
524                pubkey.to_string(),
525                config
526            ]
527        })
528        .to_string();
529        let subscription_id = PubsubProgramClientSubscription::send_subscribe(&socket_clone, body)?;
530
531        let t_cleanup = std::thread::spawn(move || {
532            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
533        });
534
535        let result = PubsubClientSubscription {
536            message_type: PhantomData,
537            operation: "program",
538            socket,
539            subscription_id,
540            t_cleanup: Some(t_cleanup),
541            exit,
542        };
543
544        Ok((result, receiver))
545    }
546
547    /// Subscribe to vote events.
548    ///
549    /// Receives messages of type [`RpcVote`] when a new vote is observed. These
550    /// votes are observed prior to confirmation and may never be confirmed.
551    ///
552    /// This method is disabled by default. It can be enabled by passing
553    /// `--rpc-pubsub-enable-vote-subscription` to `agave-validator`.
554    ///
555    /// # RPC Reference
556    ///
557    /// This method corresponds directly to the [`voteSubscribe`] RPC method.
558    ///
559    /// [`voteSubscribe`]: https://solana.com/docs/rpc/websocket/votesubscribe
560    pub fn vote_subscribe<R: IntoClientRequest>(
561        request: R,
562    ) -> Result<VoteSubscription, PubsubClientError> {
563        let client_request = request.into_client_request().map_err(Box::new)?;
564        let socket = connect_with_retry(client_request)?;
565        let (sender, receiver) = unbounded();
566
567        let socket = Arc::new(RwLock::new(socket));
568        let socket_clone = socket.clone();
569        let exit = Arc::new(AtomicBool::new(false));
570        let exit_clone = exit.clone();
571        let body = json!({
572            "jsonrpc":"2.0",
573            "id":1,
574            "method":"voteSubscribe",
575        })
576        .to_string();
577        let subscription_id = PubsubVoteClientSubscription::send_subscribe(&socket_clone, body)?;
578
579        let t_cleanup = std::thread::spawn(move || {
580            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
581        });
582
583        let result = PubsubClientSubscription {
584            message_type: PhantomData,
585            operation: "vote",
586            socket,
587            subscription_id,
588            t_cleanup: Some(t_cleanup),
589            exit,
590        };
591
592        Ok((result, receiver))
593    }
594
595    /// Subscribe to root events.
596    ///
597    /// Receives messages of type [`Slot`] when a new [root] is set by the
598    /// validator.
599    ///
600    /// [root]: https://solana.com/docs/terminology#root
601    ///
602    /// # RPC Reference
603    ///
604    /// This method corresponds directly to the [`rootSubscribe`] RPC method.
605    ///
606    /// [`rootSubscribe`]: https://solana.com/docs/rpc/websocket/rootsubscribe
607    pub fn root_subscribe<R: IntoClientRequest>(
608        request: R,
609    ) -> Result<RootSubscription, PubsubClientError> {
610        let client_request = request.into_client_request().map_err(Box::new)?;
611        let socket = connect_with_retry(client_request)?;
612        let (sender, receiver) = unbounded();
613
614        let socket = Arc::new(RwLock::new(socket));
615        let socket_clone = socket.clone();
616        let exit = Arc::new(AtomicBool::new(false));
617        let exit_clone = exit.clone();
618        let body = json!({
619            "jsonrpc":"2.0",
620            "id":1,
621            "method":"rootSubscribe",
622        })
623        .to_string();
624        let subscription_id = PubsubRootClientSubscription::send_subscribe(&socket_clone, body)?;
625
626        let t_cleanup = std::thread::spawn(move || {
627            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
628        });
629
630        let result = PubsubClientSubscription {
631            message_type: PhantomData,
632            operation: "root",
633            socket,
634            subscription_id,
635            t_cleanup: Some(t_cleanup),
636            exit,
637        };
638
639        Ok((result, receiver))
640    }
641
642    /// Subscribe to transaction confirmation events.
643    ///
644    /// Receives messages of type [`RpcSignatureResult`] when a transaction
645    /// with the given signature is committed.
646    ///
647    /// This is a subscription to a single notification. It is automatically
648    /// cancelled by the server once the notification is sent.
649    ///
650    /// # RPC Reference
651    ///
652    /// This method corresponds directly to the [`signatureSubscribe`] RPC method.
653    ///
654    /// [`signatureSubscribe`]: https://solana.com/docs/rpc/websocket/signaturesubscribe
655    pub fn signature_subscribe<R: IntoClientRequest>(
656        request: R,
657        signature: &Signature,
658        config: Option<RpcSignatureSubscribeConfig>,
659    ) -> Result<SignatureSubscription, PubsubClientError> {
660        let client_request = request.into_client_request().map_err(Box::new)?;
661        let socket = connect_with_retry(client_request)?;
662        let (sender, receiver) = unbounded();
663
664        let socket = Arc::new(RwLock::new(socket));
665        let socket_clone = socket.clone();
666        let exit = Arc::new(AtomicBool::new(false));
667        let exit_clone = exit.clone();
668        let body = json!({
669            "jsonrpc":"2.0",
670            "id":1,
671            "method":"signatureSubscribe",
672            "params":[
673                signature.to_string(),
674                config
675            ]
676        })
677        .to_string();
678        let subscription_id =
679            PubsubSignatureClientSubscription::send_subscribe(&socket_clone, body)?;
680
681        let t_cleanup = std::thread::spawn(move || {
682            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
683        });
684
685        let result = PubsubClientSubscription {
686            message_type: PhantomData,
687            operation: "signature",
688            socket,
689            subscription_id,
690            t_cleanup: Some(t_cleanup),
691            exit,
692        };
693
694        Ok((result, receiver))
695    }
696
697    /// Subscribe to slot events.
698    ///
699    /// Receives messages of type [`SlotInfo`] when processing of a slot begins.
700    ///
701    /// # RPC Reference
702    ///
703    /// This method corresponds directly to the [`slotSubscribe`] RPC method.
704    ///
705    /// [`slotSubscribe`]: https://solana.com/docs/rpc/websocket/slotsubscribe
706    pub fn slot_subscribe<R: IntoClientRequest>(
707        request: R,
708    ) -> Result<SlotsSubscription, PubsubClientError> {
709        let client_request = request.into_client_request().map_err(Box::new)?;
710        let socket = connect_with_retry(client_request)?;
711        let (sender, receiver) = unbounded::<SlotInfo>();
712
713        let socket = Arc::new(RwLock::new(socket));
714        let socket_clone = socket.clone();
715        let exit = Arc::new(AtomicBool::new(false));
716        let exit_clone = exit.clone();
717        let body = json!({
718            "jsonrpc":"2.0",
719            "id":1,
720            "method":"slotSubscribe",
721            "params":[]
722        })
723        .to_string();
724        let subscription_id = PubsubSlotClientSubscription::send_subscribe(&socket_clone, body)?;
725
726        let t_cleanup = std::thread::spawn(move || {
727            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
728        });
729
730        let result = PubsubClientSubscription {
731            message_type: PhantomData,
732            operation: "slot",
733            socket,
734            subscription_id,
735            t_cleanup: Some(t_cleanup),
736            exit,
737        };
738
739        Ok((result, receiver))
740    }
741
742    /// Subscribe to slot update events.
743    ///
744    /// Receives messages of type [`SlotUpdate`] when various updates to a slot occur.
745    ///
746    /// Note that this method operates differently than other subscriptions:
747    /// instead of sending the message to a receiver on a channel, it accepts a
748    /// `handler` callback that processes the message directly. This processing
749    /// occurs on another thread.
750    ///
751    /// # RPC Reference
752    ///
753    /// This method corresponds directly to the [`slotUpdatesSubscribe`] RPC method.
754    ///
755    /// [`slotUpdatesSubscribe`]: https://solana.com/docs/rpc/websocket/slotsupdatessubscribe
756    pub fn slot_updates_subscribe<R: IntoClientRequest>(
757        request: R,
758        handler: impl Fn(SlotUpdate) + Send + 'static,
759    ) -> Result<PubsubClientSubscription<SlotUpdate>, PubsubClientError> {
760        let client_request = request.into_client_request().map_err(Box::new)?;
761        let socket = connect_with_retry(client_request)?;
762
763        let socket = Arc::new(RwLock::new(socket));
764        let socket_clone = socket.clone();
765        let exit = Arc::new(AtomicBool::new(false));
766        let exit_clone = exit.clone();
767        let body = json!({
768            "jsonrpc":"2.0",
769            "id":1,
770            "method":"slotsUpdatesSubscribe",
771            "params":[]
772        })
773        .to_string();
774        let subscription_id = PubsubSlotClientSubscription::send_subscribe(&socket, body)?;
775
776        let t_cleanup = std::thread::spawn(move || {
777            Self::cleanup_with_handler(exit_clone, &socket_clone, handler)
778        });
779
780        Ok(PubsubClientSubscription {
781            message_type: PhantomData,
782            operation: "slotsUpdates",
783            socket,
784            subscription_id,
785            t_cleanup: Some(t_cleanup),
786            exit,
787        })
788    }
789
790    fn cleanup_with_sender<T>(
791        exit: Arc<AtomicBool>,
792        socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
793        sender: Sender<T>,
794    ) where
795        T: DeserializeOwned + Send + 'static,
796    {
797        let handler = move |message| match sender.send(message) {
798            Ok(_) => (),
799            Err(err) => {
800                info!("receive error: {err:?}");
801            }
802        };
803        Self::cleanup_with_handler(exit, socket, handler);
804    }
805
806    fn cleanup_with_handler<T, F>(
807        exit: Arc<AtomicBool>,
808        socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
809        handler: F,
810    ) where
811        T: DeserializeOwned,
812        F: Fn(T) + Send + 'static,
813    {
814        loop {
815            if exit.load(Ordering::Relaxed) {
816                break;
817            }
818
819            match PubsubClientSubscription::read_message(socket) {
820                Ok(Some(message)) => handler(message),
821                Ok(None) => {
822                    // Nothing useful, means we received a ping message
823                }
824                Err(err) => {
825                    info!("receive error: {err:?}");
826                    break;
827                }
828            }
829        }
830
831        info!("websocket - exited receive loop");
832    }
833}
834
835#[cfg(test)]
836mod tests {
837    // see client-test/test/client.rs
838}
solana_pubsub_client/pubsub_client.rs

solana_pubsub_client/
pubsub_client.rs
