videosdk/resources/
rtmp.rs

1//! The RTMP-out (livestream) API.
2
3use std::sync::Arc;
4
5use futures_util::Stream;
6use reqwest::Method;
7use serde::{Deserialize, Serialize};
8use serde_json::{Map, Value};
9
10use crate::client::{CallOptions, Client};
11use crate::common::{
12    CompositionConfig, LayoutPriority, ResourceLinks, TranscriptionConfig, WebhookDeliverySummary,
13};
14use crate::error::{Error, Result};
15use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
16use crate::query::QueryBuilder;
17use crate::resources::egress::{
18    composition_to_config, to_egress_handle, Composition, EgressHandle, EgressType, StopTarget,
19    StopWire,
20};
21use crate::resources::escape;
22
23const PATH: &str = "/v2/livestreams";
24
25/// An RTMP destination.
26#[derive(Debug, Clone, Serialize, Deserialize)]
27#[serde(rename_all = "camelCase")]
28pub struct RtmpStream {
29    /// The RTMP ingest URL.
30    pub url: String,
31    /// The RTMP stream key.
32    pub stream_key: String,
33}
34
35/// An RTMP-out livestream.
36#[derive(Debug, Clone, Deserialize)]
37#[serde(rename_all = "camelCase")]
38pub struct Livestream {
39    /// The livestream id.
40    pub id: String,
41    /// The composer producing the stream.
42    pub composer_id: Option<String>,
43    /// The room being streamed.
44    pub room_id: Option<String>,
45    /// The session being streamed.
46    pub session_id: Option<String>,
47    /// The RTMP destinations.
48    #[serde(default, deserialize_with = "crate::common::null_to_default")]
49    pub outputs: Vec<RtmpStream>,
50    /// When the livestream started.
51    pub start: Option<String>,
52    /// When the livestream ended, or `None` if it is still running.
53    pub end: Option<String>,
54    /// The region the livestream ran in.
55    pub region: Option<String>,
56    /// A summary of the webhook deliveries for this livestream.
57    pub webhook: Option<WebhookDeliverySummary>,
58    /// HATEOAS-style links to related resources.
59    #[serde(default, deserialize_with = "crate::common::null_to_default")]
60    pub links: ResourceLinks,
61    /// Any fields the server returned that this SDK does not model yet.
62    #[serde(flatten)]
63    pub extra: Map<String, Value>,
64}
65
66/// The parameters for [`RtmpResource::start`].
67#[derive(Debug, Clone, Default)]
68pub struct RtmpStartParams {
69    /// The RTMP destinations. Required, and must be non-empty.
70    pub streams: Vec<RtmpStream>,
71    /// The layout, quality, orientation and theme. RTMP requires a layout.
72    pub composition: Option<Composition>,
73    /// Transcribes and optionally summarizes the livestream.
74    pub transcription: Option<TranscriptionConfig>,
75    /// A pre-acquired resource-pool unit id.
76    pub resource_id: Option<String>,
77}
78
79/// `streams` is sent as `outputs`.
80#[derive(Debug, Serialize)]
81#[serde(rename_all = "camelCase")]
82struct RtmpStartWire<'a> {
83    room_id: &'a str,
84    outputs: &'a [RtmpStream],
85    #[serde(skip_serializing_if = "Option::is_none")]
86    config: Option<CompositionConfig>,
87    #[serde(skip_serializing_if = "Option::is_none")]
88    template_url: Option<String>,
89    #[serde(skip_serializing_if = "Option::is_none")]
90    transcription: Option<&'a TranscriptionConfig>,
91    #[serde(skip_serializing_if = "Option::is_none")]
92    resource_id: Option<&'a str>,
93}
94
95/// The query parameters for [`RtmpResource::list`].
96#[derive(Debug, Clone, Default)]
97pub struct RtmpListParams {
98    /// The 1-based page number.
99    pub page: Option<u32>,
100    /// Items per page.
101    pub per_page: Option<u32>,
102    /// An opaque cursor from a previous page.
103    pub cursor: Option<String>,
104    /// Filters by room id.
105    pub room_id: Option<String>,
106    /// Filters by session id.
107    pub session_id: Option<String>,
108    /// Scopes to a specific user. Admin tokens only.
109    pub user_id: Option<String>,
110    /// Matches the room's meeting id.
111    pub query: Option<String>,
112}
113
114impl RtmpListParams {
115    fn pagination(&self) -> ListParams {
116        ListParams {
117            page: self.page,
118            per_page: self.per_page,
119            cursor: self.cursor.clone(),
120        }
121    }
122}
123
124/// The RTMP-out API: push a room to RTMP destinations. Reached via [`Client::rtmp`].
125#[derive(Debug, Clone, Copy)]
126pub struct RtmpResource<'a> {
127    client: &'a Client,
128}
129
130impl<'a> RtmpResource<'a> {
131    pub(crate) fn new(client: &'a Client) -> Self {
132        Self { client }
133    }
134
135    /// Begins an RTMP livestream.
136    pub async fn start(&self, room_id: &str, params: RtmpStartParams) -> Result<EgressHandle> {
137        if params.streams.is_empty() {
138            return Err(Error::validation(
139                "rtmp.start() requires at least one destination in `streams`",
140            ));
141        }
142        let mapped = composition_to_config(
143            params.composition.as_ref(),
144            // RTMP requires a priority whenever a layout is present.
145            Some(LayoutPriority::Speaker),
146        );
147        let body = RtmpStartWire {
148            room_id,
149            outputs: &params.streams,
150            config: mapped.config,
151            template_url: mapped.template_url,
152            transcription: params.transcription.as_ref(),
153            resource_id: params.resource_id.as_deref(),
154        };
155        let path = format!("{PATH}/start");
156        let raw = self
157            .client
158            .maybe_json(Method::POST, &path, CallOptions::json(&body)?)
159            .await?;
160        Ok(to_egress_handle(EgressType::Livestream, room_id, raw))
161    }
162
163    /// Stops an RTMP livestream. Accepts the start handle, or a bare room id.
164    pub async fn stop(&self, target: impl Into<StopTarget>) -> Result<String> {
165        let target = target.into();
166        let path = format!("{PATH}/end");
167        self.client
168            .message(
169                Method::POST,
170                &path,
171                CallOptions::json(StopWire::from(&target))?,
172            )
173            .await
174    }
175
176    /// Lists livestreams, one page at a time.
177    pub async fn list(&self, params: RtmpListParams) -> Result<Page<Livestream>> {
178        paginate(self.fetcher(&params), &params.pagination(), "data", None).await
179    }
180
181    /// Lists livestreams, transparently fetching every page.
182    pub fn list_stream(
183        &self,
184        params: RtmpListParams,
185    ) -> impl Stream<Item = Result<Livestream>> + Send {
186        auto_page(self.fetcher(&params), params.pagination(), "data", None)
187    }
188
189    /// Fetches a livestream by id.
190    pub async fn get(&self, id: &str) -> Result<Livestream> {
191        let path = format!("{PATH}/{}", escape(id));
192        self.client
193            .json(Method::GET, &path, CallOptions::new())
194            .await
195    }
196
197    fn fetcher(&self, params: &RtmpListParams) -> PageFetcher {
198        let client = self.client.clone();
199        let params = params.clone();
200        Arc::new(move |page, per_page| {
201            let client = client.clone();
202            let params = params.clone();
203            Box::pin(async move {
204                let query = QueryBuilder::new()
205                    .opt("page", page)
206                    .opt("perPage", per_page)
207                    .opt_str("roomId", params.room_id.as_deref())
208                    .opt_str("sessionId", params.session_id.as_deref())
209                    .opt_str("userId", params.user_id.as_deref())
210                    .opt_str("query", params.query.as_deref())
211                    .into_pairs();
212                client
213                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
214                    .await
215            })
216        })
217    }
218}
videosdk/resources/rtmp.rs

videosdk/resources/
rtmp.rs
