gst_client/
client.rs

1//! Defines [`GstClient`] for communication with
2//! [`GStreamer Daemon`][1] API.
3//!
4//! [1]: https://developer.ridgerun.com/wiki/index.php/GStreamer_Daemon
5use crate::{gstd_types, resources, Error};
6use reqwest::{Client, Response};
7use url::Url;
8
9/// [`GstClient`] for [`GStreamer Daemon`][1] API.
10///
11/// [1]: https://developer.ridgerun.com/wiki/index.php/GStreamer_Daemon
12#[derive(Debug, Clone)]
13pub struct GstClient {
14    http_client: Client,
15    pub(crate) base_url: Url,
16}
17
18impl GstClient {
19    /// Build [`GstClient`] for future call to [`GStreamer Daemon`][1] API.
20    ///
21    /// # Errors
22    ///
23    /// If incorrect `base_url` passed
24    ///
25    /// [1]: https://developer.ridgerun.com/wiki/index.php/GStreamer_Daemon
26    pub fn build<S: Into<String>>(base_url: S) -> Result<Self, Error> {
27        Ok(Self {
28            http_client: Client::new(),
29            base_url: Url::parse(&base_url.into()).map_err(Error::IncorrectBaseUrl)?,
30        })
31    }
32
33    pub(crate) async fn get(&self, url: &str) -> Result<Response, Error> {
34        self.http_client
35            .get(self.base_url.join(url).map_err(Error::IncorrectApiUrl)?)
36            .send()
37            .await
38            .map_err(Error::RequestFailed)
39    }
40
41    pub(crate) async fn post(&self, url: &str) -> Result<Response, Error> {
42        self.http_client
43            .post(self.base_url.join(url).map_err(Error::IncorrectApiUrl)?)
44            .send()
45            .await
46            .map_err(Error::RequestFailed)
47    }
48
49    pub(crate) async fn put(&self, url: &str) -> Result<Response, Error> {
50        self.http_client
51            .put(self.base_url.join(url).map_err(Error::IncorrectApiUrl)?)
52            .send()
53            .await
54            .map_err(Error::RequestFailed)
55    }
56
57    pub(crate) async fn delete(&self, url: &str) -> Result<Response, Error> {
58        self.http_client
59            .delete(self.base_url.join(url).map_err(Error::IncorrectApiUrl)?)
60            .send()
61            .await
62            .map_err(Error::RequestFailed)
63    }
64
65    pub(crate) async fn process_resp(
66        &self,
67        resp: Response,
68    ) -> Result<gstd_types::SuccessResponse, Error> {
69        let api_response = resp
70            .json::<gstd_types::ApiResponse>()
71            .await
72            .map_err(Error::RequestFailed)?;
73        match api_response {
74            gstd_types::ApiResponse::Success(resp) => Ok(resp),
75            gstd_types::ApiResponse::Error(resp) => {
76                Err(Error::GstdError(resp.code, resp.description))
77            }
78        }
79    }
80
81    /// Performs `GET /pipelines` API request, returning the
82    /// parsed [`gstd_types::SuccessResponse`]
83    ///
84    /// # Errors
85    ///
86    /// If API request cannot be performed, or fails.
87    /// See [`Error`] for details.
88    pub async fn pipelines(&self) -> Result<gstd_types::SuccessResponse, Error> {
89        let resp = self.get("pipelines").await?;
90        self.process_resp(resp).await
91    }
92    /// Operate with [`GStreamer Daemon`][1] pipelines.
93    ///
94    /// # Arguments
95    ///
96    /// * `name` - name of the pipeline
97    ///
98    /// [1]: https://developer.ridgerun.com/wiki/index.php/GStreamer_Daemon
99    #[must_use]
100    pub fn pipeline<S>(&self, name: S) -> resources::Pipeline
101    where
102        S: Into<String>,
103    {
104        resources::Pipeline::new(name, self)
105    }
106    /// Manage [`GStreamer Daemon`][1] Debug mode.
107    ///
108    /// [1]: https://developer.ridgerun.com/wiki/index.php/GStreamer_Daemon
109    #[must_use]
110    pub fn debug(&self) -> resources::Debug {
111        resources::Debug::new(self)
112    }
113}
114
115impl Default for GstClient {
116    fn default() -> Self {
117        Self {
118            http_client: Client::new(),
119            base_url: Url::parse("http://127.0.0.1:5000").unwrap(),
120        }
121    }
122}
123
124impl From<Url> for GstClient {
125    fn from(url: Url) -> Self {
126        Self {
127            http_client: Client::new(),
128            base_url: url,
129        }
130    }
131}
132
133impl From<&Url> for GstClient {
134    fn from(url: &Url) -> Self {
135        Self {
136            http_client: Client::new(),
137            base_url: url.clone(),
138        }
139    }
140}
141
142#[cfg(test)]
143mod spec {
144    use super::*;
145    const BASE_URL: &'static str = "http://localhost:8080";
146    const PIPELINE_NAME: &'static str = "test pipeline";
147
148    fn expect_url() -> Url {
149        Url::parse(BASE_URL).unwrap()
150    }
151
152    #[test]
153    fn create_client_with_build() {
154        let client = GstClient::build(BASE_URL).unwrap();
155        assert_eq!(client.base_url, expect_url());
156
157        let client = GstClient::build(BASE_URL.to_string()).unwrap();
158        assert_eq!(client.base_url, expect_url());
159    }
160
161    #[test]
162    fn create_client_from() {
163        let url = expect_url();
164        let client = GstClient::from(&url);
165        assert_eq!(client.base_url, expect_url());
166
167        let client = GstClient::from(url);
168        assert_eq!(client.base_url, expect_url());
169    }
170    #[tokio::test]
171    async fn create_pipeline() {
172        if let Ok(client) = GstClient::build(BASE_URL) {
173            let res = client.pipeline(PIPELINE_NAME).create("").await;
174            println!("{:?}", res);
175            assert!(res.is_ok());
176        };
177    }
178
179    #[tokio::test]
180    async fn retrieve_pipelines() {
181        if let Ok(client) = GstClient::build(BASE_URL) {
182            let res = client.pipelines().await;
183            println!("{:?}", res);
184            assert!(res.is_ok());
185        };
186    }
187
188    #[tokio::test]
189    async fn retrieve_pipeline_graph() {
190        if let Ok(client) = GstClient::build(BASE_URL) {
191            let res = client.pipeline(PIPELINE_NAME).graph().await;
192            println!("{:?}", res);
193            assert!(res.is_ok());
194        };
195    }
196
197    #[tokio::test]
198    async fn retrieve_pipeline_elements() {
199        if let Ok(client) = GstClient::build(BASE_URL) {
200            let res = client.pipeline(PIPELINE_NAME).elements().await;
201            println!("{:?}", res);
202            assert!(res.is_ok());
203        };
204    }
205    #[tokio::test]
206    async fn retrieve_pipeline_properties() {
207        if let Ok(client) = GstClient::build(BASE_URL) {
208            let res = client.pipeline(PIPELINE_NAME).properties().await;
209            println!("{:?}", res);
210            assert!(res.is_ok());
211        };
212    }
213    #[tokio::test]
214    async fn retrieve_pipeline_element_property() {
215        if let Ok(client) = GstClient::build(BASE_URL) {
216            let res = client
217                .pipeline(PIPELINE_NAME)
218                .element("rtmp2src")
219                .property("location")
220                .await;
221            println!("{:?}", res);
222            assert!(res.is_ok());
223        };
224    }
225    #[tokio::test]
226    async fn retrieve_pipeline_bus_read() {
227        if let Ok(client) = GstClient::build(BASE_URL) {
228            let res = client.pipeline(PIPELINE_NAME).bus().read().await;
229            println!("{:?}", res);
230            assert!(res.is_ok());
231        };
232    }
233    #[tokio::test]
234    async fn delete_pipeline() {
235        if let Ok(client) = GstClient::build(BASE_URL) {
236            let res = client.pipeline(PIPELINE_NAME).delete().await;
237            println!("{:?}", res);
238            assert!(res.is_ok());
239        };
240    }
241}
gst_client/client.rs

gst_client/
client.rs
