From 303f3235448e1acd004f93ccb0a2651b36053a27 Mon Sep 17 00:00:00 2001 From: Leynos Date: Thu, 31 Jul 2025 22:31:53 +0100 Subject: [PATCH 1/2] Document PushHandleInner --- src/push.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/push.rs b/src/push.rs index 01103745..5145ff3a 100644 --- a/src/push.rs +++ b/src/push.rs @@ -85,6 +85,11 @@ impl std::fmt::Display for PushConfigError { impl std::error::Error for PushConfigError {} +/// Shared state for [`PushHandle`] clones. +/// +/// Holds the per-priority send channels, optional rate limiter and +/// optional dead letter queue. Wrapped in an [`Arc`] so handles can be +/// cloned cheaply. pub(crate) struct PushHandleInner { high_prio_tx: mpsc::Sender, low_prio_tx: mpsc::Sender, From cf6aa11069d08fcda900af397e3ea481bf5c2f7b Mon Sep 17 00:00:00 2001 From: Leynos Date: Sat, 2 Aug 2025 09:54:10 +0100 Subject: [PATCH 2/2] Refactor push processing helper --- examples/metadata_routing.rs | 2 +- src/connection.rs | 129 ++++++++++++++++++++++++----------- src/push.rs | 16 +++-- src/server.rs | 14 ++-- 4 files changed, 108 insertions(+), 53 deletions(-) diff --git a/examples/metadata_routing.rs b/examples/metadata_routing.rs index bc326c51..946938a6 100644 --- a/examples/metadata_routing.rs +++ b/examples/metadata_routing.rs @@ -60,7 +60,7 @@ impl FrameMetadata for HeaderSerializer { struct Ping; #[derive(bincode::Decode, bincode::Encode)] -#[expect(dead_code, reason = "used only in documentation example")] +#[allow(dead_code)] struct Pong; #[tokio::main] diff --git a/src/connection.rs b/src/connection.rs index 6e8c737e..35a7b526 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -43,7 +43,9 @@ impl Drop for ActiveConnection { /// Return the current number of active connections. #[must_use] -pub fn active_connection_count() -> u64 { ACTIVE_CONNECTIONS.load(Ordering::Relaxed) } +pub fn active_connection_count() -> u64 { + ACTIVE_CONNECTIONS.load(Ordering::Relaxed) +} use crate::{ hooks::{ConnectionContext, ProtocolHooks}, @@ -117,6 +119,18 @@ pub struct ConnectionActor { peer_addr: Option, } +/// Context for processing frames during actor execution. +struct ProcessContext<'a, F> { + state: &'a mut ActorState, + out: &'a mut Vec, +} + +impl<'a, F> ProcessContext<'a, F> { + fn new(state: &'a mut ActorState, out: &'a mut Vec) -> Self { + Self { state, out } + } +} + impl ConnectionActor where F: FrameLike, @@ -188,14 +202,20 @@ where } /// Replace the fairness configuration. - pub fn set_fairness(&mut self, fairness: FairnessConfig) { self.fairness = fairness; } + pub fn set_fairness(&mut self, fairness: FairnessConfig) { + self.fairness = fairness; + } /// Set or replace the current streaming response. - pub fn set_response(&mut self, stream: Option>) { self.response = stream; } + pub fn set_response(&mut self, stream: Option>) { + self.response = stream; + } /// Get a clone of the shutdown token used by the actor. #[must_use] - pub fn shutdown_token(&self) -> CancellationToken { self.shutdown.clone() } + pub fn shutdown_token(&self) -> CancellationToken { + self.shutdown.clone() + } /// Drive the actor until all sources are exhausted or shutdown is triggered. /// @@ -283,11 +303,12 @@ where state: &mut ActorState, out: &mut Vec, ) -> Result<(), WireframeError> { - match self.next_event(state).await { - Event::Shutdown => self.process_shutdown(state), - Event::High(res) => self.process_high(res, state, out), - Event::Low(res) => self.process_low(res, state, out), - Event::Response(res) => self.process_response(res, state, out)?, + let mut ctx = ProcessContext::new(state, out); + match self.next_event(ctx.state).await { + Event::Shutdown => self.process_shutdown(ctx.state), + Event::High(res) => self.process_high(res, &mut ctx), + Event::Low(res) => self.process_low(res, &mut ctx), + Event::Response(res) => self.process_response(res, ctx.state, ctx.out)?, Event::Idle => {} } @@ -301,23 +322,39 @@ where } /// Handle the result of polling the high-priority queue. - fn process_high(&mut self, res: Option, state: &mut ActorState, out: &mut Vec) { - if let Some(frame) = res { - self.process_frame_common(frame, out); - self.after_high(out, state); - } else { - Self::handle_closed_receiver(&mut self.high_rx, state); - self.reset_high_counter(); - } + fn process_high(&mut self, res: Option, ctx: &mut ProcessContext) { + self.process_push(res, ctx, Self::after_high, |this, state| { + Self::handle_closed_receiver(&mut this.high_rx, state); + this.reset_high_counter(); + }); } /// Handle the result of polling the low-priority queue. - fn process_low(&mut self, res: Option, state: &mut ActorState, out: &mut Vec) { + fn process_low(&mut self, res: Option, ctx: &mut ProcessContext) { + self.process_push( + res, + ctx, + |this, _, _| this.after_low(), + |this, state| Self::handle_closed_receiver(&mut this.low_rx, state), + ); + } + + /// Helper to process push queue results with queue-specific callbacks. + fn process_push( + &mut self, + res: Option, + ctx: &mut ProcessContext, + on_some: OnSome, + on_none: OnNone, + ) where + OnSome: FnOnce(&mut Self, &mut Vec, &mut ActorState), + OnNone: FnOnce(&mut Self, &mut ActorState), + { if let Some(frame) = res { - self.process_frame_common(frame, out); - self.after_low(); + self.process_frame_common(frame, ctx.out); + on_some(self, ctx.out, ctx.state); } else { - Self::handle_closed_receiver(&mut self.low_rx, state); + on_none(self, ctx.state); } } @@ -374,19 +411,19 @@ where self.high_start = Some(Instant::now()); } - if self.should_yield_to_low_priority() - && let Some(rx) = &mut self.low_rx - { - match rx.try_recv() { - Ok(mut frame) => { - self.hooks.before_send(&mut frame, &mut self.ctx); - out.push(frame); - self.after_low(); - } - Err(mpsc::error::TryRecvError::Empty) => {} - Err(mpsc::error::TryRecvError::Disconnected) => { - self.low_rx = None; - state.mark_closed(); + if self.should_yield_to_low_priority() { + if let Some(rx) = &mut self.low_rx { + match rx.try_recv() { + Ok(mut frame) => { + self.hooks.before_send(&mut frame, &mut self.ctx); + out.push(frame); + self.after_low(); + } + Err(mpsc::error::TryRecvError::Empty) => {} + Err(mpsc::error::TryRecvError::Disconnected) => { + self.low_rx = None; + state.mark_closed(); + } } } } @@ -405,7 +442,9 @@ where } /// Reset counters after processing a low-priority frame. - fn after_low(&mut self) { self.reset_high_counter(); } + fn after_low(&mut self) { + self.reset_high_counter(); + } /// Clear the burst counter and associated timestamp. fn reset_high_counter(&mut self) { @@ -448,11 +487,15 @@ where /// Await cancellation on the provided shutdown token. #[inline] - async fn wait_shutdown(token: CancellationToken) { token.cancelled_owned().await; } + async fn wait_shutdown(token: CancellationToken) { + token.cancelled_owned().await; + } /// Receive the next frame from a push queue. #[inline] - async fn recv_push(rx: &mut mpsc::Receiver) -> Option { rx.recv().await } + async fn recv_push(rx: &mut mpsc::Receiver) -> Option { + rx.recv().await + } /// Poll `f` if `opt` is `Some`, returning `None` otherwise. #[expect( @@ -535,11 +578,17 @@ impl ActorState { } /// Returns `true` while the actor is actively processing sources. - fn is_active(&self) -> bool { matches!(self.run_state, RunState::Active) } + fn is_active(&self) -> bool { + matches!(self.run_state, RunState::Active) + } /// Returns `true` once shutdown has begun. - fn is_shutting_down(&self) -> bool { matches!(self.run_state, RunState::ShuttingDown) } + fn is_shutting_down(&self) -> bool { + matches!(self.run_state, RunState::ShuttingDown) + } /// Returns `true` when all sources have finished. - fn is_done(&self) -> bool { matches!(self.run_state, RunState::Finished) } + fn is_done(&self) -> bool { + matches!(self.run_state, RunState::Finished) + } } diff --git a/src/push.rs b/src/push.rs index 5145ff3a..c6266a7d 100644 --- a/src/push.rs +++ b/src/push.rs @@ -102,7 +102,9 @@ pub(crate) struct PushHandleInner { pub struct PushHandle(Arc>); impl PushHandle { - pub(crate) fn from_arc(arc: Arc>) -> Self { Self(arc) } + pub(crate) fn from_arc(arc: Arc>) -> Self { + Self(arc) + } /// Internal helper to push a frame with the requested priority. /// @@ -258,7 +260,9 @@ impl PushHandle { } /// Downgrade to a `Weak` reference for storage in a registry. - pub(crate) fn downgrade(&self) -> Weak> { Arc::downgrade(&self.0) } + pub(crate) fn downgrade(&self) -> Weak> { + Arc::downgrade(&self.0) + } } /// Receiver ends of the push queues stored by the connection actor. @@ -387,10 +391,10 @@ impl PushQueues { rate: Option, dlq: Option>, ) -> Result<(Self, PushHandle), PushConfigError> { - if let Some(r) = rate - && (r == 0 || r > MAX_PUSH_RATE) - { - return Err(PushConfigError::InvalidRate(r)); + if let Some(r) = rate { + if r == 0 || r > MAX_PUSH_RATE { + return Err(PushConfigError::InvalidRate(r)); + } } let (high_tx, high_rx) = mpsc::channel(high_capacity); let (low_tx, low_rx) = mpsc::channel(low_capacity); diff --git a/src/server.rs b/src/server.rs index f1592a19..f6b39ebf 100644 --- a/src/server.rs +++ b/src/server.rs @@ -229,7 +229,9 @@ where /// ``` #[inline] #[must_use] - pub const fn worker_count(&self) -> usize { self.workers } + pub const fn worker_count(&self) -> usize { + self.workers + } /// Get the socket address the server is bound to, if available. #[must_use] @@ -469,10 +471,10 @@ async fn process_stream( { match read_preamble::<_, T>(&mut stream).await { Ok((preamble, leftover)) => { - if let Some(handler) = on_success.as_ref() - && let Err(e) = handler(&preamble, &mut stream).await - { - eprintln!("preamble callback error: {e}"); + if let Some(handler) = on_success.as_ref() { + if let Err(e) = handler(&preamble, &mut stream).await { + eprintln!("preamble callback error: {e}"); + } } let stream = RewindStream::new(leftover, stream); // Hand the connection to the application for processing. @@ -520,7 +522,7 @@ mod tests { /// Test helper preamble carrying no data. #[derive(Debug, Clone, PartialEq, Encode, Decode)] - #[expect(dead_code, reason = "test helper for unused preamble type")] + #[allow(dead_code)] struct EmptyPreamble; #[fixture]