videosdk/resources/
ingress.rs

1//! WHIP publish, WHEP playback, and socket ingest.
2
3use std::time::Duration;
4
5use reqwest::{Method, Url};
6use serde::{Deserialize, Serialize};
7use serde_json::{Map, Value};
8
9use crate::client::{CallOptions, Client};
10use crate::error::{Error, Result};
11use crate::query::QueryBuilder;
12use crate::resources::escape;
13use crate::resources::random_uuid;
14use crate::resources::session_utils::resolve_active_session_id;
15
16/// Builds `<base><path>?<query>`, including only non-empty parameters.
17fn build_ingress_url(base: &str, path: &str, params: &[(&str, &str)]) -> Result<String> {
18    let mut url = Url::parse(&format!("{}{path}", base.trim_end_matches('/')))
19        .map_err(|e| Error::config(format!("invalid base URL {base:?}: {e}")))?;
20
21    let pairs: Vec<_> = params
22        .iter()
23        .filter(|(_, value)| !value.is_empty())
24        .collect();
25    if !pairs.is_empty() {
26        let mut query = url.query_pairs_mut();
27        for (name, value) in pairs {
28            query.append_pair(name, value);
29        }
30    }
31    Ok(url.into())
32}
33
34/* ------------------------------------ WHIP ------------------------------------ */
35
36/// The parameters for [`WhipResource::create`].
37#[derive(Debug, Clone, Default)]
38pub struct CreateWhipParams {
39    /// The publisher's participant id. Auto-generated as `whip-<uuid>` when absent.
40    pub participant_id: Option<String>,
41    /// The display name shown for the publisher.
42    pub name: Option<String>,
43    /// How long the publish credential stays valid. Defaults to one hour.
44    pub expires_in: Option<Duration>,
45    /// Reuses an existing participant instead of creating a new one.
46    pub use_existing_peer: Option<bool>,
47}
48
49/// A WHIP ingest credential.
50///
51/// [`WhipResource::create`] builds this locally, with no network call: `POST`
52/// your SDP offer to [`url`](WhipIngress::url) with `Content-Type:
53/// application/sdp` and `Authorization: <token>`, and the server replies with
54/// the SDP answer.
55#[derive(Debug, Clone)]
56pub struct WhipIngress {
57    /// The WHIP endpoint. POST the SDP offer here.
58    pub url: String,
59    /// Authorizes the publish. Short-lived and sensitive — do not log it.
60    pub token: String,
61    /// An alias of [`token`](WhipIngress::token). Some WHIP tools call it the
62    /// stream key.
63    pub stream_key: String,
64    /// The room being published into.
65    pub room_id: String,
66    /// The publisher's participant id.
67    pub participant_id: String,
68    /// The publisher's display name.
69    pub display_name: Option<String>,
70    /// Targets a specific session on delete, e.g. from the publish `Location`
71    /// header. The room's active session is resolved when absent.
72    pub session_id: Option<String>,
73}
74
75/// WHIP ingress: WebRTC-HTTP publish into a room. Reached via [`Client::whip`].
76#[derive(Debug, Clone, Copy)]
77pub struct WhipResource<'a> {
78    client: &'a Client,
79}
80
81impl<'a> WhipResource<'a> {
82    pub(crate) fn new(client: &'a Client) -> Self {
83        Self { client }
84    }
85
86    /// Builds a WHIP publish credential.
87    ///
88    /// Synchronous: this mints a token and assembles a URL locally, and makes no
89    /// network call.
90    pub fn create(&self, room_id: &str, params: CreateWhipParams) -> Result<WhipIngress> {
91        let participant_id = params
92            .participant_id
93            .unwrap_or_else(|| format!("whip-{}", random_uuid()));
94        let token = self
95            .client
96            .mint_api_token(params.expires_in.unwrap_or_default())?;
97
98        let url = build_ingress_url(
99            self.client.base_url(),
100            "/v2/whip",
101            &[
102                ("roomId", room_id),
103                ("participantId", &participant_id),
104                ("displayName", params.name.as_deref().unwrap_or("")),
105                (
106                    "useExistingPeer",
107                    if params.use_existing_peer == Some(true) {
108                        "true"
109                    } else {
110                        ""
111                    },
112                ),
113            ],
114        )?;
115
116        Ok(WhipIngress {
117            url,
118            stream_key: token.clone(),
119            token,
120            room_id: room_id.to_string(),
121            participant_id,
122            display_name: params.name,
123            session_id: None,
124        })
125    }
126
127    /// Tears down a WHIP publisher.
128    ///
129    /// The target's `session_id` is used when set; otherwise the room's active
130    /// session is resolved.
131    pub async fn delete(&self, target: &WhipIngress) -> Result<()> {
132        let session_id = match &target.session_id {
133            Some(session_id) => session_id.clone(),
134            None => resolve_active_session_id(self.client, &target.room_id)
135                .await?
136                .ok_or_else(|| {
137                    Error::not_found(format!("no active session for room {}", target.room_id))
138                })?,
139        };
140
141        let query = QueryBuilder::new()
142            .str_val("participantId", &target.participant_id)
143            .opt_str("displayName", target.display_name.as_deref())
144            .into_pairs();
145        let path = format!("/v2/whip/sessions/{}", escape(&session_id));
146        self.client
147            .none(Method::DELETE, &path, CallOptions::new().query(query))
148            .await
149    }
150}
151
152/* ------------------------------------ WHEP ------------------------------------ */
153
154/// Selects the remote participant a WHEP subscriber pulls.
155#[derive(Debug, Clone)]
156pub struct WhepSource {
157    /// The remote participant to pull.
158    pub participant_id: String,
159}
160
161/// The parameters for [`WhepResource::create`].
162#[derive(Debug, Clone, Default)]
163pub struct CreateWhepParams {
164    /// The subscriber's own participant id. Auto-generated as `whep-<uuid>` when
165    /// absent.
166    pub participant_id: Option<String>,
167    /// Which remote participant to pull. Omit to let the server choose.
168    pub source: Option<WhepSource>,
169    /// How long the playback credential stays valid. Defaults to one hour.
170    pub expires_in: Option<Duration>,
171}
172
173/// A WHEP playback credential.
174///
175/// [`WhepResource::create`] builds this locally, with no network call.
176#[derive(Debug, Clone)]
177pub struct WhepPlayback {
178    /// The WHEP endpoint. POST the SDP offer here.
179    pub url: String,
180    /// Authorizes the pull. Short-lived and sensitive — do not log it.
181    pub token: String,
182    /// The room being pulled from.
183    pub room_id: String,
184    /// The subscriber's participant id.
185    pub participant_id: String,
186    /// The remote participant being pulled, when a source was given.
187    pub remote_peer_id: Option<String>,
188    /// Targets a specific session on delete.
189    pub session_id: Option<String>,
190}
191
192/// WHEP egress: standards-based WebRTC pull from a room. Requires an active
193/// session. Reached via [`Client::whep`].
194#[derive(Debug, Clone, Copy)]
195pub struct WhepResource<'a> {
196    client: &'a Client,
197}
198
199impl<'a> WhepResource<'a> {
200    pub(crate) fn new(client: &'a Client) -> Self {
201        Self { client }
202    }
203
204    /// Builds a WHEP playback credential.
205    ///
206    /// Synchronous: this mints a token and assembles a URL locally, and makes no
207    /// network call.
208    pub fn create(&self, room_id: &str, params: CreateWhepParams) -> Result<WhepPlayback> {
209        let participant_id = params
210            .participant_id
211            .unwrap_or_else(|| format!("whep-{}", random_uuid()));
212        let remote_peer_id = params.source.map(|source| source.participant_id);
213        let token = self
214            .client
215            .mint_api_token(params.expires_in.unwrap_or_default())?;
216
217        let url = build_ingress_url(
218            self.client.base_url(),
219            "/v2/whep",
220            &[
221                ("roomId", room_id),
222                ("participantId", &participant_id),
223                ("remotePeerId", remote_peer_id.as_deref().unwrap_or("")),
224            ],
225        )?;
226
227        Ok(WhepPlayback {
228            url,
229            token,
230            room_id: room_id.to_string(),
231            participant_id,
232            remote_peer_id,
233            session_id: None,
234        })
235    }
236
237    /// Tears down a WHEP subscriber.
238    pub async fn delete(&self, target: &WhepPlayback) -> Result<()> {
239        let session_id = match &target.session_id {
240            Some(session_id) => session_id.clone(),
241            None => resolve_active_session_id(self.client, &target.room_id)
242                .await?
243                .ok_or_else(|| {
244                    Error::not_found(format!("no active session for room {}", target.room_id))
245                })?,
246        };
247
248        let query = QueryBuilder::new()
249            .str_val("participantId", &target.participant_id)
250            .opt_str("remotePeerId", target.remote_peer_id.as_deref())
251            .into_pairs();
252        let path = format!("/v2/whep/sessions/{}", escape(&session_id));
253        self.client
254            .none(Method::DELETE, &path, CallOptions::new().query(query))
255            .await
256    }
257}
258
259/* -------------------------------- socket ingest -------------------------------- */
260
261/// Agent metadata to associate with an ingest session.
262#[derive(Debug, Clone, Serialize)]
263#[serde(rename_all = "camelCase")]
264pub struct SocketIngressAgent {
265    /// The agent id.
266    pub id: String,
267    /// The agent type.
268    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
269    pub kind: Option<String>,
270    /// Free-form metadata.
271    #[serde(skip_serializing_if = "Option::is_none")]
272    pub metadata: Option<Map<String, Value>>,
273}
274
275/// The parameters for [`SocketIngressResource::create`].
276#[derive(Debug, Clone, Default)]
277pub struct CreateSocketIngressParams {
278    /// The streamed-in participant's id. Auto-generated when absent.
279    pub participant_id: Option<String>,
280    /// The participant's display name.
281    pub name: Option<String>,
282    /// Free-form participant metadata.
283    pub metadata: Option<Map<String, Value>>,
284    /// Agent metadata to associate with the session.
285    pub agent: Option<SocketIngressAgent>,
286    /// Where to run the ingest session.
287    pub region: Option<String>,
288}
289
290#[derive(Debug, Serialize)]
291#[serde(rename_all = "camelCase")]
292struct SocketIngressParticipant {
293    #[serde(skip_serializing_if = "Option::is_none")]
294    id: Option<String>,
295    #[serde(skip_serializing_if = "Option::is_none")]
296    name: Option<String>,
297    #[serde(skip_serializing_if = "Option::is_none")]
298    metadata: Option<Map<String, Value>>,
299}
300
301#[derive(Debug, Serialize)]
302#[serde(rename_all = "camelCase")]
303struct SocketIngressWire<'a> {
304    room_id: &'a str,
305    #[serde(skip_serializing_if = "Option::is_none")]
306    participant: Option<SocketIngressParticipant>,
307    #[serde(skip_serializing_if = "Option::is_none")]
308    agent: Option<SocketIngressAgent>,
309    #[serde(skip_serializing_if = "Option::is_none")]
310    region: Option<String>,
311}
312
313/// A socket-ingest session.
314#[derive(Debug, Clone, Deserialize)]
315#[serde(rename_all = "camelCase")]
316pub struct SocketIngress {
317    /// The single-use WebSocket URL, valid for about 90 seconds. Connect and
318    /// stream media or data frames.
319    pub ws_url: String,
320    /// The room being streamed into.
321    #[serde(default, deserialize_with = "crate::common::null_to_default")]
322    pub room_id: String,
323    /// How long the URL stays valid, in seconds.
324    #[serde(default, deserialize_with = "crate::common::null_to_default")]
325    pub expires_in: u32,
326    /// The single-use handle parsed out of [`ws_url`](SocketIngress::ws_url).
327    #[serde(skip)]
328    pub ws_ref: Option<String>,
329    /// Any fields the server returned that this SDK does not model yet.
330    #[serde(flatten)]
331    pub extra: Map<String, Value>,
332}
333
334/// Reads the `ref` query parameter out of a WebSocket URL.
335fn extract_ws_ref(ws_url: &str) -> Option<String> {
336    let url = Url::parse(ws_url).ok()?;
337    url.query_pairs()
338        .find(|(name, _)| name == "ref")
339        .map(|(_, value)| value.into_owned())
340}
341
342/// Streams media and data frames into a room over a single-use WebSocket. To
343/// tear down, close the socket; the URL also expires on its own. Reached via
344/// [`Client::socket_ingress`].
345#[derive(Debug, Clone, Copy)]
346pub struct SocketIngressResource<'a> {
347    client: &'a Client,
348}
349
350impl<'a> SocketIngressResource<'a> {
351    pub(crate) fn new(client: &'a Client) -> Self {
352        Self { client }
353    }
354
355    /// Creates a socket-ingest session for a room.
356    pub async fn create(
357        &self,
358        room_id: &str,
359        params: CreateSocketIngressParams,
360    ) -> Result<SocketIngress> {
361        let has_participant =
362            params.participant_id.is_some() || params.name.is_some() || params.metadata.is_some();
363        let participant = if has_participant {
364            Some(SocketIngressParticipant {
365                id: params.participant_id,
366                name: params.name,
367                metadata: params.metadata,
368            })
369        } else {
370            None
371        };
372
373        let body = SocketIngressWire {
374            room_id,
375            participant,
376            agent: params.agent,
377            region: params.region,
378        };
379
380        let mut ingress: SocketIngress = self
381            .client
382            .data(
383                Method::POST,
384                "/v2/ingest/sessions",
385                CallOptions::json(&body)?,
386            )
387            .await?;
388        ingress.ws_ref = extract_ws_ref(&ingress.ws_url);
389        Ok(ingress)
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use super::*;
396
397    #[test]
398    fn ingress_urls_omit_empty_parameters() {
399        let url = build_ingress_url(
400            "https://api.videosdk.live/",
401            "/v2/whip",
402            &[
403                ("roomId", "r-1"),
404                ("displayName", ""),
405                ("useExistingPeer", ""),
406            ],
407        )
408        .unwrap();
409        assert_eq!(url, "https://api.videosdk.live/v2/whip?roomId=r-1");
410    }
411
412    #[test]
413    fn ingress_urls_have_no_trailing_question_mark_when_empty() {
414        let url = build_ingress_url("https://api.videosdk.live", "/v2/whip", &[("a", "")]).unwrap();
415        assert_eq!(url, "https://api.videosdk.live/v2/whip");
416    }
417
418    #[test]
419    fn ingress_urls_percent_encode_their_values() {
420        let url =
421            build_ingress_url("https://x.test", "/v2/whep", &[("displayName", "Ada L")]).unwrap();
422        assert!(url.contains("displayName=Ada+L") || url.contains("displayName=Ada%20L"));
423    }
424
425    #[test]
426    fn the_ws_ref_is_read_out_of_the_url() {
427        assert_eq!(
428            extract_ws_ref("wss://x.test/ingest?ref=abc123&other=1").as_deref(),
429            Some("abc123")
430        );
431        assert_eq!(extract_ws_ref("wss://x.test/ingest"), None);
432        assert_eq!(extract_ws_ref("not a url"), None);
433    }
434}
videosdk/resources/ingress.rs

videosdk/resources/
ingress.rs
