pipecrab_runtime/
outbound.rs

1use std::sync::atomic::{AtomicU64, Ordering};
2
3use futures::channel::mpsc::{SendError, Sender};
4use futures::sink::SinkExt;
5use pipecrab_core::{DataFrame, Direction, SystemFrame};
6
7use crate::inbound::Stamped;
8
9/// The send surface of a stage: typed sends for the data and system lanes.
10///
11/// `send_data` targets the downstream data lane; `send_system` targets the
12/// system lane with an explicit [`Direction`]. Every send stamps its frame from
13/// one per-link counter shared by both lanes, which is what lets the receiving
14/// side's flush distinguish frames queued before an `Interrupt` from frames
15/// queued after it.
16///
17/// Constructed only by [`link`](crate::link).
18pub struct Outbound {
19    /// Downstream data channel.
20    pub(crate) data: Sender<Stamped<DataFrame>>,
21    /// Bidirectional system channel.
22    pub(crate) sys: Sender<Stamped<(Direction, SystemFrame)>>,
23    /// Per-link monotonic stamp shared by both lanes. Atomic so sends work
24    /// through `&self`; sends on one `Outbound` are not otherwise synchronised,
25    /// so issue them sequentially.
26    pub(crate) seq: AtomicU64,
27}
28
29impl Outbound {
30    /// Send a data frame downstream.
31    ///
32    /// Takes `&self` (not `&mut self`) so a stage can send while it is borrowed
33    /// immutably by the run loop. `futures`' `Sink::send` needs `&mut`, so we
34    /// send on a cheap clone of the shared sender; clones feed the same channel.
35    pub async fn send_data(&self, frame: DataFrame) -> Result<(), SendError> {
36        let seq = self.seq.fetch_add(1, Ordering::Relaxed);
37        self.data.clone().send(Stamped { seq, frame }).await
38    }
39
40    /// Send a system frame in the given direction. Takes `&self` for the same
41    /// reason as [`send_data`](Self::send_data).
42    pub async fn send_system(&self, dir: Direction, frame: SystemFrame) -> Result<(), SendError> {
43        let seq = self.seq.fetch_add(1, Ordering::Relaxed);
44        self.sys
45            .clone()
46            .send(Stamped {
47                seq,
48                frame: (dir, frame),
49            })
50            .await
51    }
52}
pipecrab_runtime/outbound.rs

pipecrab_runtime/
outbound.rs
