Refactor connection module into submodules

src/connection/dispatch.rs

@@ -0,0 +1,34 @@
+//! Event dispatching for the connection actor.
+
+use super::{ConnectionActor, event::Event, state::ActorState};
+use crate::{
+    app::Packet,
+    correlation::CorrelatableFrame,
+    push::FrameLike,
+    response::WireframeError,
+};
+
+impl<F, E> ConnectionActor<F, E>
+where
+    F: FrameLike + CorrelatableFrame + Packet,
+    E: std::fmt::Debug,
+{
+    /// Dispatch the given event to the appropriate handler.
+    pub(super) fn dispatch_event(
+        &mut self,
+        event: Event<F, E>,
+        state: &mut ActorState,
+        out: &mut Vec<F>,
+    ) -> Result<(), WireframeError<E>> {
+        match event {
+            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::MultiPacket(res) => self.process_multi_packet(res, state, out),
+            Event::Response(res) => self.process_response(res, state, out)?,
+            Event::Idle => {}
+        }
+
+        Ok(())
+    }
+}

src/connection/drain.rs

@@ -0,0 +1,183 @@
+//! Queue drain operations and fairness-aware helpers.
+
+use tokio::sync::mpsc::error::TryRecvError;
+
+use super::{
+    ConnectionActor,
+    DrainContext,
+    QueueKind,
+    multi_packet::MultiPacketTerminationReason,
+    state::ActorState,
+};
+use crate::{app::Packet, correlation::CorrelatableFrame, push::FrameLike};
+
+impl<F, E> ConnectionActor<F, E>
+where
+    F: FrameLike + CorrelatableFrame + Packet,
+    E: std::fmt::Debug,
+{
+    /// Handle the result of polling the high-priority queue.
+    pub(super) fn process_high(
+        &mut self,
+        res: Option<F>,
+        state: &mut ActorState,
+        out: &mut Vec<F>,
+    ) {
+        self.process_queue(QueueKind::High, res, DrainContext { out, state });
+    }
+
+    /// Process a queue-backed source with shared low-priority semantics.
+    pub(super) fn process_queue(
+        &mut self,
+        kind: QueueKind,
+        res: Option<F>,
+        ctx: DrainContext<'_, F>,
+    ) {
+        let DrainContext { out, state } = ctx;
+        match res {
+            Some(frame) => {
+                match kind {
+                    QueueKind::Multi if self.multi_packet.is_stamping_enabled() => {
+                        self.emit_multi_packet_frame(frame, out);
+                    }
+                    _ => {
+                        self.process_frame_with_hooks_and_metrics(frame, out);
+                    }
+                }
+                match kind {
+                    QueueKind::High => self.after_high(out, state),
+                    QueueKind::Low | QueueKind::Multi => self.after_low(),
+                }
+            }
+            None => match kind {
+                QueueKind::High => {
+                    Self::handle_closed_receiver(&mut self.high_rx, state);
+                    self.fairness.reset();
+                }
+                QueueKind::Low => {
+                    Self::handle_closed_receiver(&mut self.low_rx, state);
+                }
+                QueueKind::Multi => {
+                    self.handle_multi_packet_closed(
+                        MultiPacketTerminationReason::Drained,
+                        state,
+                        out,
+                    );
+                }
+            },
+        }
+    }
+
+    /// Handle the result of polling the low-priority queue.
+    pub(super) fn process_low(&mut self, res: Option<F>, state: &mut ActorState, out: &mut Vec<F>) {
+        self.process_queue(QueueKind::Low, res, DrainContext { out, state });
+    }
+
+    /// Handle frames drained from the multi-packet channel.
+    pub(super) fn process_multi_packet(
+        &mut self,
+        res: Option<F>,
+        state: &mut ActorState,
+        out: &mut Vec<F>,
+    ) {
+        self.process_queue(QueueKind::Multi, res, DrainContext { out, state });
+    }
+
+    /// Update counters and opportunistically drain the low-priority and multi-packet queues.
+    pub(super) fn after_high(&mut self, out: &mut Vec<F>, state: &mut ActorState) {
+        self.fairness.record_high_priority();
+
+        if !self.fairness.should_yield_to_low_priority() {
+            return;
+        }
+
+        if self.try_opportunistic_drain(
+            QueueKind::Low,
+            DrainContext {
+                out: &mut *out,
+                state: &mut *state,
+            },
+        ) {
+            return;
+        }
+
+        let _ = self.try_opportunistic_drain(
+            QueueKind::Multi,
+            DrainContext {
+                out: &mut *out,
+                state: &mut *state,
+            },
+        );
+    }
+
+    /// Try to opportunistically drain a queue-backed source when fairness allows.
+    ///
+    /// Returns `true` when a frame is forwarded to `out`.
+    pub(super) fn try_opportunistic_drain(
+        &mut self,
+        kind: QueueKind,
+        ctx: DrainContext<'_, F>,
+    ) -> bool {
+        let DrainContext { out, state } = ctx;
+        match kind {
+            QueueKind::High => {
+                debug_assert!(false, "try_opportunistic_drain(High) should not be called");
+                false
+            }
+            QueueKind::Low => {
+                let res = match self.low_rx.as_mut() {
+                    Some(receiver) => receiver.try_recv(),
+                    None => return false,
+                };
+
+                match res {
+                    Ok(frame) => {
+                        self.process_frame_with_hooks_and_metrics(frame, out);
+                        self.after_low();
+                        true
+                    }
+                    Err(TryRecvError::Empty) => false,
+                    Err(TryRecvError::Disconnected) => {
+                        Self::handle_closed_receiver(&mut self.low_rx, state);
+                        false
+                    }
+                }
+            }
+            QueueKind::Multi => {
+                let result = match self.multi_packet.channel_mut() {
+                    Some(rx) => rx.try_recv(),
+                    None => return false,
+                };
+
+                match result {
+                    Ok(frame) => {
+                        self.emit_multi_packet_frame(frame, out);
+                        self.after_low();
+                        true
+                    }
+                    Err(TryRecvError::Empty) => false,
+                    Err(TryRecvError::Disconnected) => {
+                        self.handle_multi_packet_closed(
+                            MultiPacketTerminationReason::Disconnected,
+                            state,
+                            out,
+                        );
+                        false
+                    }
+                }
+            }
+        }
+    }
+
+    /// Reset counters after processing a low-priority frame.
+    pub(super) fn after_low(&mut self) { self.fairness.reset(); }
+
+    /// Common logic for handling closed receivers.
+    pub(super) fn handle_closed_receiver(
+        receiver: &mut Option<tokio::sync::mpsc::Receiver<F>>,
+        state: &mut ActorState,
+    ) {
+        *receiver = None;
+        state.mark_closed();
+    }
+}

src/connection/event.rs

@@ -0,0 +1,22 @@
+//! Internal event types for the connection actor select loop.
+
+use crate::response::WireframeError;
+
+/// Events returned by [`ConnectionActor::next_event`][super::ConnectionActor::next_event].
+///
+/// Only `Debug` is derived because `WireframeError<E>` does not implement
+/// `Clone` or `PartialEq`.
+#[derive(Debug)]
+pub(super) enum Event<F, E> {
+    Shutdown,
+    High(Option<F>),
+    Low(Option<F>),
+    /// Frames drained from the multi-packet response channel.
+    /// Frames are forwarded in channel order after low-priority queues to
+    /// preserve fairness and reuse the existing back-pressure.
+    /// The actor emits the protocol terminator when the sender closes the
+    /// channel so downstream observers see end-of-stream signalling.
+    MultiPacket(Option<F>),
+    Response(Option<Result<F, WireframeError<E>>>),
+    Idle,
+}

src/connection/frame.rs

@@ -0,0 +1,88 @@
+//! Frame processing and emission helpers.
+
+use log::warn;
+
+use super::ConnectionActor;
+use crate::{
+    app::{Packet, fragment_utils::fragment_packet},
+    correlation::CorrelatableFrame,
+    push::FrameLike,
+};
+
+impl<F, E> ConnectionActor<F, E>
+where
+    F: FrameLike + CorrelatableFrame + Packet,
+    E: std::fmt::Debug,
+{
+    /// Emit a multi-packet frame with correlation stamping applied.
+    pub(super) fn emit_multi_packet_frame(&mut self, frame: F, out: &mut Vec<F>) {
+        let mut frame = frame;
+        self.apply_multi_packet_correlation(&mut frame);
+        self.process_frame_with_hooks_and_metrics(frame, out);
+    }
+
+    /// Apply correlation stamping to a multi-packet frame.
+    pub(super) fn apply_multi_packet_correlation(&mut self, frame: &mut F) {
+        if !self.multi_packet.is_stamping_enabled() {
+            // No channel is active, so there is nothing to stamp.
+            return;
+        }
+
+        let correlation_id = self.multi_packet.correlation_id();
+        frame.set_correlation_id(correlation_id);
+
+        if let Some(expected) = correlation_id {
+            debug_assert!(
+                CorrelatableFrame::correlation_id(frame) == Some(expected)
+                    || CorrelatableFrame::correlation_id(frame).is_none(),
+                "multi-packet frame correlation mismatch: expected={:?}, got={:?}",
+                Some(expected),
+                CorrelatableFrame::correlation_id(frame),
+            );
+        } else {
+            debug_assert!(
+                CorrelatableFrame::correlation_id(frame).is_none(),
+                "multi-packet frame correlation unexpectedly present: got={:?}",
+                CorrelatableFrame::correlation_id(frame),
+            );
+        }
+    }
+
+    /// Apply protocol hooks and increment metrics before emitting a frame.
+    ///
+    /// # Examples
+    ///
+    /// ```ignore
+    /// actor.process_frame_with_hooks_and_metrics(frame, &mut out);
+    /// ```
+    pub(super) fn process_frame_with_hooks_and_metrics(&mut self, frame: F, out: &mut Vec<F>)
+    where
+        F: Packet,
+    {
+        if let Some(fragmenter) = self.fragmenter.as_deref() {
+            let fragmented = fragment_packet(fragmenter, frame);
+            match fragmented {
+                Ok(frames) => frames
+                    .into_iter()
+                    .for_each(|frame| self.push_frame(frame, out)),
+                Err(err) => {
+                    warn!(
+                        "failed to fragment frame: connection_id={:?}, peer={:?}, error={err:?}",
+                        self.connection_id, self.peer_addr,
+                    );
+                    crate::metrics::inc_handler_errors();
+                }
+            }
+        } else {
+            self.push_frame(frame, out);
+        }
+    }
+
+    /// Push a single frame to output after applying hooks and metrics.
+    pub(super) fn push_frame(&mut self, frame: F, out: &mut Vec<F>) {
+        let mut frame = frame;
+        self.hooks.before_send(&mut frame, &mut self.ctx);
+        out.push(frame);
+        crate::metrics::inc_frames(crate::metrics::Direction::Outbound);
+    }
+}