pipecrab_runtime/outbound.rs
1use futures::channel::mpsc::{SendError, Sender};
2use futures::sink::SinkExt;
3use pipecrab_core::{DataFrame, Direction, SystemFrame};
4
5/// The send surface of a stage: typed sends for the data and system lanes.
6///
7/// `send_data` targets the downstream data lane; `send_system` targets the
8/// system lane with an explicit [`Direction`].
9pub struct Outbound {
10 /// Downstream data channel.
11 pub data: Sender<DataFrame>,
12 /// Bidirectional system channel.
13 pub sys: Sender<(Direction, SystemFrame)>,
14}
15
16impl Outbound {
17 /// Send a data frame downstream.
18 ///
19 /// Takes `&self` (not `&mut self`) so a stage can send while it is borrowed
20 /// immutably by the run loop. `futures`' `Sink::send` needs `&mut`, so we
21 /// send on a cheap clone of the shared sender; clones feed the same channel.
22 pub async fn send_data(&self, frame: DataFrame) -> Result<(), SendError> {
23 self.data.clone().send(frame).await
24 }
25
26 /// Send a system frame in the given direction. Takes `&self` for the same
27 /// reason as [`send_data`](Self::send_data).
28 pub async fn send_system(&self, dir: Direction, frame: SystemFrame) -> Result<(), SendError> {
29 self.sys.clone().send((dir, frame)).await
30 }
31}