solana_client/
tpu_client.rs

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

solana_client/
tpu_client.rs
