citadel_sdk/media/
endpoint.rs

1use super::config::MediaTransportConfig;
2use super::receiver::MediaReceiver;
3use super::sender::MediaSender;
4use super::transport::{BoxedSink, BoxedSource, MediaTransportKind, ReliableSink};
5use crate::prelude::{CitadelClientServerConnection, PeerChannel, UdpChannel};
6use crate::remote_ext::remote_specialization::PeerRemote;
7use crate::remote_ext::results::PeerConnectSuccess;
8use citadel_io::time::{timeout, Instant};
9use citadel_io::tokio::sync::oneshot::Receiver;
10use citadel_io::ErrorCode;
11use citadel_proto::prelude::{NetworkError, OutboundUdpSender, PeerChannelRecvHalf, Ratchet};
12
13/// A media session bound to one connection. Build with
14/// [`MediaEndpoint::from_peer_connection`] or [`MediaEndpoint::from_c2s`],
15/// then [`MediaEndpoint::split`].
16pub struct MediaEndpoint {
17    kind: MediaTransportKind,
18    sender: MediaSender,
19    receiver: MediaReceiver,
20}
21
22impl std::fmt::Debug for MediaEndpoint {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        f.debug_struct("MediaEndpoint")
25            .field("kind", &self.kind)
26            .finish()
27    }
28}
29
30/// The two halves of a [`MediaEndpoint`].
31#[derive(Debug)]
32pub struct MediaEndpointParts {
33    pub kind: MediaTransportKind,
34    pub sender: MediaSender,
35    pub receiver: MediaReceiver,
36}
37
38impl MediaEndpoint {
39    /// Consumes a P2P connection's channels and returns the endpoint together
40    /// with the connection's [`PeerRemote`] (take any file-transfer handle
41    /// receiver with `get_incoming_file_transfer_handle` beforehand).
42    /// Waits `cfg.udp_wait` for UDP; falls back to reliable mode.
43    pub async fn from_peer_connection<R: Ratchet>(
44        conn: PeerConnectSuccess<R>,
45        cfg: MediaTransportConfig,
46    ) -> Result<(Self, PeerRemote<R>), NetworkError> {
47        let endpoint = Self::from_channels(conn.channel, conn.udp_channel_rx, cfg).await?;
48        Ok((endpoint, conn.remote))
49    }
50
51    /// Builds directly from a reliable channel and the optional pending UDP
52    /// channel receiver (as found on every connection-success type).
53    pub async fn from_channels<R: Ratchet>(
54        channel: PeerChannel<R>,
55        udp_channel_rx: Option<Receiver<UdpChannel<R>>>,
56        cfg: MediaTransportConfig,
57    ) -> Result<Self, NetworkError> {
58        Self::build(channel, udp_channel_rx, cfg).await
59    }
60
61    /// Same as [`Self::from_peer_connection`] for a client↔server connection.
62    pub async fn from_c2s<R: Ratchet>(
63        conn: &mut CitadelClientServerConnection<R>,
64        cfg: MediaTransportConfig,
65    ) -> Result<Self, NetworkError> {
66        let channel = conn.take_channel().ok_or_else(|| {
67            citadel_io::error!(
68                ErrorCode::MediaTransportClosed,
69                "reliable channel already taken"
70            )
71        })?;
72        Self::from_channels(channel, conn.udp_channel_rx.take(), cfg).await
73    }
74
75    async fn build<R: Ratchet>(
76        channel: PeerChannel<R>,
77        udp_rx: Option<Receiver<UdpChannel<R>>>,
78        cfg: MediaTransportConfig,
79    ) -> Result<Self, NetworkError> {
80        cfg.validate()?;
81        let start = Instant::now();
82        let (reliable_tx, reliable_rx) = channel.split();
83        let control_sink = ReliableSink::spawn(reliable_tx);
84        let control_src: BoxedSource = Box::pin(reliable_rx);
85
86        match await_udp(udp_rx, &cfg).await {
87            Some((udp_tx, udp_rx)) => {
88                if cfg.udp_payload_budget > udp_tx.max_payload_len() {
89                    return Err(citadel_io::error!(
90                        ErrorCode::MediaConfigInvalid,
91                        format!(
92                            "udp_payload_budget {} exceeds the UDP channel's max payload {}",
93                            cfg.udp_payload_budget,
94                            udp_tx.max_payload_len()
95                        )
96                    ));
97                }
98                Self::assemble(
99                    MediaTransportKind::Unreliable,
100                    Box::new(udp_tx),
101                    Box::new(control_sink),
102                    Box::pin(udp_rx),
103                    Some(control_src),
104                    cfg,
105                    start,
106                )
107            }
108            None => Self::assemble(
109                MediaTransportKind::Reliable,
110                Box::new(control_sink.clone()),
111                Box::new(control_sink),
112                control_src,
113                None,
114                cfg,
115                start,
116            ),
117        }
118    }
119
120    pub(crate) fn assemble(
121        kind: MediaTransportKind,
122        media_sink: BoxedSink,
123        control_sink: BoxedSink,
124        media_src: BoxedSource,
125        control_src: Option<BoxedSource>,
126        cfg: MediaTransportConfig,
127        start: Instant,
128    ) -> Result<Self, NetworkError> {
129        Ok(Self {
130            kind,
131            sender: MediaSender::new(
132                kind,
133                media_sink,
134                control_sink,
135                cfg.media,
136                cfg.send_queue_frames,
137            )?,
138            receiver: MediaReceiver::new(kind, media_src, control_src, cfg.media, start)?,
139        })
140    }
141
142    pub fn kind(&self) -> MediaTransportKind {
143        self.kind
144    }
145
146    /// Splits into independent send/receive halves. Dropping the receiver in
147    /// unreliable mode drops the UDP receive half (⇒ `DisconnectUDP`).
148    pub fn split(self) -> (MediaSender, MediaReceiver) {
149        (self.sender, self.receiver)
150    }
151
152    pub fn into_parts(self) -> MediaEndpointParts {
153        MediaEndpointParts {
154            kind: self.kind,
155            sender: self.sender,
156            receiver: self.receiver,
157        }
158    }
159}
160
161/// Resolves the UDP channel within `cfg.udp_wait`, or `None` to fall back.
162/// A missing receiver (`UdpMode::Disabled`), a timeout, or a dropped sender
163/// all mean "no UDP" — logged, never fatal.
164async fn await_udp<R: Ratchet>(
165    udp_rx: Option<Receiver<UdpChannel<R>>>,
166    cfg: &MediaTransportConfig,
167) -> Option<(OutboundUdpSender, PeerChannelRecvHalf<R>)> {
168    let rx = udp_rx?;
169    match timeout(cfg.udp_wait, rx).await {
170        Ok(Ok(chan)) => Some(chan.split()),
171        Ok(Err(_)) => {
172            log::warn!(target: "citadel", "media: UDP channel sender dropped; using reliable transport");
173            None
174        }
175        Err(_) => {
176            log::warn!(target: "citadel", "media: UDP channel not ready within {:?}; using reliable transport", cfg.udp_wait);
177            None
178        }
179    }
180}
citadel_sdk/media/endpoint.rs

citadel_sdk/media/
endpoint.rs
