Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit d1840e3

Browse files
hungfnthungfnt
committedMay 2, 2024··
refactor: [#681] udp return errors instead of panicking
1 parent 9a127f1 commit d1840e3

File tree

5 files changed

+189
-174
lines changed

5 files changed

+189
-174
lines changed
 

‎http_trackers.sh

-29
This file was deleted.

‎src/console/clients/udp/checker.rs

+8-8
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ impl Client {
6464
let binding_address = local_bind_to.parse().context("binding local address")?;
6565

6666
debug!("Binding to: {local_bind_to}");
67-
let udp_client = UdpClient::bind(&local_bind_to).await;
67+
let udp_client = UdpClient::bind(&local_bind_to).await?;
6868

6969
let bound_to = udp_client.socket.local_addr().context("bound local address")?;
7070
debug!("Bound to: {bound_to}");
@@ -88,7 +88,7 @@ impl Client {
8888

8989
match &self.udp_tracker_client {
9090
Some(client) => {
91-
client.udp_client.connect(&tracker_socket_addr.to_string()).await;
91+
client.udp_client.connect(&tracker_socket_addr.to_string()).await?;
9292
self.remote_socket = Some(*tracker_socket_addr);
9393
Ok(())
9494
}
@@ -116,9 +116,9 @@ impl Client {
116116

117117
match &self.udp_tracker_client {
118118
Some(client) => {
119-
client.send(connect_request.into()).await;
119+
client.send(connect_request.into()).await?;
120120

121-
let response = client.receive().await;
121+
let response = client.receive().await?;
122122

123123
debug!("connection request response:\n{response:#?}");
124124

@@ -163,9 +163,9 @@ impl Client {
163163

164164
match &self.udp_tracker_client {
165165
Some(client) => {
166-
client.send(announce_request.into()).await;
166+
client.send(announce_request.into()).await?;
167167

168-
let response = client.receive().await;
168+
let response = client.receive().await?;
169169

170170
debug!("announce request response:\n{response:#?}");
171171

@@ -200,9 +200,9 @@ impl Client {
200200

201201
match &self.udp_tracker_client {
202202
Some(client) => {
203-
client.send(scrape_request.into()).await;
203+
client.send(scrape_request.into()).await?;
204204

205-
let response = client.receive().await;
205+
let response = client.receive().await?;
206206

207207
debug!("scrape request response:\n{response:#?}");
208208

‎src/shared/bit_torrent/tracker/udp/client.rs

+124-94
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,17 @@
1+
use core::result::Result::{Err, Ok};
12
use std::io::Cursor;
23
use std::net::SocketAddr;
34
use std::sync::Arc;
45
use std::time::Duration;
56

6-
use crate::shared::bit_torrent::tracker::udp::{source_address, MAX_PACKET_SIZE};
7-
use anyhow::anyhow;
8-
use anyhow::{Context, Result};
9-
use core::result::Result::{Ok, Err};
10-
use anyhow::Error as AError;
7+
use anyhow::{anyhow, Context, Result};
118
use aquatic_udp_protocol::{ConnectRequest, Request, Response, TransactionId};
129
use log::debug;
1310
use tokio::net::UdpSocket;
1411
use tokio::time;
1512

13+
use crate::shared::bit_torrent::tracker::udp::{source_address, MAX_PACKET_SIZE};
14+
1615
/// Default timeout for sending and receiving packets. And waiting for sockets
1716
/// to be readable and writable.
1817
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
@@ -28,107 +27,120 @@
2827
}
2928

3029
impl UdpClient {
31-
/// # Panics
30+
/// # Errors
3231
///
33-
/// Will panic if the local address can't be bound.
32+
/// Will return error if the local address can't be bound.
3433
///
3534
pub async fn bind(local_address: &str) -> Result<Self> {
36-
let socket_addr = local_address.parse::<SocketAddr>().map_err(|err| err).context("{local_address} is not a valid socket address")?;
35+
let socket_addr = local_address
36+
.parse::<SocketAddr>()
37+
.context(format!("{local_address} is not a valid socket address"))?;
38+
3739
let socket = UdpSocket::bind(socket_addr).await?;
40+
3841
let udp_client = Self {
3942
socket: Arc::new(socket),
4043
timeout: DEFAULT_TIMEOUT,
4144
};
4245
Ok(udp_client)
4346
}
4447

45-
/// # Panics
48+
/// # Errors
4649
///
47-
/// Will panic if can't connect to the socket.
48-
pub async fn connect(&self, remote_address: &str) -> anyhow::Result<()> {
49-
let socket_addr = remote_address.parse::<SocketAddr>().map_err(|err| err).context(format!("{} is not a valid socket address", remote_address))?;
50-
self.socket.connect(socket_addr).await.map_err(|err| err)?;
51-
Ok(())
50+
/// Will return error if can't connect to the socket.
51+
pub async fn connect(&self, remote_address: &str) -> Result<()> {
52+
let socket_addr = remote_address
53+
.parse::<SocketAddr>()
54+
.context(format!("{remote_address} is not a valid socket address"))?;
55+
56+
match self.socket.connect(socket_addr).await {
57+
Ok(()) => {
58+
debug!("Connected successfully");
59+
Ok(())
60+
}
61+
Err(e) => Err(anyhow!("Failed to connect: {e:?}")),
62+
}
5263
}
5364

54-
/// # Panics
65+
/// # Errors
5566
///
56-
/// Will panic if:
67+
/// Will return error if:
5768
///
5869
/// - Can't write to the socket.
5970
/// - Can't send data.
60-
pub async fn send(&self, bytes: &[u8]) -> Result<usize, anyhow::Error> {
71+
pub async fn send(&self, bytes: &[u8]) -> Result<usize> {
6172
debug!(target: "UDP client", "sending {bytes:?} ...");
6273

63-
let _:Result<(), anyhow::Error> = match time::timeout(self.timeout, self.socket.writable()).await {
74+
match time::timeout(self.timeout, self.socket.writable()).await {
6475
Ok(writable_result) => {
65-
let writable_result_status : Result<(), anyhow::Error> = match writable_result {
66-
Ok(()) => Ok(()),
67-
Err(e) => Err(anyhow!("IO error waiting for the socket to become readable: {e:?}"))
76+
match writable_result {
77+
Ok(()) => (),
78+
Err(e) => return Err(anyhow!("IO error waiting for the socket to become readable: {e:?}")),
6879
};
69-
writable_result_status
7080
}
71-
Err(e) => Err(anyhow!("Timeout waiting for the socket to become readable: {e:?}"))
81+
Err(e) => return Err(anyhow!("Timeout waiting for the socket to become readable: {e:?}")),
7282
};
7383

74-
let send_status:Result<usize, anyhow::Error> = match time::timeout(self.timeout, self.socket.send(bytes)).await {
75-
Ok(send_result) => {
76-
let send_result_status: Result<usize, anyhow::Error> = match send_result {
77-
Ok(size) => Ok(size),
78-
Err(e) => Err(anyhow!("IO error waiting for the socket to become readable: {}", e))
79-
};
80-
send_result_status
81-
}
82-
Err(e) => Err(anyhow!("IO error waiting for the socket to become readable: {}", e))
83-
};
84-
send_status
84+
match time::timeout(self.timeout, self.socket.send(bytes)).await {
85+
Ok(send_result) => match send_result {
86+
Ok(size) => Ok(size),
87+
Err(e) => Err(anyhow!("IO error during send: {e:?}")),
88+
},
89+
Err(e) => Err(anyhow!("Send operation timed out: {e:?}")),
90+
}
8591
}
8692

87-
/// # Panics
93+
/// # Errors
8894
///
89-
/// Will panic if:
95+
/// Will return error if:
9096
///
9197
/// - Can't read from the socket.
9298
/// - Can't receive data.
99+
///
100+
/// # Panics
101+
///
93102
pub async fn receive(&self, bytes: &mut [u8]) -> Result<usize> {
94103
debug!(target: "UDP client", "receiving ...");
95104

96-
let _ :Result<(), anyhow::Error>= match time::timeout(self.timeout, self.socket.readable()).await {
105+
match time::timeout(self.timeout, self.socket.readable()).await {
97106
Ok(readable_result) => {
98-
let readable_result_status = match readable_result {
99-
Ok(()) => Ok(()),
100-
Err(e) => Err(anyhow!("IO error waiting for the socket to become readable: {e:?}")),
107+
match readable_result {
108+
Ok(()) => (),
109+
Err(e) => return Err(anyhow!("IO error waiting for the socket to become readable: {e:?}")),
101110
};
102-
readable_result_status
103-
},
104-
Err(e) => Err(anyhow!("Timeout waiting for the socket to become readable: {e:?}")),
111+
}
112+
Err(e) => return Err(anyhow!("Timeout waiting for the socket to become readable: {e:?}")),
105113
};
106114

107-
let size: Result<usize, anyhow::Error> = match time::timeout(self.timeout, self.socket.recv(bytes)).await {
115+
let size_result = match time::timeout(self.timeout, self.socket.recv(bytes)).await {
108116
Ok(recv_result) => match recv_result {
109117
Ok(size) => Ok(size),
110118
Err(e) => Err(anyhow!("IO error during send: {e:?}")),
111119
},
112120
Err(e) => Err(anyhow!("Receive operation timed out: {e:?}")),
113121
};
114122

115-
debug!(target: "UDP client", "{size} bytes received {bytes:?}");
116-
117-
size
123+
if size_result.is_ok() {
124+
let size = size_result.as_ref().unwrap();
125+
debug!(target: "UDP client", "{size} bytes received {bytes:?}");
126+
size_result
127+
} else {
128+
size_result
129+
}
118130
}
119-
131+
}
120132

121133
/// Creates a new `UdpClient` connected to a Udp server
134+
///
135+
/// # Errors
136+
///
137+
/// Will return any errors present in the call stack
138+
///
122139
pub async fn new_udp_client_connected(remote_address: &str) -> Result<UdpClient> {
123140
let port = 0; // Let OS choose an unused port.
124-
match UdpClient::bind(&source_address(port)).await {
125-
Ok(client) => {
126-
client.connect(remote_address).await;
127-
Ok(client)
128-
}
129-
Err(err) => Err(err),
130-
}
131-
}
141+
let client = UdpClient::bind(&source_address(port)).await?;
142+
client.connect(remote_address).await?;
143+
Ok(client)
132144
}
133145

134146
#[allow(clippy::module_name_repetitions)]
@@ -138,85 +150,103 @@
138150
}
139151

140152
impl UdpTrackerClient {
141-
/// # Panics
153+
/// # Errors
142154
///
143-
/// Will panic if can't write request to bytes.
155+
/// Will return error if can't write request to bytes.
144156
pub async fn send(&self, request: Request) -> Result<usize> {
145157
debug!(target: "UDP tracker client", "send request {request:?}");
146158

147159
// Write request into a buffer
148160
let request_buffer = vec![0u8; MAX_PACKET_SIZE];
149161
let mut cursor = Cursor::new(request_buffer);
150162

151-
let request_data = match request.write(&mut cursor) {
163+
let request_data_result = match request.write(&mut cursor) {
152164
Ok(()) => {
153165
#[allow(clippy::cast_possible_truncation)]
154166
let position = cursor.position() as usize;
155167
let inner_request_buffer = cursor.get_ref();
156168
// Return slice which contains written request data
157-
&inner_request_buffer[..position]
169+
Ok(&inner_request_buffer[..position])
158170
}
159171
Err(e) => Err(anyhow!("could not write request to bytes: {e}.")),
160172
};
161173

174+
let request_data = request_data_result?;
175+
162176
self.udp_client.send(request_data).await
163177
}
164178

165-
/// # Panics
179+
/// # Errors
166180
///
167-
/// Will panic if can't create response from the received payload (bytes buffer).
168-
pub async fn receive(&self) -> Response {
181+
/// Will return error if can't create response from the received payload (bytes buffer).
182+
pub async fn receive(&self) -> Result<Response> {
169183
let mut response_buffer = [0u8; MAX_PACKET_SIZE];
170184

171-
let payload_size = self.udp_client.receive(&mut response_buffer).await;
185+
let payload_size = self.udp_client.receive(&mut response_buffer).await?;
172186

173187
debug!(target: "UDP tracker client", "received {payload_size} bytes. Response {response_buffer:?}");
174188

175-
Response::from_bytes(&response_buffer[..payload_size], true).unwrap()
189+
let response = Response::from_bytes(&response_buffer[..payload_size], true)?;
190+
191+
Ok(response)
176192
}
177193
}
178194

179195
/// Creates a new `UdpTrackerClient` connected to a Udp Tracker server
180-
pub async fn new_udp_tracker_client_connected(remote_address: &str) -> UdpTrackerClient {
196+
///
197+
/// # Errors
198+
///
199+
/// Will return any errors present in the call stack
200+
///
201+
pub async fn new_udp_tracker_client_connected(remote_address: &str) -> Result<UdpTrackerClient> {
181202
let udp_client = new_udp_client_connected(remote_address).await?;
182-
UdpTrackerClient { udp_client.unwrap() }
203+
let udp_tracker_client = UdpTrackerClient { udp_client };
204+
Ok(udp_tracker_client)
183205
}
184206

185207
/// Helper Function to Check if a UDP Service is Connectable
186208
///
187-
/// # Errors
209+
/// # Panics
188210
///
189211
/// It will return an error if unable to connect to the UDP service.
190212
///
191-
/// # Panics
213+
/// # Errors
214+
///
192215
pub async fn check(binding: &SocketAddr) -> Result<String, String> {
193216
debug!("Checking Service (detail): {binding:?}.");
194217

195-
let client = new_udp_tracker_client_connected(binding.to_string().as_str()).await;
196-
197-
let connect_request = ConnectRequest {
198-
transaction_id: TransactionId(123),
199-
};
200-
201-
client.send(connect_request.into()).await;
202-
203-
let process = move |response| {
204-
if matches!(response, Response::Connect(_connect_response)) {
205-
Ok("Connected".to_string())
206-
} else {
207-
Error("Did not Connect".to_string())
208-
}
209-
};
210-
211-
let sleep = time::sleep(Duration::from_millis(2000));
212-
tokio::pin!(sleep);
213-
214-
tokio::select! {
215-
() = &mut sleep => {
216-
Error("Timed Out".to_string())
217-
}
218-
response = client.receive() => {
219-
process(response)
218+
match new_udp_tracker_client_connected(binding.to_string().as_str()).await {
219+
Ok(client) => {
220+
let connect_request = ConnectRequest {
221+
transaction_id: TransactionId(123),
222+
};
223+
224+
// client.send() return usize, but doesn't use here
225+
match client.send(connect_request.into()).await {
226+
Ok(_) => (),
227+
Err(e) => debug!("Error: {e:?}."),
228+
};
229+
230+
let process = move |response| {
231+
if matches!(response, Response::Connect(_connect_response)) {
232+
Ok("Connected".to_string())
233+
} else {
234+
Err("Did not Connect".to_string())
235+
}
236+
};
237+
238+
let sleep = time::sleep(Duration::from_millis(2000));
239+
tokio::pin!(sleep);
240+
241+
tokio::select! {
242+
() = &mut sleep => {
243+
Err("Timed Out".to_string())
244+
}
245+
response = client.receive() => {
246+
process(response.unwrap())
247+
}
248+
}
220249
}
250+
Err(e) => Err(format!("{e:?}")),
221251
}
222252
}

‎tests/servers/udp/contract.rs

+57-14
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,15 @@ fn empty_buffer() -> [u8; MAX_PACKET_SIZE] {
2424
async fn send_connection_request(transaction_id: TransactionId, client: &UdpTrackerClient) -> ConnectionId {
2525
let connect_request = ConnectRequest { transaction_id };
2626

27-
client.send(connect_request.into()).await;
27+
match client.send(connect_request.into()).await {
28+
Ok(_) => (),
29+
Err(err) => panic!("{err}"),
30+
};
2831

29-
let response = client.receive().await;
32+
let response = match client.receive().await {
33+
Ok(response) => response,
34+
Err(err) => panic!("{err}"),
35+
};
3036

3137
match response {
3238
Response::Connect(connect_response) => connect_response.connection_id,
@@ -38,12 +44,22 @@ async fn send_connection_request(transaction_id: TransactionId, client: &UdpTrac
3844
async fn should_return_a_bad_request_response_when_the_client_sends_an_empty_request() {
3945
let env = Started::new(&configuration::ephemeral().into()).await;
4046

41-
let client = new_udp_client_connected(&env.bind_address().to_string()).await;
47+
let client = match new_udp_client_connected(&env.bind_address().to_string()).await {
48+
Ok(udp_client) => udp_client,
49+
Err(err) => panic!("{err}"),
50+
};
4251

43-
client.send(&empty_udp_request()).await;
52+
match client.send(&empty_udp_request()).await {
53+
Ok(_) => (),
54+
Err(err) => panic!("{err}"),
55+
};
4456

4557
let mut buffer = empty_buffer();
46-
client.receive(&mut buffer).await;
58+
match client.receive(&mut buffer).await {
59+
Ok(_) => (),
60+
Err(err) => panic!("{err}"),
61+
};
62+
4763
let response = Response::from_bytes(&buffer, true).unwrap();
4864

4965
assert!(is_error_response(&response, "bad request"));
@@ -63,15 +79,24 @@ mod receiving_a_connection_request {
6379
async fn should_return_a_connect_response() {
6480
let env = Started::new(&configuration::ephemeral().into()).await;
6581

66-
let client = new_udp_tracker_client_connected(&env.bind_address().to_string()).await;
82+
let client = match new_udp_tracker_client_connected(&env.bind_address().to_string()).await {
83+
Ok(udp_tracker_client) => udp_tracker_client,
84+
Err(err) => panic!("{err}"),
85+
};
6786

6887
let connect_request = ConnectRequest {
6988
transaction_id: TransactionId(123),
7089
};
7190

72-
client.send(connect_request.into()).await;
91+
match client.send(connect_request.into()).await {
92+
Ok(_) => (),
93+
Err(err) => panic!("{err}"),
94+
};
7395

74-
let response = client.receive().await;
96+
let response = match client.receive().await {
97+
Ok(response) => response,
98+
Err(err) => panic!("{err}"),
99+
};
75100

76101
assert!(is_connect_response(&response, TransactionId(123)));
77102

@@ -97,7 +122,10 @@ mod receiving_an_announce_request {
97122
async fn should_return_an_announce_response() {
98123
let env = Started::new(&configuration::ephemeral().into()).await;
99124

100-
let client = new_udp_tracker_client_connected(&env.bind_address().to_string()).await;
125+
let client = match new_udp_tracker_client_connected(&env.bind_address().to_string()).await {
126+
Ok(udp_tracker_client) => udp_tracker_client,
127+
Err(err) => panic!("{err}"),
128+
};
101129

102130
let connection_id = send_connection_request(TransactionId(123), &client).await;
103131

@@ -118,9 +146,15 @@ mod receiving_an_announce_request {
118146
port: Port(client.udp_client.socket.local_addr().unwrap().port()),
119147
};
120148

121-
client.send(announce_request.into()).await;
149+
match client.send(announce_request.into()).await {
150+
Ok(_) => (),
151+
Err(err) => panic!("{err}"),
152+
};
122153

123-
let response = client.receive().await;
154+
let response = match client.receive().await {
155+
Ok(response) => response,
156+
Err(err) => panic!("{err}"),
157+
};
124158

125159
println!("test response {response:?}");
126160

@@ -143,7 +177,10 @@ mod receiving_an_scrape_request {
143177
async fn should_return_a_scrape_response() {
144178
let env = Started::new(&configuration::ephemeral().into()).await;
145179

146-
let client = new_udp_tracker_client_connected(&env.bind_address().to_string()).await;
180+
let client = match new_udp_tracker_client_connected(&env.bind_address().to_string()).await {
181+
Ok(udp_tracker_client) => udp_tracker_client,
182+
Err(err) => panic!("{err}"),
183+
};
147184

148185
let connection_id = send_connection_request(TransactionId(123), &client).await;
149186

@@ -159,9 +196,15 @@ mod receiving_an_scrape_request {
159196
info_hashes,
160197
};
161198

162-
client.send(scrape_request.into()).await;
199+
match client.send(scrape_request.into()).await {
200+
Ok(_) => (),
201+
Err(err) => panic!("{err}"),
202+
};
163203

164-
let response = client.receive().await;
204+
let response = match client.receive().await {
205+
Ok(response) => response,
206+
Err(err) => panic!("{err}"),
207+
};
165208

166209
assert!(is_scrape_response(&response));
167210

‎trackers.txt

-29
This file was deleted.

0 commit comments

Comments
 (0)
Please sign in to comment.