1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
pub use {
crate::nonblocking::tpu_client::TpuSenderError,
solana_tpu_client::tpu_client::{TpuClientConfig, DEFAULT_FANOUT_SLOTS, MAX_FANOUT_SLOTS},
};
use {
crate::{
connection_cache::ConnectionCache,
nonblocking::tpu_client::TpuClient as NonblockingTpuClient,
},
rayon::iter::{IntoParallelIterator, ParallelIterator},
solana_rpc_client::rpc_client::RpcClient,
solana_sdk::{
message::Message,
signers::Signers,
transaction::{Transaction, TransactionError},
transport::Result as TransportResult,
},
solana_tpu_client::tpu_client::temporary_pub::Result,
std::{net::UdpSocket, sync::Arc},
};
pub struct TpuClient {
_deprecated: UdpSocket, rpc_client: Arc<RpcClient>,
tpu_client: Arc<NonblockingTpuClient>,
}
impl TpuClient {
pub fn send_transaction(&self, transaction: &Transaction) -> bool {
self.invoke(self.tpu_client.send_transaction(transaction))
}
pub fn send_wire_transaction(&self, wire_transaction: Vec<u8>) -> bool {
self.invoke(self.tpu_client.send_wire_transaction(wire_transaction))
}
pub fn try_send_transaction(&self, transaction: &Transaction) -> TransportResult<()> {
self.invoke(self.tpu_client.try_send_transaction(transaction))
}
pub fn try_send_transaction_batch(&self, transactions: &[Transaction]) -> TransportResult<()> {
let wire_transactions = transactions
.into_par_iter()
.map(|tx| bincode::serialize(&tx).expect("serialize Transaction in send_batch"))
.collect::<Vec<_>>();
self.invoke(
self.tpu_client
.try_send_wire_transaction_batch(wire_transactions),
)
}
pub fn try_send_wire_transaction(&self, wire_transaction: Vec<u8>) -> TransportResult<()> {
self.invoke(self.tpu_client.try_send_wire_transaction(wire_transaction))
}
pub fn new(
rpc_client: Arc<RpcClient>,
websocket_url: &str,
config: TpuClientConfig,
) -> Result<Self> {
let create_tpu_client =
NonblockingTpuClient::new(rpc_client.get_inner_client().clone(), websocket_url, config);
let tpu_client =
tokio::task::block_in_place(|| rpc_client.runtime().block_on(create_tpu_client))?;
Ok(Self {
_deprecated: UdpSocket::bind("0.0.0.0:0").unwrap(),
rpc_client,
tpu_client: Arc::new(tpu_client),
})
}
pub fn new_with_connection_cache(
rpc_client: Arc<RpcClient>,
websocket_url: &str,
config: TpuClientConfig,
connection_cache: Arc<ConnectionCache>,
) -> Result<Self> {
let create_tpu_client = NonblockingTpuClient::new_with_connection_cache(
rpc_client.get_inner_client().clone(),
websocket_url,
config,
connection_cache,
);
let tpu_client =
tokio::task::block_in_place(|| rpc_client.runtime().block_on(create_tpu_client))?;
Ok(Self {
_deprecated: UdpSocket::bind("0.0.0.0:0").unwrap(),
rpc_client,
tpu_client: Arc::new(tpu_client),
})
}
pub fn send_and_confirm_messages_with_spinner<T: Signers>(
&self,
messages: &[Message],
signers: &T,
) -> Result<Vec<Option<TransactionError>>> {
self.invoke(
self.tpu_client
.send_and_confirm_messages_with_spinner(messages, signers),
)
}
pub fn rpc_client(&self) -> &RpcClient {
&self.rpc_client
}
fn invoke<T, F: std::future::Future<Output = T>>(&self, f: F) -> T {
tokio::task::block_in_place(move || self.rpc_client.runtime().block_on(f))
}
}