Refactor ConnectionActor queue handling

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, reason = "used only in documentation example")]
 struct Pong;
 
 #[tokio::main]

examples/ping_pong.rs

@@ -40,7 +40,10 @@ const PING_ID: u32 = 1;
 ///
 /// The middleware chain generates the actual response, so this
 /// handler intentionally performs no work.
-#[allow(clippy::unused_async)]
+#[allow(
+    clippy::unused_async,
+    reason = "example handler intentionally performs no work"
+)]
 async fn ping_handler() {}
 
 struct PongMiddleware;

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},
@@ -65,6 +67,18 @@ enum Event<F, E> {
     Idle,
 }
 
+/// Context for processing frames during actor execution.
+struct ProcessContext<'a, F> {
+    state: &'a mut ActorState,
+    out: &'a mut Vec<F>,
+}
+
+impl<'a, F> ProcessContext<'a, F> {
+    fn new(state: &'a mut ActorState, out: &'a mut Vec<F>) -> Self {
+        Self { state, out }
+    }
+}
+
 /// Configuration controlling fairness when draining push queues.
 #[derive(Clone, Copy)]
 pub struct FairnessConfig {
@@ -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<FrameStream<F, E>>) { self.response = stream; }
+    pub fn set_response(&mut self, stream: Option<FrameStream<F, E>>) {
+        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,13 @@ where
         state: &mut ActorState,
         out: &mut Vec<F>,
     ) -> Result<(), WireframeError<E>> {
-        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 +323,40 @@ where
     }
 
     /// Handle the result of polling the high-priority queue.
-    fn process_high(&mut self, res: Option<F>, state: &mut ActorState, out: &mut Vec<F>) {
-        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<F>, ctx: &mut ProcessContext<F>) {
+        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<F>, state: &mut ActorState, out: &mut Vec<F>) {
-        if let Some(frame) = res {
-            self.process_frame_common(frame, out);
-            self.after_low();
-        } else {
-            Self::handle_closed_receiver(&mut self.low_rx, state);
+    fn process_low(&mut self, res: Option<F>, ctx: &mut ProcessContext<F>) {
+        self.process_push(
+            res,
+            ctx,
+            |this, _, _| this.after_low(),
+            |this, state| Self::handle_closed_receiver(&mut this.low_rx, state),
+        );
+    }
+
+    /// Handle the result of polling a push queue.
+    fn process_push<OnSome, OnNone>(
+        &mut self,
+        res: Option<F>,
+        ctx: &mut ProcessContext<F>,
+        on_some: OnSome,
+        on_none: OnNone,
+    ) where
+        OnSome: FnOnce(&mut Self, &mut Vec<F>, &mut ActorState),
+        OnNone: FnOnce(&mut Self, &mut ActorState),
+    {
+        match res {
+            Some(frame) => {
+                self.process_frame_common(frame, ctx.out);
+                on_some(self, ctx.out, ctx.state);
+            }
+            None => on_none(self, ctx.state),
         }
     }
 
@@ -374,19 +413,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 +444,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 +489,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<F>) -> Option<F> { rx.recv().await }
+    async fn recv_push(rx: &mut mpsc::Receiver<F>) -> Option<F> {
+        rx.recv().await
+    }
 
     /// Poll `f` if `opt` is `Some`, returning `None` otherwise.
     #[expect(
@@ -535,11 +580,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)
+    }
 }

src/middleware.rs

@@ -17,19 +17,27 @@ pub struct FrameContainer<F> {
 impl<F> FrameContainer<F> {
     /// Create a new container holding `frame` bytes.
     #[must_use]
-    pub fn new(frame: F) -> Self { Self { frame } }
+    pub fn new(frame: F) -> Self {
+        Self { frame }
+    }
 
     /// Borrow the inner frame data.
     #[must_use]
-    pub fn frame(&self) -> &F { &self.frame }
+    pub fn frame(&self) -> &F {
+        &self.frame
+    }
 
     /// Mutable access to the frame data.
     #[must_use]
-    pub fn frame_mut(&mut self) -> &mut F { &mut self.frame }
+    pub fn frame_mut(&mut self) -> &mut F {
+        &mut self.frame
+    }
 
     /// Consume the container, returning the frame.
     #[must_use]
-    pub fn into_inner(self) -> F { self.frame }
+    pub fn into_inner(self) -> F {
+        self.frame
+    }
 }
 
 /// Incoming request wrapper passed through middleware.
@@ -49,15 +57,21 @@ impl ServiceRequest {
 
     /// Borrow the underlying frame bytes.
     #[must_use]
-    pub fn frame(&self) -> &[u8] { self.inner.frame().as_slice() }
+    pub fn frame(&self) -> &[u8] {
+        self.inner.frame().as_slice()
+    }
 
     /// Mutable access to the inner frame bytes.
     #[must_use]
-    pub fn frame_mut(&mut self) -> &mut Vec<u8> { self.inner.frame_mut() }
+    pub fn frame_mut(&mut self) -> &mut Vec<u8> {
+        self.inner.frame_mut()
+    }
 
     /// Consume the request, returning the inner frame bytes.
     #[must_use]
-    pub fn into_inner(self) -> Vec<u8> { self.inner.into_inner() }
+    pub fn into_inner(self) -> Vec<u8> {
+        self.inner.into_inner()
+    }
 }
 
 /// Response produced by a handler or middleware.
@@ -77,15 +91,21 @@ impl ServiceResponse {
 
     /// Borrow the inner frame bytes.
     #[must_use]
-    pub fn frame(&self) -> &[u8] { self.inner.frame().as_slice() }
+    pub fn frame(&self) -> &[u8] {
+        self.inner.frame().as_slice()
+    }
 
     /// Mutable access to the response frame bytes.
     #[must_use]
-    pub fn frame_mut(&mut self) -> &mut Vec<u8> { self.inner.frame_mut() }
+    pub fn frame_mut(&mut self) -> &mut Vec<u8> {
+        self.inner.frame_mut()
+    }
 
     /// Consume the response, yielding the raw frame bytes.
     #[must_use]
-    pub fn into_inner(self) -> Vec<u8> { self.inner.into_inner() }
+    pub fn into_inner(self) -> Vec<u8> {
+        self.inner.into_inner()
+    }
 }
 
 /// Continuation used by middleware to call the next service in the chain.
@@ -120,7 +140,9 @@ where
     /// ```
     #[inline]
     #[must_use]
-    pub fn new(service: &'a S) -> Self { Self { service } }
+    pub fn new(service: &'a S) -> Self {
+        Self { service }
+    }
 
     /// Call the next service with the provided request.
     ///
@@ -157,7 +179,11 @@ where
 
     /// Create a new middleware service wrapping `service`.
     #[inline]
-    #[allow(clippy::inline_fn_without_body, unused_attributes)]
+    #[allow(
+        clippy::inline_fn_without_body,
+        unused_attributes,
+        reason = "future-proof attribute and inline hint without body"
+    )]
     #[must_use = "use the returned middleware service"]
     async fn transform(&self, service: S) -> Self::Output;
 }
@@ -173,7 +199,9 @@ pub struct FromFn<F> {
 
 impl<F> FromFn<F> {
     /// Construct middleware from the provided asynchronous function.
-    pub fn new(f: F) -> Self { Self { f } }
+    pub fn new(f: F) -> Self {
+        Self { f }
+    }
 }
 
 /// Convenience constructor to build middleware from an async function.
@@ -202,7 +230,9 @@ impl<F> FromFn<F> {
 /// # }
 /// let mw = from_fn(logging);
 /// ```
-pub fn from_fn<F>(f: F) -> FromFn<F> { FromFn::new(f) }
+pub fn from_fn<F>(f: F) -> FromFn<F> {
+    FromFn::new(f)
+}
 
 /// Service wrapper that applies a middleware function to requests.
 ///
@@ -283,7 +313,9 @@ impl<E: Packet> HandlerService<E> {
 
     /// Returns the route identifier associated with this service.
     #[must_use]
-    pub const fn id(&self) -> u32 { self.id }
+    pub const fn id(&self) -> u32 {
+        self.id
+    }
 }
 
 struct RouteService<E: Packet> {