soruh/centralex
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

cleanup

Browse Source
This commit is contained in:
soruh 2023-11-21 13:10:18 +01:00
parent db138f5123
commit 5484140688
4 changed files with 26 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/client.rs
View File

@ -161,12 +161,12 @@ async fn idle(
break Ok(Some(IdleResult::Disconnect { packet })) break Ok(Some(IdleResult::Disconnect { packet }))
} }
}, },
_ = sleep(send_next_ping_in) => { () = sleep(send_next_ping_in) => {
trace!("sending ping"); trace!("sending ping");
writer.write_all(bytemuck::bytes_of(& Header { kind: PacketKind::Ping.raw(), length: 0 })).await?; writer.write_all(bytemuck::bytes_of(& Header { kind: PacketKind::Ping.raw(), length: 0 })).await?;
last_ping_sent_at = Instant::now(); last_ping_sent_at = Instant::now();
} }
_ = sleep(next_ping_expected_in) => { () = sleep(next_ping_expected_in) => {
writer.write_all(REJECT_TIMEOUT).await?; writer.write_all(REJECT_TIMEOUT).await?;
break Ok(None); break Ok(None);
@ -332,7 +332,8 @@ pub async fn handler(
let mut packet = Packet::default(); let mut packet = Packet::default();
let Ok(res) = timeout(AUTH_TIMEOUT, packet.recv_into_cancelation_safe(&mut reader)).await else { let Ok(res) = timeout(AUTH_TIMEOUT, packet.recv_into_cancelation_safe(&mut reader)).await
else {
writer.write_all(REJECT_TIMEOUT).await?; writer.write_all(REJECT_TIMEOUT).await?;
return Ok(()); return Ok(());
}; };
@ -373,18 +374,21 @@ pub async fn handler(
packet.data.clear(); packet.data.clear();
packet.send(&mut writer).await?; packet.send(&mut writer).await?;
let Some(idle_result) = idle( let Some(idle_result) = idle(listener, packet, &mut reader, &mut writer).await? else {
listener,
packet,
&mut reader,
&mut writer,
).await? else {
return Ok(()); return Ok(());
}; };
let Some((mut caller, mut packet)) = notify_or_disconnect(idle_result, handler_metadata, port_handler, port, &mut writer).await? else { let Some((mut caller, mut packet)) = notify_or_disconnect(
idle_result,
handler_metadata,
port_handler,
port,
&mut writer,
)
.await?
else {
return Ok(()); return Ok(());
}; };
let notify_at = Instant::now(); let notify_at = Instant::now();
@ -395,9 +399,9 @@ pub async fn handler(
); );
let Ok(res) = recv.await else { let Ok(res) = recv.await else {
writer.write_all(REJECT_TIMEOUT).await?; writer.write_all(REJECT_TIMEOUT).await?;
return Ok(()); return Ok(());
}; };
res?; res?;
match packet.kind() { match packet.kind() {

4
src/http.rs
View File

@ -34,7 +34,7 @@ impl<T: Send + 'static, Fut: Future<Output = T> + Send + 'static> Executor<Fut>
const COMPRESSED_HTML: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/minified.html.gz")); const COMPRESSED_HTML: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/minified.html.gz"));
async fn index(req: &Request<Body>) -> Result<Response<Body>, hyper::http::Error> { fn index(req: &Request<Body>) -> Result<Response<Body>, hyper::http::Error> {
let response = Response::builder(); let response = Response::builder();
let accepts_gzip = req let accepts_gzip = req
@ -160,7 +160,7 @@ pub async fn debug_server(
async move { async move {
match (req.method(), req.uri().path()) { match (req.method(), req.uri().path()) {
(&Method::GET, "/") => index(&req).await, (&Method::GET, "/") => index(&req),
(&Method::GET, "/data") => data(&req, port_handler).await, (&Method::GET, "/data") => data(&req, port_handler).await,
(&Method::GET, "/events") => events(&req, change_receiver), (&Method::GET, "/events") => events(&req, change_receiver),
_ => Response::builder() _ => Response::builder()

2
src/main.rs
View File

@ -337,7 +337,7 @@ async fn tokio_main(config: Arc<Config>) -> eyre::Result<()> {
let (change_sender, change_receiver) = tokio::sync::watch::channel(std::time::Instant::now()); let (change_sender, change_receiver) = tokio::sync::watch::channel(std::time::Instant::now());
let mut port_handler = PortHandler::load_or_default(&cache_path, change_sender); let mut port_handler = PortHandler::load_or_default(&cache_path, change_sender);
port_handler.update_allowed_ports(&config.allowed_ports); port_handler.update_allowed_ports(config.allowed_ports.clone());
let port_handler = Arc::new(Mutex::new(port_handler)); let port_handler = Arc::new(Mutex::new(port_handler));

8
src/ports.rs
View File

@ -173,6 +173,8 @@ impl AllowedList {
} }
impl PortHandler { impl PortHandler {
/// # Panics
/// If the the `change_sender` could not be notified
pub fn register_update(&mut self) { pub fn register_update(&mut self) {
let now = std::time::Instant::now(); let now = std::time::Instant::now();
self.last_update = Some(now); self.last_update = Some(now);
@ -242,10 +244,10 @@ impl PortHandler {
this this
} }
pub fn update_allowed_ports(&mut self, allowed_ports: &AllowedList) { pub fn update_allowed_ports(&mut self, allowed_ports: AllowedList) {
self.register_update(); self.register_update();
self.allowed_ports = allowed_ports.clone(); self.allowed_ports = allowed_ports;
self.free_ports.clear(); // remove all ports self.free_ports.clear(); // remove all ports
self.free_ports self.free_ports
@ -338,7 +340,7 @@ impl Debug for Rejector {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Rejector") f.debug_struct("Rejector")
.field("message", &self.state.1) .field("message", &self.state.1)
.finish() .finish_non_exhaustive()
} }
} }