citadel_sdk/media/
endpoint.rs

1use super::config::MediaTransportConfig;
2use super::negotiation::agree_on_udp;
3use super::receiver::MediaReceiver;
4use super::sender::MediaSender;
5use super::transport::{BoxedSink, BoxedSource, MediaTransportKind, ReliableSink};
6use crate::prelude::{CitadelClientServerConnection, PeerChannel, UdpChannel};
7use crate::remote_ext::remote_specialization::PeerRemote;
8use crate::remote_ext::results::PeerConnectSuccess;
9use citadel_io::time::{timeout, Instant};
10use citadel_io::tokio::sync::oneshot::Receiver;
11use citadel_io::ErrorCode;
12use citadel_proto::prelude::{
13    NetworkError, OutboundUdpSender, PathClass, PeerChannelRecvHalf, Ratchet,
14};
15
16/// A media session bound to one connection. Build with
17/// [`MediaEndpoint::from_peer_connection`] or [`MediaEndpoint::from_c2s`],
18/// then [`MediaEndpoint::split`].
19pub struct MediaEndpoint {
20    kind: MediaTransportKind,
21    sender: MediaSender,
22    receiver: MediaReceiver,
23}
24
25impl std::fmt::Debug for MediaEndpoint {
26    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        f.debug_struct("MediaEndpoint")
28            .field("kind", &self.kind)
29            .finish()
30    }
31}
32
33/// The two halves of a [`MediaEndpoint`].
34#[derive(Debug)]
35pub struct MediaEndpointParts {
36    pub kind: MediaTransportKind,
37    pub sender: MediaSender,
38    pub receiver: MediaReceiver,
39}
40
41impl MediaEndpoint {
42    /// Consumes a P2P connection's channels and returns the endpoint together
43    /// with the connection's [`PeerRemote`] (take any file-transfer handle
44    /// receiver with `get_incoming_file_transfer_handle` beforehand).
45    /// Waits `cfg.udp_wait` for UDP, then exchanges transport offers with the
46    /// peer over the reliable channel: UDP is used only if both sides hold
47    /// it, otherwise both fall back to reliable mode.
48    ///
49    /// The peer must build its endpoint too. The wait for its offer has no
50    /// timer: it lasts until the offer arrives or the channel closes
51    /// (`MediaTransportClosed`), however long the peer takes to build. A
52    /// timer here could not be made safe, because this side's offer is
53    /// already sent: a late peer would act on it while this side had
54    /// already fallen back. Bound the call yourself if the peer may never
55    /// build; dropping the future drops the channel, which the peer then
56    /// sees as closed.
57    ///
58    /// Refused with `ErrorCode::PathClassDenied` when, once the transport is agreed, the
59    /// connection's path may not carry media: the server's path policy keeps media off it.
60    ///
61    /// A peer whose protocol version predates transport offers
62    /// ([`citadel_proto::constants::MEDIA_TRANSPORT_OFFER_SINCE`], or unknown because the peer or the
63    /// server relaying its key exchange predates carrying the version) is
64    /// sent none and never waited on: this side uses reliable mode.
65    pub async fn from_peer_connection<R: Ratchet>(
66        conn: PeerConnectSuccess<R>,
67        cfg: MediaTransportConfig,
68    ) -> Result<(Self, PeerRemote<R>), NetworkError> {
69        let endpoint = Self::from_channels(conn.channel, conn.udp_channel_rx, cfg).await?;
70        Ok((endpoint, conn.remote))
71    }
72
73    /// Builds directly from a reliable channel and the optional pending UDP
74    /// channel receiver (as found on every connection-success type).
75    pub async fn from_channels<R: Ratchet>(
76        channel: PeerChannel<R>,
77        udp_channel_rx: Option<Receiver<UdpChannel<R>>>,
78        cfg: MediaTransportConfig,
79    ) -> Result<Self, NetworkError> {
80        Self::build(channel, udp_channel_rx, cfg).await
81    }
82
83    /// Same as [`Self::from_peer_connection`] for a client↔server connection.
84    pub async fn from_c2s<R: Ratchet>(
85        conn: &mut CitadelClientServerConnection<R>,
86        cfg: MediaTransportConfig,
87    ) -> Result<Self, NetworkError> {
88        let channel = conn.take_channel().ok_or_else(|| {
89            citadel_io::error!(
90                ErrorCode::MediaTransportClosed,
91                "reliable channel already taken"
92            )
93        })?;
94        Self::from_channels(channel, conn.udp_channel_rx.take(), cfg).await
95    }
96
97    async fn build<R: Ratchet>(
98        channel: PeerChannel<R>,
99        udp_rx: Option<Receiver<UdpChannel<R>>>,
100        cfg: MediaTransportConfig,
101    ) -> Result<Self, NetworkError> {
102        cfg.validate()?;
103        let start = Instant::now();
104        let peer_protocol_version = channel.peer_protocol_version();
105        // Media may ride only a path the server's policy gives it: on a server relay that carries
106        // no media this refuses the call (PathClassDenied) instead of relaying it. Checked before
107        // anything is sent, so each side is refused on its own gates. Checked after the transport
108        // offers, the side refused first dropped its channel (a disconnect) while its own offer
109        // could still be queued, and the other side failed with MediaTransportClosed instead.
110        channel.p2p_path_cell().check(PathClass::Media)?;
111        let (reliable_tx, reliable_rx) = channel.split();
112        let mut control_sink = ReliableSink::spawn(reliable_tx);
113        let mut control_src: BoxedSource = Box::pin(reliable_rx);
114
115        let local_udp = await_udp(udp_rx, &cfg).await;
116        if let Some((udp_tx, _)) = &local_udp {
117            if cfg.udp_payload_budget > udp_tx.max_payload_len() {
118                return Err(citadel_io::error!(
119                    ErrorCode::MediaConfigInvalid,
120                    format!(
121                        "udp_payload_budget {} exceeds the UDP channel's max payload {}",
122                        cfg.udp_payload_budget,
123                        udp_tx.max_payload_len()
124                    )
125                ));
126            }
127        }
128        let peer_udp = agree_on_udp(
129            peer_protocol_version,
130            local_udp.is_some(),
131            &mut control_sink,
132            &mut control_src,
133        )
134        .await?;
135
136        match local_udp.filter(|_| peer_udp) {
137            Some((udp_tx, udp_rx)) => Self::assemble(
138                MediaTransportKind::Unreliable,
139                Box::new(udp_tx),
140                Box::new(control_sink),
141                Box::pin(udp_rx),
142                Some(control_src),
143                cfg,
144                start,
145            ),
146            None => Self::assemble(
147                MediaTransportKind::Reliable,
148                Box::new(control_sink.clone()),
149                Box::new(control_sink),
150                control_src,
151                None,
152                cfg,
153                start,
154            ),
155        }
156    }
157
158    pub(crate) fn assemble(
159        kind: MediaTransportKind,
160        media_sink: BoxedSink,
161        control_sink: BoxedSink,
162        media_src: BoxedSource,
163        control_src: Option<BoxedSource>,
164        cfg: MediaTransportConfig,
165        start: Instant,
166    ) -> Result<Self, NetworkError> {
167        Ok(Self {
168            kind,
169            sender: MediaSender::new(
170                kind,
171                media_sink,
172                control_sink,
173                cfg.media,
174                cfg.send_queue_frames,
175            )?,
176            receiver: MediaReceiver::new(kind, media_src, control_src, cfg.media, start)?,
177        })
178    }
179
180    pub fn kind(&self) -> MediaTransportKind {
181        self.kind
182    }
183
184    /// Splits into independent send/receive halves. Dropping the receiver in
185    /// unreliable mode drops the UDP receive half (⇒ `DisconnectUDP`).
186    pub fn split(self) -> (MediaSender, MediaReceiver) {
187        (self.sender, self.receiver)
188    }
189
190    pub fn into_parts(self) -> MediaEndpointParts {
191        MediaEndpointParts {
192            kind: self.kind,
193            sender: self.sender,
194            receiver: self.receiver,
195        }
196    }
197}
198
199/// Resolves the UDP channel within `cfg.udp_wait`, or `None` to fall back.
200/// A missing receiver (`UdpMode::Disabled`), a timeout, or a dropped sender
201/// all mean "no UDP" — logged, never fatal.
202async fn await_udp<R: Ratchet>(
203    udp_rx: Option<Receiver<UdpChannel<R>>>,
204    cfg: &MediaTransportConfig,
205) -> Option<(OutboundUdpSender, PeerChannelRecvHalf<R>)> {
206    let rx = udp_rx?;
207    match timeout(cfg.udp_wait, rx).await {
208        Ok(Ok(chan)) => Some(chan.split()),
209        Ok(Err(_)) => {
210            log::warn!(target: "citadel", "media: UDP channel sender dropped; using reliable transport");
211            None
212        }
213        Err(_) => {
214            log::warn!(target: "citadel", "media: UDP channel not ready within {:?}; using reliable transport", cfg.udp_wait);
215            None
216        }
217    }
218}
citadel_sdk/media/endpoint.rs

citadel_sdk/media/
endpoint.rs
