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//! ```no_run
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            && let Some(Number(x)) = json_msg.get("result")
186            && let Some(x) = x.as_u64()
187        {
188            return Ok(x);
189        }
190
191        Err(PubsubClientError::UnexpectedSubscriptionResponse(format!(
192            "msg={message_text}"
193        )))
194    }
195
196    /// Send an unsubscribe message to the server.
197    ///
198    /// Note that this will block as long as the internal subscription receiver
199    /// is waiting on messages from the server, and this can take an unbounded
200    /// amount of time if the server does not send any messages.
201    ///
202    /// If a pubsub client needs to shutdown reliably it should use
203    /// the async client in [`crate::nonblocking::pubsub_client`].
204    pub fn send_unsubscribe(&self) -> Result<(), PubsubClientError> {
205        let method = format!("{}Unsubscribe", self.operation);
206        self.socket
207            .write()
208            .unwrap()
209            .send(Message::Text(
210                json!({
211                "jsonrpc":"2.0","id":1,"method":method,"params":[self.subscription_id]
212                })
213                .to_string()
214                .into(),
215            ))
216            .map_err(Box::new)
217            .map_err(|err| err.into())
218    }
219
220    fn read_message(
221        writable_socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
222    ) -> Result<Option<T>, PubsubClientError> {
223        let message = writable_socket.write().unwrap().read().map_err(Box::new)?;
224        if message.is_ping() {
225            return Ok(None);
226        }
227        let message_text = &message.into_text().map_err(Box::new)?;
228        if let Ok(json_msg) = serde_json::from_str::<Map<String, Value>>(message_text)
229            && let Some(Object(params)) = json_msg.get("params")
230            && let Some(result) = params.get("result")
231            && let Ok(x) = T::deserialize(result)
232        {
233            return Ok(Some(x));
234        }
235
236        Err(PubsubClientError::UnexpectedMessageError(format!(
237            "msg={message_text}"
238        )))
239    }
240
241    /// Shutdown the internel message receiver and wait for its thread to exit.
242    ///
243    /// Note that this will block as long as the subscription receiver is
244    /// waiting on messages from the server, and this can take an unbounded
245    /// amount of time if the server does not send any messages.
246    ///
247    /// If a pubsub client needs to shutdown reliably it should use
248    /// the async client in [`crate::nonblocking::pubsub_client`].
249    pub fn shutdown(&mut self) -> std::thread::Result<()> {
250        if self.t_cleanup.is_some() {
251            info!("websocket thread - shutting down");
252            self.exit.store(true, Ordering::Relaxed);
253            let x = self.t_cleanup.take().unwrap().join();
254            info!("websocket thread - shut down.");
255            x
256        } else {
257            warn!("websocket thread - already shut down.");
258            Ok(())
259        }
260    }
261}
262
263pub type PubsubLogsClientSubscription = PubsubClientSubscription<RpcResponse<RpcLogsResponse>>;
264pub type LogsSubscription = (
265    PubsubLogsClientSubscription,
266    Receiver<RpcResponse<RpcLogsResponse>>,
267);
268
269pub type PubsubSlotClientSubscription = PubsubClientSubscription<SlotInfo>;
270pub type SlotsSubscription = (PubsubSlotClientSubscription, Receiver<SlotInfo>);
271
272pub type PubsubSignatureClientSubscription =
273    PubsubClientSubscription<RpcResponse<RpcSignatureResult>>;
274pub type SignatureSubscription = (
275    PubsubSignatureClientSubscription,
276    Receiver<RpcResponse<RpcSignatureResult>>,
277);
278
279pub type PubsubBlockClientSubscription = PubsubClientSubscription<RpcResponse<RpcBlockUpdate>>;
280pub type BlockSubscription = (
281    PubsubBlockClientSubscription,
282    Receiver<RpcResponse<RpcBlockUpdate>>,
283);
284
285pub type PubsubProgramClientSubscription = PubsubClientSubscription<RpcResponse<RpcKeyedAccount>>;
286pub type ProgramSubscription = (
287    PubsubProgramClientSubscription,
288    Receiver<RpcResponse<RpcKeyedAccount>>,
289);
290
291pub type PubsubAccountClientSubscription = PubsubClientSubscription<RpcResponse<UiAccount>>;
292pub type AccountSubscription = (
293    PubsubAccountClientSubscription,
294    Receiver<RpcResponse<UiAccount>>,
295);
296
297pub type PubsubVoteClientSubscription = PubsubClientSubscription<RpcVote>;
298pub type VoteSubscription = (PubsubVoteClientSubscription, Receiver<RpcVote>);
299
300pub type PubsubRootClientSubscription = PubsubClientSubscription<Slot>;
301pub type RootSubscription = (PubsubRootClientSubscription, Receiver<Slot>);
302
303/// A client for subscribing to messages from the RPC server.
304///
305/// See the [module documentation][self].
306pub struct PubsubClient {}
307
308fn connect_with_retry<R: IntoClientRequest>(
309    request: R,
310) -> Result<WebSocket<MaybeTlsStream<TcpStream>>, Box<tungstenite::Error>> {
311    let mut connection_retries = 5;
312    let client_request = request.into_client_request().map_err(Box::new)?;
313    loop {
314        let result = connect(client_request.clone()).map(|(socket, _)| socket);
315        if let Err(tungstenite::Error::Http(response)) = &result
316            && response.status() == StatusCode::TOO_MANY_REQUESTS
317            && connection_retries > 0
318        {
319            let mut duration = Duration::from_millis(500);
320            if let Some(retry_after) = response.headers().get(header::RETRY_AFTER)
321                && let Ok(retry_after) = retry_after.to_str()
322                && let Ok(retry_after) = retry_after.parse::<u64>()
323                && retry_after < 120
324            {
325                duration = Duration::from_secs(retry_after);
326            }
327
328            connection_retries -= 1;
329            debug!(
330                "Too many requests: server responded with {response:?}, {connection_retries} \
331                 retries left, pausing for {duration:?}"
332            );
333
334            sleep(duration);
335            continue;
336        }
337        return result.map_err(Box::new);
338    }
339}
340
341impl PubsubClient {
342    /// Subscribe to account events.
343    ///
344    /// Receives messages of type [`UiAccount`] when an account's lamports or data changes.
345    ///
346    /// # RPC Reference
347    ///
348    /// This method corresponds directly to the [`accountSubscribe`] RPC method.
349    ///
350    /// [`accountSubscribe`]: https://solana.com/docs/rpc/websocket/accountsubscribe
351    pub fn account_subscribe<R: IntoClientRequest>(
352        request: R,
353        pubkey: &Pubkey,
354        config: Option<RpcAccountInfoConfig>,
355    ) -> Result<AccountSubscription, PubsubClientError> {
356        let client_request = request.into_client_request().map_err(Box::new)?;
357        let socket = connect_with_retry(client_request)?;
358        let (sender, receiver) = unbounded();
359
360        let socket = Arc::new(RwLock::new(socket));
361        let socket_clone = socket.clone();
362        let exit = Arc::new(AtomicBool::new(false));
363        let exit_clone = exit.clone();
364        let body = json!({
365            "jsonrpc":"2.0",
366            "id":1,
367            "method":"accountSubscribe",
368            "params":[
369                pubkey.to_string(),
370                config
371            ]
372        })
373        .to_string();
374        let subscription_id = PubsubAccountClientSubscription::send_subscribe(&socket_clone, body)?;
375
376        let t_cleanup = std::thread::spawn(move || {
377            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
378        });
379
380        let result = PubsubClientSubscription {
381            message_type: PhantomData,
382            operation: "account",
383            socket,
384            subscription_id,
385            t_cleanup: Some(t_cleanup),
386            exit,
387        };
388
389        Ok((result, receiver))
390    }
391
392    /// Subscribe to block events.
393    ///
394    /// Receives messages of type [`RpcBlockUpdate`] when a block is confirmed or finalized.
395    ///
396    /// This method is disabled by default. It can be enabled by passing
397    /// `--rpc-pubsub-enable-block-subscription` to `agave-validator`.
398    ///
399    /// # RPC Reference
400    ///
401    /// This method corresponds directly to the [`blockSubscribe`] RPC method.
402    ///
403    /// [`blockSubscribe`]: https://solana.com/docs/rpc/websocket/blocksubscribe
404    pub fn block_subscribe<R: IntoClientRequest>(
405        request: R,
406        filter: RpcBlockSubscribeFilter,
407        config: Option<RpcBlockSubscribeConfig>,
408    ) -> Result<BlockSubscription, PubsubClientError> {
409        let client_request = request.into_client_request().map_err(Box::new)?;
410        let socket = connect_with_retry(client_request)?;
411        let (sender, receiver) = unbounded();
412
413        let socket = Arc::new(RwLock::new(socket));
414        let socket_clone = socket.clone();
415        let exit = Arc::new(AtomicBool::new(false));
416        let exit_clone = exit.clone();
417        let body = json!({
418            "jsonrpc":"2.0",
419            "id":1,
420            "method":"blockSubscribe",
421            "params":[filter, config]
422        })
423        .to_string();
424
425        let subscription_id = PubsubBlockClientSubscription::send_subscribe(&socket_clone, body)?;
426
427        let t_cleanup = std::thread::spawn(move || {
428            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
429        });
430
431        let result = PubsubClientSubscription {
432            message_type: PhantomData,
433            operation: "block",
434            socket,
435            subscription_id,
436            t_cleanup: Some(t_cleanup),
437            exit,
438        };
439
440        Ok((result, receiver))
441    }
442
443    /// Subscribe to transaction log events.
444    ///
445    /// Receives messages of type [`RpcLogsResponse`] when a transaction is committed.
446    ///
447    /// # RPC Reference
448    ///
449    /// This method corresponds directly to the [`logsSubscribe`] RPC method.
450    ///
451    /// [`logsSubscribe`]: https://solana.com/docs/rpc/websocket/logssubscribe
452    pub fn logs_subscribe<R: IntoClientRequest>(
453        request: R,
454        filter: RpcTransactionLogsFilter,
455        config: RpcTransactionLogsConfig,
456    ) -> Result<LogsSubscription, PubsubClientError> {
457        let client_request = request.into_client_request().map_err(Box::new)?;
458        let socket = connect_with_retry(client_request)?;
459        let (sender, receiver) = unbounded();
460
461        let socket = Arc::new(RwLock::new(socket));
462        let socket_clone = socket.clone();
463        let exit = Arc::new(AtomicBool::new(false));
464        let exit_clone = exit.clone();
465        let body = json!({
466            "jsonrpc":"2.0",
467            "id":1,
468            "method":"logsSubscribe",
469            "params":[filter, config]
470        })
471        .to_string();
472
473        let subscription_id = PubsubLogsClientSubscription::send_subscribe(&socket_clone, body)?;
474
475        let t_cleanup = std::thread::spawn(move || {
476            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
477        });
478
479        let result = PubsubClientSubscription {
480            message_type: PhantomData,
481            operation: "logs",
482            socket,
483            subscription_id,
484            t_cleanup: Some(t_cleanup),
485            exit,
486        };
487
488        Ok((result, receiver))
489    }
490
491    /// Subscribe to program account events.
492    ///
493    /// Receives messages of type [`RpcKeyedAccount`] when an account owned
494    /// by the given program changes.
495    ///
496    /// # RPC Reference
497    ///
498    /// This method corresponds directly to the [`programSubscribe`] RPC method.
499    ///
500    /// [`programSubscribe`]: https://solana.com/docs/rpc/websocket/programsubscribe
501    pub fn program_subscribe<R: IntoClientRequest>(
502        request: R,
503        pubkey: &Pubkey,
504        config: Option<RpcProgramAccountsConfig>,
505    ) -> Result<ProgramSubscription, PubsubClientError> {
506        let client_request = request.into_client_request().map_err(Box::new)?;
507        let socket = connect_with_retry(client_request)?;
508        let (sender, receiver) = unbounded();
509
510        let socket = Arc::new(RwLock::new(socket));
511        let socket_clone = socket.clone();
512        let exit = Arc::new(AtomicBool::new(false));
513        let exit_clone = exit.clone();
514
515        let body = json!({
516            "jsonrpc":"2.0",
517            "id":1,
518            "method":"programSubscribe",
519            "params":[
520                pubkey.to_string(),
521                config
522            ]
523        })
524        .to_string();
525        let subscription_id = PubsubProgramClientSubscription::send_subscribe(&socket_clone, body)?;
526
527        let t_cleanup = std::thread::spawn(move || {
528            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
529        });
530
531        let result = PubsubClientSubscription {
532            message_type: PhantomData,
533            operation: "program",
534            socket,
535            subscription_id,
536            t_cleanup: Some(t_cleanup),
537            exit,
538        };
539
540        Ok((result, receiver))
541    }
542
543    /// Subscribe to vote events.
544    ///
545    /// Receives messages of type [`RpcVote`] when a new vote is observed. These
546    /// votes are observed prior to confirmation and may never be confirmed.
547    ///
548    /// This method is disabled by default. It can be enabled by passing
549    /// `--rpc-pubsub-enable-vote-subscription` to `agave-validator`.
550    ///
551    /// # RPC Reference
552    ///
553    /// This method corresponds directly to the [`voteSubscribe`] RPC method.
554    ///
555    /// [`voteSubscribe`]: https://solana.com/docs/rpc/websocket/votesubscribe
556    pub fn vote_subscribe<R: IntoClientRequest>(
557        request: R,
558    ) -> Result<VoteSubscription, PubsubClientError> {
559        let client_request = request.into_client_request().map_err(Box::new)?;
560        let socket = connect_with_retry(client_request)?;
561        let (sender, receiver) = unbounded();
562
563        let socket = Arc::new(RwLock::new(socket));
564        let socket_clone = socket.clone();
565        let exit = Arc::new(AtomicBool::new(false));
566        let exit_clone = exit.clone();
567        let body = json!({
568            "jsonrpc":"2.0",
569            "id":1,
570            "method":"voteSubscribe",
571        })
572        .to_string();
573        let subscription_id = PubsubVoteClientSubscription::send_subscribe(&socket_clone, body)?;
574
575        let t_cleanup = std::thread::spawn(move || {
576            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
577        });
578
579        let result = PubsubClientSubscription {
580            message_type: PhantomData,
581            operation: "vote",
582            socket,
583            subscription_id,
584            t_cleanup: Some(t_cleanup),
585            exit,
586        };
587
588        Ok((result, receiver))
589    }
590
591    /// Subscribe to root events.
592    ///
593    /// Receives messages of type [`Slot`] when a new [root] is set by the
594    /// validator.
595    ///
596    /// [root]: https://solana.com/docs/terminology#root
597    ///
598    /// # RPC Reference
599    ///
600    /// This method corresponds directly to the [`rootSubscribe`] RPC method.
601    ///
602    /// [`rootSubscribe`]: https://solana.com/docs/rpc/websocket/rootsubscribe
603    pub fn root_subscribe<R: IntoClientRequest>(
604        request: R,
605    ) -> Result<RootSubscription, PubsubClientError> {
606        let client_request = request.into_client_request().map_err(Box::new)?;
607        let socket = connect_with_retry(client_request)?;
608        let (sender, receiver) = unbounded();
609
610        let socket = Arc::new(RwLock::new(socket));
611        let socket_clone = socket.clone();
612        let exit = Arc::new(AtomicBool::new(false));
613        let exit_clone = exit.clone();
614        let body = json!({
615            "jsonrpc":"2.0",
616            "id":1,
617            "method":"rootSubscribe",
618        })
619        .to_string();
620        let subscription_id = PubsubRootClientSubscription::send_subscribe(&socket_clone, body)?;
621
622        let t_cleanup = std::thread::spawn(move || {
623            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
624        });
625
626        let result = PubsubClientSubscription {
627            message_type: PhantomData,
628            operation: "root",
629            socket,
630            subscription_id,
631            t_cleanup: Some(t_cleanup),
632            exit,
633        };
634
635        Ok((result, receiver))
636    }
637
638    /// Subscribe to transaction confirmation events.
639    ///
640    /// Receives messages of type [`RpcSignatureResult`] when a transaction
641    /// with the given signature is committed.
642    ///
643    /// This is a subscription to a single notification. It is automatically
644    /// cancelled by the server once the notification is sent.
645    ///
646    /// # RPC Reference
647    ///
648    /// This method corresponds directly to the [`signatureSubscribe`] RPC method.
649    ///
650    /// [`signatureSubscribe`]: https://solana.com/docs/rpc/websocket/signaturesubscribe
651    pub fn signature_subscribe<R: IntoClientRequest>(
652        request: R,
653        signature: &Signature,
654        config: Option<RpcSignatureSubscribeConfig>,
655    ) -> Result<SignatureSubscription, PubsubClientError> {
656        let client_request = request.into_client_request().map_err(Box::new)?;
657        let socket = connect_with_retry(client_request)?;
658        let (sender, receiver) = unbounded();
659
660        let socket = Arc::new(RwLock::new(socket));
661        let socket_clone = socket.clone();
662        let exit = Arc::new(AtomicBool::new(false));
663        let exit_clone = exit.clone();
664        let body = json!({
665            "jsonrpc":"2.0",
666            "id":1,
667            "method":"signatureSubscribe",
668            "params":[
669                signature.to_string(),
670                config
671            ]
672        })
673        .to_string();
674        let subscription_id =
675            PubsubSignatureClientSubscription::send_subscribe(&socket_clone, body)?;
676
677        let t_cleanup = std::thread::spawn(move || {
678            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
679        });
680
681        let result = PubsubClientSubscription {
682            message_type: PhantomData,
683            operation: "signature",
684            socket,
685            subscription_id,
686            t_cleanup: Some(t_cleanup),
687            exit,
688        };
689
690        Ok((result, receiver))
691    }
692
693    /// Subscribe to slot events.
694    ///
695    /// Receives messages of type [`SlotInfo`] when processing of a slot begins.
696    ///
697    /// # RPC Reference
698    ///
699    /// This method corresponds directly to the [`slotSubscribe`] RPC method.
700    ///
701    /// [`slotSubscribe`]: https://solana.com/docs/rpc/websocket/slotsubscribe
702    pub fn slot_subscribe<R: IntoClientRequest>(
703        request: R,
704    ) -> Result<SlotsSubscription, PubsubClientError> {
705        let client_request = request.into_client_request().map_err(Box::new)?;
706        let socket = connect_with_retry(client_request)?;
707        let (sender, receiver) = unbounded::<SlotInfo>();
708
709        let socket = Arc::new(RwLock::new(socket));
710        let socket_clone = socket.clone();
711        let exit = Arc::new(AtomicBool::new(false));
712        let exit_clone = exit.clone();
713        let body = json!({
714            "jsonrpc":"2.0",
715            "id":1,
716            "method":"slotSubscribe",
717            "params":[]
718        })
719        .to_string();
720        let subscription_id = PubsubSlotClientSubscription::send_subscribe(&socket_clone, body)?;
721
722        let t_cleanup = std::thread::spawn(move || {
723            Self::cleanup_with_sender(exit_clone, &socket_clone, sender)
724        });
725
726        let result = PubsubClientSubscription {
727            message_type: PhantomData,
728            operation: "slot",
729            socket,
730            subscription_id,
731            t_cleanup: Some(t_cleanup),
732            exit,
733        };
734
735        Ok((result, receiver))
736    }
737
738    /// Subscribe to slot update events.
739    ///
740    /// Receives messages of type [`SlotUpdate`] when various updates to a slot occur.
741    ///
742    /// Note that this method operates differently than other subscriptions:
743    /// instead of sending the message to a receiver on a channel, it accepts a
744    /// `handler` callback that processes the message directly. This processing
745    /// occurs on another thread.
746    ///
747    /// # RPC Reference
748    ///
749    /// This method corresponds directly to the [`slotUpdatesSubscribe`] RPC method.
750    ///
751    /// [`slotUpdatesSubscribe`]: https://solana.com/docs/rpc/websocket/slotsupdatessubscribe
752    pub fn slot_updates_subscribe<R: IntoClientRequest>(
753        request: R,
754        handler: impl Fn(SlotUpdate) + Send + 'static,
755    ) -> Result<PubsubClientSubscription<SlotUpdate>, PubsubClientError> {
756        let client_request = request.into_client_request().map_err(Box::new)?;
757        let socket = connect_with_retry(client_request)?;
758
759        let socket = Arc::new(RwLock::new(socket));
760        let socket_clone = socket.clone();
761        let exit = Arc::new(AtomicBool::new(false));
762        let exit_clone = exit.clone();
763        let body = json!({
764            "jsonrpc":"2.0",
765            "id":1,
766            "method":"slotsUpdatesSubscribe",
767            "params":[]
768        })
769        .to_string();
770        let subscription_id = PubsubSlotClientSubscription::send_subscribe(&socket, body)?;
771
772        let t_cleanup = std::thread::spawn(move || {
773            Self::cleanup_with_handler(exit_clone, &socket_clone, handler)
774        });
775
776        Ok(PubsubClientSubscription {
777            message_type: PhantomData,
778            operation: "slotsUpdates",
779            socket,
780            subscription_id,
781            t_cleanup: Some(t_cleanup),
782            exit,
783        })
784    }
785
786    fn cleanup_with_sender<T>(
787        exit: Arc<AtomicBool>,
788        socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
789        sender: Sender<T>,
790    ) where
791        T: DeserializeOwned + Send + 'static,
792    {
793        let handler = move |message| match sender.send(message) {
794            Ok(_) => (),
795            Err(err) => {
796                info!("receive error: {err:?}");
797            }
798        };
799        Self::cleanup_with_handler(exit, socket, handler);
800    }
801
802    fn cleanup_with_handler<T, F>(
803        exit: Arc<AtomicBool>,
804        socket: &Arc<RwLock<WebSocket<MaybeTlsStream<TcpStream>>>>,
805        handler: F,
806    ) where
807        T: DeserializeOwned,
808        F: Fn(T) + Send + 'static,
809    {
810        loop {
811            if exit.load(Ordering::Relaxed) {
812                break;
813            }
814
815            match PubsubClientSubscription::read_message(socket) {
816                Ok(Some(message)) => handler(message),
817                Ok(None) => {
818                    // Nothing useful, means we received a ping message
819                }
820                Err(err) => {
821                    info!("receive error: {err:?}");
822                    break;
823                }
824            }
825        }
826
827        info!("websocket - exited receive loop");
828    }
829}
830
831#[cfg(test)]
832mod tests {
833    // see client-test/test/client.rs
834}
solana_pubsub_client/pubsub_client.rs

solana_pubsub_client/
pubsub_client.rs
