YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-04-01 18:25:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

aquatic: run_event_loop: extract inner logic to own function

Browse source
This commit is contained in:
Joakim Frostegård 2020-04-06 19:55:30 +02:00
parent 8426e8f3f7
commit ad1fa5b833
1 changed files with 97 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

170
aquatic/src/lib/network.rs
View file

@ -81,86 +81,110 @@ pub fn run_event_loop(
if token.0 == token_num { if token.0 == token_num {
if event.is_readable(){ if event.is_readable(){
loop { handle_readable_socket(
match socket.recv_from(&mut buffer) { &state,
Ok((amt, src)) => { &mut socket,
let request = request_from_bytes( &mut buffer,
&buffer[..amt], &mut responses,
255u8 // FIXME &mut connect_requests,
); &mut announce_requests,
&mut scrape_requests
);
match request { poll.registry()
Ok(Request::Connect(r)) => { .reregister(&mut socket, token, interests)
connect_requests.push((r, src)); .unwrap();
}, }
Ok(Request::Announce(r)) => { }
announce_requests.push((r, src)); }
}, }
Ok(Request::Scrape(r)) => { }
scrape_requests.push((r, src));
},
Ok(Request::Invalid(r)) => {
let response = Response::Error(ErrorResponse {
transaction_id: r.transaction_id,
message: "Invalid request".to_string(),
});
responses.push((response, src));
}, /// Read requests, generate and send back responses
Err(err) => { fn handle_readable_socket(
eprintln!("Request parse error: {:?}", err); state: &State,
}, socket: &mut UdpSocket,
} buffer: &mut [u8],
}, responses: &mut Vec<(Response, SocketAddr)>,
Err(err) => { connect_requests: &mut Vec<(ConnectRequest, SocketAddr)>,
match err.kind() { announce_requests: &mut Vec<(AnnounceRequest, SocketAddr)>,
ErrorKind::WouldBlock => { scrape_requests: &mut Vec<(ScrapeRequest, SocketAddr)>,
break; ){
}, loop {
err => { match socket.recv_from(buffer) {
eprintln!("recv_from error: {:?}", err); Ok((amt, src)) => {
} let request = request_from_bytes(
} &buffer[..amt],
} 255u8 // FIXME
} );
match request {
Ok(Request::Connect(r)) => {
connect_requests.push((r, src));
},
Ok(Request::Announce(r)) => {
announce_requests.push((r, src));
},
Ok(Request::Scrape(r)) => {
scrape_requests.push((r, src));
},
Ok(Request::Invalid(r)) => {
let response = Response::Error(ErrorResponse {
transaction_id: r.transaction_id,
message: "Invalid request".to_string(),
});
responses.push((response, src));
},
Err(err) => {
eprintln!("request_from_bytes error: {:?}", err);
},
}
},
Err(err) => {
match err.kind() {
ErrorKind::WouldBlock => {
break;
},
err => {
eprintln!("recv_from error: {:?}", err);
} }
}
}
}
}
handle_connect_requests( handle_connect_requests(
&state, state,
&mut responses, responses,
connect_requests.drain(..) connect_requests.drain(..)
); );
handle_announce_requests( handle_announce_requests(
&state, state,
&mut responses, responses,
announce_requests.drain(..), announce_requests.drain(..),
); );
handle_scrape_requests( handle_scrape_requests(
&state, state,
&mut responses, responses,
scrape_requests.drain(..), scrape_requests.drain(..),
); );
for (response, src) in responses.drain(..) { for (response, src) in responses.drain(..) {
let bytes = response_to_bytes(&response, IpVersion::IPv4); let bytes = response_to_bytes(&response, IpVersion::IPv4);
match socket.send_to(&bytes[..], src){ match socket.send_to(&bytes[..], src){
Ok(_bytes_sent) => { Ok(_bytes_sent) => {
}, },
Err(err) => { Err(err) => {
match err.kind(){ match err.kind(){
ErrorKind::WouldBlock => { ErrorKind::WouldBlock => {
break; break;
}, },
err => { err => {
eprintln!("send_to error: {:?}", err); eprintln!("send_to error: {:?}", err);
}
}
}
}
} }
poll.registry().reregister(&mut socket, token, interests).unwrap();
} }
} }
} }