citadel_sdk/media/
transport.rs

1//! Datagram sink/source abstractions that decouple the media sender/receiver
2//! from the concrete Citadel channel halves.
3use bytes::BytesMut;
4use citadel_io::tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
5use citadel_io::ErrorCode;
6
7use citadel_proto::prelude::{
8    NetworkError, OutboundUdpSender, PeerChannelSendHalf, Ratchet, SecBuffer,
9};
10use futures::{SinkExt, Stream};
11use std::pin::Pin;
12
13/// Which path media fragments take.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum MediaTransportKind {
16    /// Fragments over the session UDP channel (lossy, unordered).
17    Unreliable,
18    /// Fragments over the ordered-reliable channel (UDP unavailable).
19    Reliable,
20}
21
22/// Synchronous, never-blocking datagram sink. Mirrors
23/// `OutboundUdpSender::unbounded_send`: a failure is terminal for that datagram.
24pub trait MediaDatagramSink: Send + Sync + 'static {
25    fn send_datagram(&mut self, buf: BytesMut) -> Result<(), NetworkError>;
26}
27
28impl MediaDatagramSink for OutboundUdpSender {
29    fn send_datagram(&mut self, buf: BytesMut) -> Result<(), NetworkError> {
30        self.unbounded_send(buf)
31    }
32}
33
34/// Stream of inbound datagrams (one `SecBuffer` per wire message).
35pub trait MediaDatagramSource: Stream<Item = SecBuffer> + Send + Sync + 'static {}
36impl<S: Stream<Item = SecBuffer> + Send + Sync + 'static> MediaDatagramSource for S {}
37
38pub(crate) type BoxedSink = Box<dyn MediaDatagramSink>;
39pub(crate) type BoxedSource = Pin<Box<dyn MediaDatagramSource>>;
40
41/// Sync façade over the ordered-reliable channel: datagrams are queued on an
42/// unbounded channel and a pump task forwards them through
43/// `PeerChannelSendHalf::into_sink`. Cloning shares the same pump.
44#[derive(Debug, Clone)]
45pub struct ReliableSink {
46    tx: UnboundedSender<BytesMut>,
47}
48
49impl ReliableSink {
50    /// Creates the sink and spawns its pump over `send_half`.
51    pub fn spawn<R: Ratchet>(send_half: PeerChannelSendHalf<R>) -> Self {
52        let (sink, rx) = Self::pair();
53        let pump = Self::pump(rx, send_half);
54        drop(citadel_io::spawn(pump));
55        sink
56    }
57
58    /// The unpumped channel pair. Exposed for tests and for callers that
59    /// drive the pump themselves.
60    pub fn pair() -> (Self, UnboundedReceiver<BytesMut>) {
61        let (tx, rx) = unbounded_channel();
62        (Self { tx }, rx)
63    }
64
65    /// Forwards every queued datagram until the queue or the channel closes.
66    pub async fn pump<R: Ratchet>(
67        mut rx: UnboundedReceiver<BytesMut>,
68        send_half: PeerChannelSendHalf<R>,
69    ) {
70        let mut sink = send_half.into_sink();
71        while let Some(buf) = rx.recv().await {
72            if let Err(err) = sink.send(buf.freeze()).await {
73                log::warn!(target: "citadel", "media reliable pump ended: {err}");
74                return;
75            }
76        }
77    }
78}
79
80impl MediaDatagramSink for ReliableSink {
81    fn send_datagram(&mut self, buf: BytesMut) -> Result<(), NetworkError> {
82        self.tx.send(buf).map_err(|_| {
83            citadel_io::error!(ErrorCode::MediaTransportClosed, "reliable pump is gone")
84        })
85    }
86}
citadel_sdk/media/transport.rs

citadel_sdk/media/
transport.rs
