pipecrab_runtime/
inbound.rs

1//! Each stage has an Inbound mailbox with two typed lanes:
2//! `sys` — the priority lane, drains first, carries `(Direction, SystemFrame)`.
3//! `data` — the data lane, carries bare `DataFrame` (downstream only).
4//!
5//! Keeping the lanes typed prevents misrouting a media frame onto the system
6//! lane and removes the per-frame is-system check from the hot path.
7
8use futures::channel::mpsc::Receiver;
9use futures::stream::StreamExt;
10use pipecrab_core::{DataFrame, Direction, SystemFrame};
11
12/// A frame received from [`Inbound::recv`]: either a system frame (with its
13/// travel direction) or a data frame (always downstream).
14#[derive(Debug)]
15pub enum Received {
16    /// A system frame and the direction it is travelling.
17    Sys(Direction, SystemFrame),
18    /// A data frame, implicitly travelling downstream.
19    Data(DataFrame),
20}
21
22/// The receive surface of a stage: a preempting system lane and the data lane.
23///
24/// Within a lane, frames keep FIFO order. Across lanes, `sys` always wins, so a
25/// system frame is taken even when `data` is backed up.
26pub struct Inbound {
27    /// System-tier frames (lifecycle, interruption, errors). Drained first.
28    /// `Error` rides this lane *upstream*; `Interrupt`/`Start`/`Stop` ride it
29    /// downstream. Sparse and latency-critical.
30    pub sys: Receiver<(Direction, SystemFrame)>,
31    /// Data-tier frames (media, transcripts), in FIFO order, downstream only.
32    pub data: Receiver<DataFrame>,
33}
34
35impl Inbound {
36    /// Receive the next frame, draining the system lane before the data lane.
37    ///
38    /// Returns [`Received::Sys`] or [`Received::Data`], or `None` once *both*
39    /// lanes are closed — the run-loop's shutdown signal.
40    ///
41    /// [`futures::select_biased`] polls `sys` first, so a system frame preempts
42    /// any data backlog deterministically. When a lane closes, its receiver
43    /// (a [`FusedStream`]) yields `None`; the `loop` swallows that first `None`
44    /// so the next iteration just skips the dead lane instead of treating it as
45    /// shutdown. This is so the sys lane can keep draining even after the data
46    /// lane shuts down — `None` is returned only once *both* lanes have closed.
47    ///
48    /// [`FusedStream`]: futures::stream::FusedStream
49    pub async fn recv(&mut self) -> Option<Received> {
50        loop {
51            futures::select_biased! {
52                sys = self.sys.next() => {
53                    if let Some((dir, f)) = sys {
54                        return Some(Received::Sys(dir, f));
55                    }
56                }
57                data = self.data.next() => {
58                    if let Some(f) = data {
59                        return Some(Received::Data(f));
60                    }
61                }
62                complete => return None,
63            }
64        }
65    }
66
67    /// Drain everything currently queued on the data lane. Frames where
68    /// `survives_flush()` is false are dropped; survivors are returned in the
69    /// order they arrived, for the caller to re-process. Does not block and does
70    /// not touch the sys lane.
71    pub fn flush_data(&mut self) -> Vec<DataFrame> {
72        let mut kept = Vec::new();
73        while let Ok(frame) = self.data.try_recv() {
74            if frame.survives_flush() {
75                kept.push(frame);
76            }
77        }
78        kept
79    }
80}
pipecrab_runtime/inbound.rs

pipecrab_runtime/
inbound.rs
