videosdk/resources/
resource_pool.rs

1//! Pre-acquires and releases composition units. Enterprise tier only.
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::string_enum;
12use crate::error::Result;
13use crate::pagination::{auto_page, paginate, ListParams, Page, PageFetcher};
14use crate::query::QueryBuilder;
15use crate::resources::escape;
16
17const PATH: &str = "/v2/resource";
18
19string_enum! {
20    /// A composition unit's status.
21    ResourceStatus {
22        /// The unit is being provisioned.
23        PENDING => "pending",
24        /// The unit is warm and unused.
25        IDLE => "idle",
26        /// The unit is composing.
27        COMPOSING => "composing",
28        /// The unit has been released.
29        RELEASED => "released",
30    }
31}
32
33string_enum! {
34    /// The composer type a unit is acquired for.
35    ComposerType {
36        /// A recording composer.
37        RECORDING => "recording",
38        /// An HLS composer.
39        HLS => "hls",
40        /// An RTMP composer.
41        RTMP => "rtmp",
42    }
43}
44
45string_enum! {
46    /// A unit's capture mode.
47    ResourceMode {
48        /// Capture both video and audio.
49        VIDEO_AND_AUDIO => "video-and-audio",
50        /// Capture audio only.
51        AUDIO => "audio",
52    }
53}
54
55string_enum! {
56    /// A unit's output quality.
57    ResourceQuality {
58        /// Lowest bitrate.
59        LOW => "low",
60        /// Medium bitrate.
61        MED => "med",
62        /// High bitrate.
63        HIGH => "high",
64    }
65}
66
67/// A pre-acquired composition resource unit.
68#[derive(Debug, Clone, Deserialize)]
69#[serde(rename_all = "camelCase")]
70pub struct ResourceUnit {
71    /// The unit id. Pass it as `resource_id` when starting a composition.
72    pub id: String,
73    /// The unit's status.
74    pub status: Option<ResourceStatus>,
75    /// The unit's composer type.
76    #[serde(rename = "type")]
77    pub kind: Option<String>,
78    /// The unit's capture mode.
79    pub mode: Option<ResourceMode>,
80    /// The unit's output quality.
81    pub quality: Option<ResourceQuality>,
82    /// The composers bound to this unit.
83    #[serde(default, deserialize_with = "crate::common::null_to_default")]
84    pub composer_ids: Vec<String>,
85    /// Where unit events are delivered.
86    pub webhook_url: Option<String>,
87    /// Any fields the server returned that this SDK does not model yet.
88    #[serde(flatten)]
89    pub extra: Map<String, Value>,
90}
91
92/// The query parameters for [`ResourcePoolResource::list`].
93#[derive(Debug, Clone, Default)]
94pub struct ListResourcesParams {
95    /// The 1-based page number.
96    pub page: Option<u32>,
97    /// Items per page.
98    pub per_page: Option<u32>,
99    /// An opaque cursor from a previous page.
100    pub cursor: Option<String>,
101    /// Filters by unit status.
102    pub status: Option<ResourceStatus>,
103}
104
105impl ListResourcesParams {
106    fn pagination(&self) -> ListParams {
107        ListParams {
108            page: self.page,
109            per_page: self.per_page,
110            cursor: self.cursor.clone(),
111        }
112    }
113}
114
115/// The parameters for [`ResourcePoolResource::acquire`].
116#[derive(Debug, Clone, Default, Serialize)]
117#[serde(rename_all = "camelCase")]
118pub struct AcquireResourceParams {
119    /// The composer type to pre-warm. Required.
120    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
121    pub kind: Option<ComposerType>,
122    /// Where to deliver unit events. Required.
123    pub webhook_url: String,
124    /// How many units to acquire. Defaults to 1.
125    #[serde(skip_serializing_if = "Option::is_none")]
126    pub units: Option<u32>,
127    /// The capture mode.
128    #[serde(skip_serializing_if = "Option::is_none")]
129    pub mode: Option<ResourceMode>,
130    /// The output quality.
131    #[serde(skip_serializing_if = "Option::is_none")]
132    pub quality: Option<ResourceQuality>,
133}
134
135/// The outcome of releasing a single unit.
136#[derive(Debug, Clone, Deserialize)]
137pub struct ReleaseResourceResult {
138    /// The unit that was released.
139    pub id: String,
140    /// Whether it was released.
141    pub success: bool,
142    /// Why the release failed, when it did.
143    pub msg: Option<String>,
144}
145
146/// The resource pool. Reached via [`Client::resource_pool`].
147#[derive(Debug, Clone, Copy)]
148pub struct ResourcePoolResource<'a> {
149    client: &'a Client,
150}
151
152impl<'a> ResourcePoolResource<'a> {
153    pub(crate) fn new(client: &'a Client) -> Self {
154        Self { client }
155    }
156
157    /// Lists resource units, one page at a time.
158    pub async fn list(&self, params: ListResourcesParams) -> Result<Page<ResourceUnit>> {
159        paginate(self.fetcher(&params), &params.pagination(), "data", None).await
160    }
161
162    /// Lists resource units, transparently fetching every page.
163    pub fn list_stream(
164        &self,
165        params: ListResourcesParams,
166    ) -> impl Stream<Item = Result<ResourceUnit>> + Send {
167        auto_page(self.fetcher(&params), params.pagination(), "data", None)
168    }
169
170    /// Fetches a resource unit by id.
171    pub async fn get(&self, id: &str) -> Result<ResourceUnit> {
172        let path = format!("{PATH}/{}", escape(id));
173        self.client
174            .data(Method::GET, &path, CallOptions::new())
175            .await
176    }
177
178    /// Pre-acquires composition units.
179    pub async fn acquire(&self, params: AcquireResourceParams) -> Result<Vec<ResourceUnit>> {
180        let path = format!("{PATH}/acquire");
181        self.client
182            .data(Method::POST, &path, CallOptions::json(&params)?)
183            .await
184    }
185
186    /// Releases units by id. Only idle units are released.
187    pub async fn release(&self, ids: &[String]) -> Result<Vec<ReleaseResourceResult>> {
188        let body = serde_json::json!({ "ids": ids });
189        let path = format!("{PATH}/release");
190        self.client
191            .data(Method::POST, &path, CallOptions::json(&body)?)
192            .await
193    }
194
195    fn fetcher(&self, params: &ListResourcesParams) -> PageFetcher {
196        let client = self.client.clone();
197        let status = params.status.clone();
198        Arc::new(move |page, per_page| {
199            let client = client.clone();
200            let status = status.clone();
201            Box::pin(async move {
202                let query = QueryBuilder::new()
203                    .opt("page", page)
204                    .opt("perPage", per_page)
205                    .opt_str("status", status.as_ref().map(ResourceStatus::as_str))
206                    .into_pairs();
207                client
208                    .json::<Value>(Method::GET, PATH, CallOptions::new().query(query))
209                    .await
210            })
211        })
212    }
213}
videosdk/resources/resource_pool.rs

videosdk/resources/
resource_pool.rs
