Skip to main content

gstreamer_app/
app_src.rs

1// Take a look at the license at the top of the repository in the LICENSE file.
2
3use std::{
4    mem, panic,
5    pin::Pin,
6    ptr,
7    sync::{Arc, Mutex},
8    task::{Context, Poll, Waker},
9};
10
11#[cfg(not(panic = "abort"))]
12use std::sync::atomic::{AtomicBool, Ordering};
13
14use futures_sink::Sink;
15use glib::{
16    ffi::{gboolean, gpointer},
17    prelude::*,
18    translate::*,
19};
20
21use crate::{AppSrc, ffi};
22
23#[allow(clippy::type_complexity)]
24pub struct AppSrcCallbacks {
25    need_data: Option<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>,
26    enough_data: Option<Box<dyn Fn(&AppSrc) + Send + Sync + 'static>>,
27    seek_data: Option<Box<dyn Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>>,
28    #[cfg(not(panic = "abort"))]
29    panicked: AtomicBool,
30    callbacks: ffi::GstAppSrcCallbacks,
31}
32
33unsafe impl Send for AppSrcCallbacks {}
34unsafe impl Sync for AppSrcCallbacks {}
35
36impl AppSrcCallbacks {
37    pub fn builder() -> AppSrcCallbacksBuilder {
38        skip_assert_initialized!();
39
40        AppSrcCallbacksBuilder {
41            need_data: None,
42            enough_data: None,
43            seek_data: None,
44        }
45    }
46}
47
48#[allow(clippy::type_complexity)]
49#[must_use = "The builder must be built to be used"]
50pub struct AppSrcCallbacksBuilder {
51    need_data: Option<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>,
52    enough_data: Option<Box<dyn Fn(&AppSrc) + Send + Sync + 'static>>,
53    seek_data: Option<Box<dyn Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>>,
54}
55
56impl AppSrcCallbacksBuilder {
57    pub fn need_data<F: FnMut(&AppSrc, u32) + Send + 'static>(self, need_data: F) -> Self {
58        Self {
59            need_data: Some(Box::new(need_data)),
60            ..self
61        }
62    }
63
64    pub fn need_data_if<F: FnMut(&AppSrc, u32) + Send + 'static>(
65        self,
66        need_data: F,
67        predicate: bool,
68    ) -> Self {
69        if predicate {
70            self.need_data(need_data)
71        } else {
72            self
73        }
74    }
75
76    pub fn need_data_if_some<F: FnMut(&AppSrc, u32) + Send + 'static>(
77        self,
78        need_data: Option<F>,
79    ) -> Self {
80        if let Some(need_data) = need_data {
81            self.need_data(need_data)
82        } else {
83            self
84        }
85    }
86
87    pub fn enough_data<F: Fn(&AppSrc) + Send + Sync + 'static>(self, enough_data: F) -> Self {
88        Self {
89            enough_data: Some(Box::new(enough_data)),
90            ..self
91        }
92    }
93
94    pub fn enough_data_if<F: Fn(&AppSrc) + Send + Sync + 'static>(
95        self,
96        enough_data: F,
97        predicate: bool,
98    ) -> Self {
99        if predicate {
100            self.enough_data(enough_data)
101        } else {
102            self
103        }
104    }
105
106    pub fn enough_data_if_some<F: Fn(&AppSrc) + Send + Sync + 'static>(
107        self,
108        enough_data: Option<F>,
109    ) -> Self {
110        if let Some(enough_data) = enough_data {
111            self.enough_data(enough_data)
112        } else {
113            self
114        }
115    }
116
117    pub fn seek_data<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
118        self,
119        seek_data: F,
120    ) -> Self {
121        Self {
122            seek_data: Some(Box::new(seek_data)),
123            ..self
124        }
125    }
126
127    pub fn seek_data_if<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
128        self,
129        seek_data: F,
130        predicate: bool,
131    ) -> Self {
132        if predicate {
133            self.seek_data(seek_data)
134        } else {
135            self
136        }
137    }
138
139    pub fn seek_data_if_some<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
140        self,
141        seek_data: Option<F>,
142    ) -> Self {
143        if let Some(seek_data) = seek_data {
144            self.seek_data(seek_data)
145        } else {
146            self
147        }
148    }
149
150    #[must_use = "Building the callbacks without using them has no effect"]
151    pub fn build(self) -> AppSrcCallbacks {
152        let have_need_data = self.need_data.is_some();
153        let have_enough_data = self.enough_data.is_some();
154        let have_seek_data = self.seek_data.is_some();
155
156        AppSrcCallbacks {
157            need_data: self.need_data,
158            enough_data: self.enough_data,
159            seek_data: self.seek_data,
160            #[cfg(not(panic = "abort"))]
161            panicked: AtomicBool::new(false),
162            callbacks: ffi::GstAppSrcCallbacks {
163                need_data: if have_need_data {
164                    Some(trampoline_need_data)
165                } else {
166                    None
167                },
168                enough_data: if have_enough_data {
169                    Some(trampoline_enough_data)
170                } else {
171                    None
172                },
173                seek_data: if have_seek_data {
174                    Some(trampoline_seek_data)
175                } else {
176                    None
177                },
178                _gst_reserved: [
179                    ptr::null_mut(),
180                    ptr::null_mut(),
181                    ptr::null_mut(),
182                    ptr::null_mut(),
183                ],
184            },
185        }
186    }
187}
188
189unsafe extern "C" fn trampoline_need_data(
190    appsrc: *mut ffi::GstAppSrc,
191    length: u32,
192    callbacks: gpointer,
193) {
194    unsafe {
195        let callbacks = callbacks as *mut AppSrcCallbacks;
196        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
197
198        #[cfg(not(panic = "abort"))]
199        if (*callbacks).panicked.load(Ordering::Relaxed) {
200            let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
201            gst::subclass::post_panic_error_message(
202                element.upcast_ref(),
203                element.upcast_ref(),
204                None,
205            );
206            return;
207        }
208
209        if let Some(ref mut need_data) = (*callbacks).need_data {
210            let result =
211                panic::catch_unwind(panic::AssertUnwindSafe(|| need_data(&element, length)));
212            match result {
213                Ok(result) => result,
214                Err(err) => {
215                    #[cfg(panic = "abort")]
216                    {
217                        unreachable!("{err:?}");
218                    }
219                    #[cfg(not(panic = "abort"))]
220                    {
221                        (*callbacks).panicked.store(true, Ordering::Relaxed);
222                        gst::subclass::post_panic_error_message(
223                            element.upcast_ref(),
224                            element.upcast_ref(),
225                            Some(err),
226                        );
227                    }
228                }
229            }
230        }
231    }
232}
233
234unsafe extern "C" fn trampoline_enough_data(appsrc: *mut ffi::GstAppSrc, callbacks: gpointer) {
235    unsafe {
236        let callbacks = callbacks as *const AppSrcCallbacks;
237        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
238
239        #[cfg(not(panic = "abort"))]
240        if (*callbacks).panicked.load(Ordering::Relaxed) {
241            let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
242            gst::subclass::post_panic_error_message(
243                element.upcast_ref(),
244                element.upcast_ref(),
245                None,
246            );
247            return;
248        }
249
250        if let Some(ref enough_data) = (*callbacks).enough_data {
251            let result = panic::catch_unwind(panic::AssertUnwindSafe(|| enough_data(&element)));
252            match result {
253                Ok(result) => result,
254                Err(err) => {
255                    #[cfg(panic = "abort")]
256                    {
257                        unreachable!("{err:?}");
258                    }
259                    #[cfg(not(panic = "abort"))]
260                    {
261                        (*callbacks).panicked.store(true, Ordering::Relaxed);
262                        gst::subclass::post_panic_error_message(
263                            element.upcast_ref(),
264                            element.upcast_ref(),
265                            Some(err),
266                        );
267                    }
268                }
269            }
270        }
271    }
272}
273
274unsafe extern "C" fn trampoline_seek_data(
275    appsrc: *mut ffi::GstAppSrc,
276    offset: u64,
277    callbacks: gpointer,
278) -> gboolean {
279    unsafe {
280        let callbacks = callbacks as *const AppSrcCallbacks;
281        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
282
283        #[cfg(not(panic = "abort"))]
284        if (*callbacks).panicked.load(Ordering::Relaxed) {
285            let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
286            gst::subclass::post_panic_error_message(
287                element.upcast_ref(),
288                element.upcast_ref(),
289                None,
290            );
291            return false.into_glib();
292        }
293
294        let ret = if let Some(ref seek_data) = (*callbacks).seek_data {
295            let result =
296                panic::catch_unwind(panic::AssertUnwindSafe(|| seek_data(&element, offset)));
297            match result {
298                Ok(result) => result,
299                Err(err) => {
300                    #[cfg(panic = "abort")]
301                    {
302                        unreachable!("{err:?}");
303                    }
304                    #[cfg(not(panic = "abort"))]
305                    {
306                        (*callbacks).panicked.store(true, Ordering::Relaxed);
307                        gst::subclass::post_panic_error_message(
308                            element.upcast_ref(),
309                            element.upcast_ref(),
310                            Some(err),
311                        );
312
313                        false
314                    }
315                }
316            }
317        } else {
318            false
319        };
320
321        ret.into_glib()
322    }
323}
324
325unsafe extern "C" fn destroy_callbacks(ptr: gpointer) {
326    unsafe {
327        let _ = Box::<AppSrcCallbacks>::from_raw(ptr as *mut _);
328    }
329}
330
331impl AppSrc {
332    // rustdoc-stripper-ignore-next
333    /// Creates a new builder-pattern struct instance to construct [`AppSrc`] objects.
334    ///
335    /// This method returns an instance of [`AppSrcBuilder`](crate::builders::AppSrcBuilder) which can be used to create [`AppSrc`] objects.
336    pub fn builder<'a>() -> AppSrcBuilder<'a> {
337        assert_initialized_main_thread!();
338        AppSrcBuilder {
339            builder: gst::Object::builder(),
340            callbacks: None,
341            automatic_eos: None,
342        }
343    }
344
345    #[doc(alias = "gst_app_src_set_callbacks")]
346    pub fn set_callbacks(&self, callbacks: AppSrcCallbacks) {
347        unsafe {
348            let src = self.to_glib_none().0;
349            #[allow(clippy::manual_dangling_ptr)]
350            #[cfg(not(feature = "v1_18"))]
351            {
352                static SET_ONCE_QUARK: std::sync::OnceLock<glib::Quark> =
353                    std::sync::OnceLock::new();
354
355                let set_once_quark = SET_ONCE_QUARK
356                    .get_or_init(|| glib::Quark::from_str("gstreamer-rs-app-src-callbacks"));
357
358                // This is not thread-safe before 1.16.3, see
359                // https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
360                if gst::version() < (1, 16, 3, 0) {
361                    if !glib::gobject_ffi::g_object_get_qdata(
362                        src as *mut _,
363                        set_once_quark.into_glib(),
364                    )
365                    .is_null()
366                    {
367                        panic!("AppSrc callbacks can only be set once");
368                    }
369
370                    glib::gobject_ffi::g_object_set_qdata(
371                        src as *mut _,
372                        set_once_quark.into_glib(),
373                        1 as *mut _,
374                    );
375                }
376            }
377
378            ffi::gst_app_src_set_callbacks(
379                src,
380                mut_override(&callbacks.callbacks),
381                Box::into_raw(Box::new(callbacks)) as *mut _,
382                Some(destroy_callbacks),
383            );
384        }
385    }
386
387    #[doc(alias = "gst_app_src_set_latency")]
388    pub fn set_latency(
389        &self,
390        min: impl Into<Option<gst::ClockTime>>,
391        max: impl Into<Option<gst::ClockTime>>,
392    ) {
393        unsafe {
394            ffi::gst_app_src_set_latency(
395                self.to_glib_none().0,
396                min.into().into_glib(),
397                max.into().into_glib(),
398            );
399        }
400    }
401
402    #[doc(alias = "get_latency")]
403    #[doc(alias = "gst_app_src_get_latency")]
404    pub fn latency(&self) -> (Option<gst::ClockTime>, Option<gst::ClockTime>) {
405        unsafe {
406            let mut min = mem::MaybeUninit::uninit();
407            let mut max = mem::MaybeUninit::uninit();
408            ffi::gst_app_src_get_latency(self.to_glib_none().0, min.as_mut_ptr(), max.as_mut_ptr());
409            (from_glib(min.assume_init()), from_glib(max.assume_init()))
410        }
411    }
412
413    #[doc(alias = "do-timestamp")]
414    #[doc(alias = "gst_base_src_set_do_timestamp")]
415    pub fn set_do_timestamp(&self, timestamp: bool) {
416        unsafe {
417            gst_base::ffi::gst_base_src_set_do_timestamp(
418                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
419                timestamp.into_glib(),
420            );
421        }
422    }
423
424    #[doc(alias = "do-timestamp")]
425    #[doc(alias = "gst_base_src_get_do_timestamp")]
426    pub fn do_timestamp(&self) -> bool {
427        unsafe {
428            from_glib(gst_base::ffi::gst_base_src_get_do_timestamp(
429                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc
430            ))
431        }
432    }
433
434    #[doc(alias = "do-timestamp")]
435    pub fn connect_do_timestamp_notify<F: Fn(&Self) + Send + Sync + 'static>(
436        &self,
437        f: F,
438    ) -> glib::SignalHandlerId {
439        unsafe extern "C" fn notify_do_timestamp_trampoline<
440            F: Fn(&AppSrc) + Send + Sync + 'static,
441        >(
442            this: *mut ffi::GstAppSrc,
443            _param_spec: glib::ffi::gpointer,
444            f: glib::ffi::gpointer,
445        ) {
446            unsafe {
447                let f: &F = &*(f as *const F);
448                f(&AppSrc::from_glib_borrow(this))
449            }
450        }
451        unsafe {
452            let f: Box<F> = Box::new(f);
453            glib::signal::connect_raw(
454                self.as_ptr() as *mut _,
455                b"notify::do-timestamp\0".as_ptr() as *const _,
456                Some(mem::transmute::<*const (), unsafe extern "C" fn()>(
457                    notify_do_timestamp_trampoline::<F> as *const (),
458                )),
459                Box::into_raw(f),
460            )
461        }
462    }
463
464    #[doc(alias = "set-automatic-eos")]
465    #[doc(alias = "gst_base_src_set_automatic_eos")]
466    pub fn set_automatic_eos(&self, automatic_eos: bool) {
467        unsafe {
468            gst_base::ffi::gst_base_src_set_automatic_eos(
469                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
470                automatic_eos.into_glib(),
471            );
472        }
473    }
474
475    pub fn sink(&self) -> AppSrcSink {
476        AppSrcSink::new(self)
477    }
478}
479
480// rustdoc-stripper-ignore-next
481/// A [builder-pattern] type to construct [`AppSrc`] objects.
482///
483/// [builder-pattern]: https://doc.rust-lang.org/1.0.0/style/ownership/builders.html
484#[must_use = "The builder must be built to be used"]
485pub struct AppSrcBuilder<'a> {
486    builder: gst::gobject::GObjectBuilder<'a, AppSrc>,
487    callbacks: Option<AppSrcCallbacks>,
488    automatic_eos: Option<bool>,
489}
490
491impl<'a> AppSrcBuilder<'a> {
492    // rustdoc-stripper-ignore-next
493    /// Build the [`AppSrc`].
494    ///
495    /// # Panics
496    ///
497    /// This panics if the [`AppSrc`] doesn't have all the given properties or
498    /// property values of the wrong type are provided.
499    #[must_use = "Building the object from the builder is usually expensive and is not expected to have side effects"]
500    pub fn build(self) -> AppSrc {
501        let appsrc = self.builder.build().unwrap();
502
503        if let Some(callbacks) = self.callbacks {
504            appsrc.set_callbacks(callbacks);
505        }
506
507        if let Some(automatic_eos) = self.automatic_eos {
508            appsrc.set_automatic_eos(automatic_eos);
509        }
510
511        appsrc
512    }
513
514    pub fn automatic_eos(self, automatic_eos: bool) -> Self {
515        Self {
516            automatic_eos: Some(automatic_eos),
517            ..self
518        }
519    }
520
521    pub fn block(self, block: bool) -> Self {
522        Self {
523            builder: self.builder.property("block", block),
524            ..self
525        }
526    }
527
528    pub fn callbacks(self, callbacks: AppSrcCallbacks) -> Self {
529        Self {
530            callbacks: Some(callbacks),
531            ..self
532        }
533    }
534
535    pub fn caps(self, caps: &'a gst::Caps) -> Self {
536        Self {
537            builder: self.builder.property("caps", caps),
538            ..self
539        }
540    }
541
542    pub fn do_timestamp(self, do_timestamp: bool) -> Self {
543        Self {
544            builder: self.builder.property("do-timestamp", do_timestamp),
545            ..self
546        }
547    }
548
549    pub fn duration(self, duration: u64) -> Self {
550        Self {
551            builder: self.builder.property("duration", duration),
552            ..self
553        }
554    }
555
556    pub fn format(self, format: gst::Format) -> Self {
557        Self {
558            builder: self.builder.property("format", format),
559            ..self
560        }
561    }
562
563    #[cfg(feature = "v1_18")]
564    #[cfg_attr(docsrs, doc(cfg(feature = "v1_18")))]
565    pub fn handle_segment_change(self, handle_segment_change: bool) -> Self {
566        Self {
567            builder: self
568                .builder
569                .property("handle-segment-change", handle_segment_change),
570            ..self
571        }
572    }
573
574    pub fn is_live(self, is_live: bool) -> Self {
575        Self {
576            builder: self.builder.property("is-live", is_live),
577            ..self
578        }
579    }
580
581    #[cfg(feature = "v1_20")]
582    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
583    pub fn leaky_type(self, leaky_type: crate::AppLeakyType) -> Self {
584        Self {
585            builder: self.builder.property("leaky-type", leaky_type),
586            ..self
587        }
588    }
589
590    #[cfg(feature = "v1_20")]
591    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
592    pub fn max_buffers(self, max_buffers: u64) -> Self {
593        Self {
594            builder: self.builder.property("max-buffers", max_buffers),
595            ..self
596        }
597    }
598
599    pub fn max_bytes(self, max_bytes: u64) -> Self {
600        Self {
601            builder: self.builder.property("max-bytes", max_bytes),
602            ..self
603        }
604    }
605
606    pub fn max_latency(self, max_latency: i64) -> Self {
607        Self {
608            builder: self.builder.property("max-latency", max_latency),
609            ..self
610        }
611    }
612
613    #[cfg(feature = "v1_20")]
614    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
615    pub fn max_time(self, max_time: gst::ClockTime) -> Self {
616        Self {
617            builder: self.builder.property("max-time", max_time),
618            ..self
619        }
620    }
621
622    pub fn min_latency(self, min_latency: i64) -> Self {
623        Self {
624            builder: self.builder.property("min-latency", min_latency),
625            ..self
626        }
627    }
628
629    pub fn min_percent(self, min_percent: u32) -> Self {
630        Self {
631            builder: self.builder.property("min-percent", min_percent),
632            ..self
633        }
634    }
635
636    pub fn size(self, size: i64) -> Self {
637        Self {
638            builder: self.builder.property("size", size),
639            ..self
640        }
641    }
642
643    pub fn stream_type(self, stream_type: crate::AppStreamType) -> Self {
644        Self {
645            builder: self.builder.property("stream-type", stream_type),
646            ..self
647        }
648    }
649
650    #[cfg(feature = "v1_28")]
651    #[cfg_attr(docsrs, doc(cfg(feature = "v1_28")))]
652    pub fn silent(self, silent: bool) -> Self {
653        Self {
654            builder: self.builder.property("silent", silent),
655            ..self
656        }
657    }
658
659    // rustdoc-stripper-ignore-next
660    /// Sets property `name` to the given value `value`.
661    ///
662    /// Overrides any default or previously defined value for `name`.
663    #[inline]
664    pub fn property(self, name: &'a str, value: impl Into<glib::Value> + 'a) -> Self {
665        Self {
666            builder: self.builder.property(name, value),
667            ..self
668        }
669    }
670
671    // rustdoc-stripper-ignore-next
672    /// Sets property `name` to the given string value `value`.
673    #[inline]
674    pub fn property_from_str(self, name: &'a str, value: &'a str) -> Self {
675        Self {
676            builder: self.builder.property_from_str(name, value),
677            ..self
678        }
679    }
680
681    gst::impl_builder_gvalue_extra_setters!(property_and_name);
682}
683
684#[derive(Debug)]
685pub struct AppSrcSink {
686    app_src: glib::WeakRef<AppSrc>,
687    waker_reference: Arc<Mutex<Option<Waker>>>,
688}
689
690impl AppSrcSink {
691    fn new(app_src: &AppSrc) -> Self {
692        skip_assert_initialized!();
693
694        let waker_reference = Arc::new(Mutex::new(None as Option<Waker>));
695
696        app_src.set_callbacks(
697            AppSrcCallbacks::builder()
698                .need_data({
699                    let waker_reference = Arc::clone(&waker_reference);
700
701                    move |_, _| {
702                        if let Some(waker) = waker_reference.lock().unwrap().take() {
703                            waker.wake();
704                        }
705                    }
706                })
707                .build(),
708        );
709
710        Self {
711            app_src: app_src.downgrade(),
712            waker_reference,
713        }
714    }
715}
716
717impl Drop for AppSrcSink {
718    fn drop(&mut self) {
719        #[cfg(not(feature = "v1_18"))]
720        {
721            // This is not thread-safe before 1.16.3, see
722            // https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
723            if gst::version() >= (1, 16, 3, 0)
724                && let Some(app_src) = self.app_src.upgrade()
725            {
726                app_src.set_callbacks(AppSrcCallbacks::builder().build());
727            }
728        }
729    }
730}
731
732impl Sink<gst::Sample> for AppSrcSink {
733    type Error = gst::FlowError;
734
735    fn poll_ready(self: Pin<&mut Self>, context: &mut Context) -> Poll<Result<(), Self::Error>> {
736        let mut waker = self.waker_reference.lock().unwrap();
737
738        let Some(app_src) = self.app_src.upgrade() else {
739            return Poll::Ready(Err(gst::FlowError::Eos));
740        };
741
742        let current_level_bytes = app_src.current_level_bytes();
743        let max_bytes = app_src.max_bytes();
744
745        if current_level_bytes >= max_bytes && max_bytes != 0 {
746            waker.replace(context.waker().to_owned());
747
748            Poll::Pending
749        } else {
750            Poll::Ready(Ok(()))
751        }
752    }
753
754    fn start_send(self: Pin<&mut Self>, sample: gst::Sample) -> Result<(), Self::Error> {
755        let Some(app_src) = self.app_src.upgrade() else {
756            return Err(gst::FlowError::Eos);
757        };
758
759        app_src.push_sample(&sample)?;
760
761        Ok(())
762    }
763
764    fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
765        Poll::Ready(Ok(()))
766    }
767
768    fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
769        let Some(app_src) = self.app_src.upgrade() else {
770            return Poll::Ready(Ok(()));
771        };
772
773        app_src.end_of_stream()?;
774
775        Poll::Ready(Ok(()))
776    }
777}
778
779#[cfg(test)]
780mod tests {
781    use std::sync::atomic::{AtomicUsize, Ordering};
782
783    use futures_util::{sink::SinkExt, stream::StreamExt};
784    use gst::prelude::*;
785
786    use super::*;
787
788    #[test]
789    fn test_app_src_sink() {
790        gst::init().unwrap();
791
792        let appsrc = gst::ElementFactory::make("appsrc").build().unwrap();
793        let fakesink = gst::ElementFactory::make("fakesink")
794            .property("signal-handoffs", true)
795            .build()
796            .unwrap();
797
798        let pipeline = gst::Pipeline::new();
799        pipeline.add(&appsrc).unwrap();
800        pipeline.add(&fakesink).unwrap();
801
802        appsrc.link(&fakesink).unwrap();
803
804        let mut bus_stream = pipeline.bus().unwrap().stream();
805        let mut app_src_sink = appsrc.dynamic_cast::<AppSrc>().unwrap().sink();
806
807        let sample_quantity = 5;
808
809        let samples = (0..sample_quantity)
810            .map(|_| gst::Sample::builder().buffer(&gst::Buffer::new()).build())
811            .collect::<Vec<gst::Sample>>();
812
813        let mut sample_stream = futures_util::stream::iter(samples).map(Ok);
814
815        let handoff_count_reference = Arc::new(AtomicUsize::new(0));
816
817        fakesink.connect("handoff", false, {
818            let handoff_count_reference = Arc::clone(&handoff_count_reference);
819
820            move |_| {
821                handoff_count_reference.fetch_add(1, Ordering::AcqRel);
822
823                None
824            }
825        });
826
827        pipeline.set_state(gst::State::Playing).unwrap();
828
829        futures_executor::block_on(app_src_sink.send_all(&mut sample_stream)).unwrap();
830        futures_executor::block_on(app_src_sink.close()).unwrap();
831
832        while let Some(message) = futures_executor::block_on(bus_stream.next()) {
833            match message.view() {
834                gst::MessageView::Eos(_) => break,
835                gst::MessageView::Error(_) => unreachable!(),
836                _ => continue,
837            }
838        }
839
840        pipeline.set_state(gst::State::Null).unwrap();
841
842        assert_eq!(
843            handoff_count_reference.load(Ordering::Acquire),
844            sample_quantity
845        );
846    }
847
848    #[test]
849    fn builder_caps_lt() {
850        gst::init().unwrap();
851
852        let caps = &gst::Caps::new_any();
853        {
854            let stream_type = "random-access".to_owned();
855            let appsrc = AppSrc::builder()
856                .property_from_str("stream-type", &stream_type)
857                .caps(caps)
858                .build();
859            assert_eq!(
860                appsrc.property::<crate::AppStreamType>("stream-type"),
861                crate::AppStreamType::RandomAccess
862            );
863            assert!(appsrc.property::<gst::Caps>("caps").is_any());
864        }
865
866        let stream_type = &"random-access".to_owned();
867        {
868            let caps = &gst::Caps::new_any();
869            let appsrc = AppSrc::builder()
870                .property_from_str("stream-type", stream_type)
871                .caps(caps)
872                .build();
873            assert_eq!(
874                appsrc.property::<crate::AppStreamType>("stream-type"),
875                crate::AppStreamType::RandomAccess
876            );
877            assert!(appsrc.property::<gst::Caps>("caps").is_any());
878        }
879    }
880}