solana_client/nonblocking/
tpu_client.rs

1pub use solana_tpu_client::nonblocking::tpu_client::{LeaderTpuService, TpuSenderError};
2use {
3    crate::{connection_cache::ConnectionCache, tpu_client::TpuClientConfig},
4    solana_connection_cache::connection_cache::{
5        ConnectionCache as BackendConnectionCache, ConnectionManager, ConnectionPool,
6        NewConnectionConfig,
7    },
8    solana_message::Message,
9    solana_quic_client::{QuicConfig, QuicConnectionManager, QuicPool},
10    solana_rpc_client::nonblocking::rpc_client::RpcClient,
11    solana_signer::signers::Signers,
12    solana_tpu_client::nonblocking::tpu_client::{Result, TpuClient as BackendTpuClient},
13    solana_transaction::{Transaction, versioned::VersionedTransaction},
14    solana_transaction_error::{TransactionError, TransportResult},
15    std::sync::Arc,
16};
17
18/// Client which sends transactions directly to the current leader's TPU port over UDP.
19/// The client uses RPC to determine the current leader and fetch node contact info
20pub struct TpuClient<
21    P, // ConnectionPool
22    M, // ConnectionManager
23    C, // NewConnectionConfig
24> {
25    tpu_client: BackendTpuClient<P, M, C>,
26}
27
28impl<P, M, C> TpuClient<P, M, C>
29where
30    P: ConnectionPool<NewConnectionConfig = C>,
31    M: ConnectionManager<ConnectionPool = P, NewConnectionConfig = C>,
32    C: NewConnectionConfig,
33{
34    /// Serialize and send transaction to the current and upcoming leader TPUs according to fanout
35    /// size
36    pub async fn send_transaction(&self, transaction: &Transaction) -> bool {
37        self.tpu_client.send_transaction(transaction).await
38    }
39
40    /// Send a wire transaction to the current and upcoming leader TPUs according to fanout size
41    pub async fn send_wire_transaction(&self, wire_transaction: Vec<u8>) -> bool {
42        self.tpu_client
43            .send_wire_transaction(wire_transaction)
44            .await
45    }
46
47    /// Serialize and send transaction to the current and upcoming leader TPUs according to fanout
48    /// size
49    /// Returns the last error if all sends fail
50    pub async fn try_send_transaction(
51        &self,
52        transaction: &VersionedTransaction,
53    ) -> TransportResult<()> {
54        self.tpu_client.try_send_transaction(transaction).await
55    }
56
57    /// Send a wire transaction to the current and upcoming leader TPUs according to fanout size
58    /// Returns the last error if all sends fail
59    pub async fn try_send_wire_transaction(
60        &self,
61        wire_transaction: Vec<u8>,
62    ) -> TransportResult<()> {
63        self.tpu_client
64            .try_send_wire_transaction(wire_transaction)
65            .await
66    }
67
68    /// Send a batch of wire transactions to the current and upcoming leader TPUs according to
69    /// fanout size
70    /// Returns the last error if all sends fail
71    pub async fn try_send_wire_transaction_batch(
72        &self,
73        wire_transactions: Vec<Vec<u8>>,
74    ) -> TransportResult<()> {
75        self.tpu_client
76            .try_send_wire_transaction_batch(wire_transactions)
77            .await
78    }
79}
80
81impl TpuClient<QuicPool, QuicConnectionManager, QuicConfig> {
82    /// Create a new client that disconnects when dropped
83    pub async fn new(
84        name: &'static str,
85        rpc_client: Arc<RpcClient>,
86        websocket_url: &str,
87        config: TpuClientConfig,
88    ) -> Result<Self> {
89        let connection_cache = match ConnectionCache::new(name) {
90            ConnectionCache::Quic(cache) => cache,
91            ConnectionCache::Udp(_) => {
92                return Err(TpuSenderError::Custom(String::from(
93                    "Invalid default connection cache",
94                )));
95            }
96        };
97        Self::new_with_connection_cache(rpc_client, websocket_url, config, connection_cache).await
98    }
99}
100
101impl<P, M, C> TpuClient<P, M, C>
102where
103    P: ConnectionPool<NewConnectionConfig = C>,
104    M: ConnectionManager<ConnectionPool = P, NewConnectionConfig = C>,
105    C: NewConnectionConfig,
106{
107    /// Create a new client that disconnects when dropped
108    pub async fn new_with_connection_cache(
109        rpc_client: Arc<RpcClient>,
110        websocket_url: &str,
111        config: TpuClientConfig,
112        connection_cache: Arc<BackendConnectionCache<P, M, C>>,
113    ) -> Result<Self> {
114        Ok(Self {
115            tpu_client: BackendTpuClient::new_with_connection_cache(
116                rpc_client,
117                websocket_url,
118                config,
119                connection_cache,
120            )
121            .await?,
122        })
123    }
124
125    pub async fn send_and_confirm_messages_with_spinner<T: Signers + ?Sized>(
126        &self,
127        messages: &[Message],
128        signers: &T,
129    ) -> Result<Vec<Option<TransactionError>>> {
130        self.tpu_client
131            .send_and_confirm_messages_with_spinner(messages, signers)
132            .await
133    }
134
135    pub fn rpc_client(&self) -> &RpcClient {
136        self.tpu_client.rpc_client()
137    }
138
139    pub async fn shutdown(&mut self) {
140        self.tpu_client.shutdown().await
141    }
142}
solana_client/nonblocking/tpu_client.rs

solana_client/nonblocking/
tpu_client.rs
