1use 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::{ffi, AppSrc};
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    let callbacks = callbacks as *mut AppSrcCallbacks;
195    let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
196
197    #[cfg(not(panic = "abort"))]
198    if (*callbacks).panicked.load(Ordering::Relaxed) {
199        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
200        gst::subclass::post_panic_error_message(element.upcast_ref(), element.upcast_ref(), None);
201        return;
202    }
203
204    if let Some(ref mut need_data) = (*callbacks).need_data {
205        let result = panic::catch_unwind(panic::AssertUnwindSafe(|| need_data(&element, length)));
206        match result {
207            Ok(result) => result,
208            Err(err) => {
209                #[cfg(panic = "abort")]
210                {
211                    unreachable!("{err:?}");
212                }
213                #[cfg(not(panic = "abort"))]
214                {
215                    (*callbacks).panicked.store(true, Ordering::Relaxed);
216                    gst::subclass::post_panic_error_message(
217                        element.upcast_ref(),
218                        element.upcast_ref(),
219                        Some(err),
220                    );
221                }
222            }
223        }
224    }
225}
226
227unsafe extern "C" fn trampoline_enough_data(appsrc: *mut ffi::GstAppSrc, callbacks: gpointer) {
228    let callbacks = callbacks as *const AppSrcCallbacks;
229    let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
230
231    #[cfg(not(panic = "abort"))]
232    if (*callbacks).panicked.load(Ordering::Relaxed) {
233        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
234        gst::subclass::post_panic_error_message(element.upcast_ref(), element.upcast_ref(), None);
235        return;
236    }
237
238    if let Some(ref enough_data) = (*callbacks).enough_data {
239        let result = panic::catch_unwind(panic::AssertUnwindSafe(|| enough_data(&element)));
240        match result {
241            Ok(result) => result,
242            Err(err) => {
243                #[cfg(panic = "abort")]
244                {
245                    unreachable!("{err:?}");
246                }
247                #[cfg(not(panic = "abort"))]
248                {
249                    (*callbacks).panicked.store(true, Ordering::Relaxed);
250                    gst::subclass::post_panic_error_message(
251                        element.upcast_ref(),
252                        element.upcast_ref(),
253                        Some(err),
254                    );
255                }
256            }
257        }
258    }
259}
260
261unsafe extern "C" fn trampoline_seek_data(
262    appsrc: *mut ffi::GstAppSrc,
263    offset: u64,
264    callbacks: gpointer,
265) -> gboolean {
266    let callbacks = callbacks as *const AppSrcCallbacks;
267    let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
268
269    #[cfg(not(panic = "abort"))]
270    if (*callbacks).panicked.load(Ordering::Relaxed) {
271        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
272        gst::subclass::post_panic_error_message(element.upcast_ref(), element.upcast_ref(), None);
273        return false.into_glib();
274    }
275
276    let ret = if let Some(ref seek_data) = (*callbacks).seek_data {
277        let result = panic::catch_unwind(panic::AssertUnwindSafe(|| seek_data(&element, offset)));
278        match result {
279            Ok(result) => result,
280            Err(err) => {
281                #[cfg(panic = "abort")]
282                {
283                    unreachable!("{err:?}");
284                }
285                #[cfg(not(panic = "abort"))]
286                {
287                    (*callbacks).panicked.store(true, Ordering::Relaxed);
288                    gst::subclass::post_panic_error_message(
289                        element.upcast_ref(),
290                        element.upcast_ref(),
291                        Some(err),
292                    );
293
294                    false
295                }
296            }
297        }
298    } else {
299        false
300    };
301
302    ret.into_glib()
303}
304
305unsafe extern "C" fn destroy_callbacks(ptr: gpointer) {
306    let _ = Box::<AppSrcCallbacks>::from_raw(ptr as *mut _);
307}
308
309impl AppSrc {
310    pub fn builder<'a>() -> AppSrcBuilder<'a> {
315        assert_initialized_main_thread!();
316        AppSrcBuilder {
317            builder: gst::Object::builder(),
318            callbacks: None,
319            automatic_eos: None,
320        }
321    }
322
323    #[doc(alias = "gst_app_src_set_callbacks")]
324    pub fn set_callbacks(&self, callbacks: AppSrcCallbacks) {
325        unsafe {
326            let src = self.to_glib_none().0;
327            #[cfg(not(feature = "v1_18"))]
328            {
329                static SET_ONCE_QUARK: std::sync::OnceLock<glib::Quark> =
330                    std::sync::OnceLock::new();
331
332                let set_once_quark = SET_ONCE_QUARK
333                    .get_or_init(|| glib::Quark::from_str("gstreamer-rs-app-src-callbacks"));
334
335                if gst::version() < (1, 16, 3, 0) {
338                    if !glib::gobject_ffi::g_object_get_qdata(
339                        src as *mut _,
340                        set_once_quark.into_glib(),
341                    )
342                    .is_null()
343                    {
344                        panic!("AppSrc callbacks can only be set once");
345                    }
346
347                    glib::gobject_ffi::g_object_set_qdata(
348                        src as *mut _,
349                        set_once_quark.into_glib(),
350                        1 as *mut _,
351                    );
352                }
353            }
354
355            ffi::gst_app_src_set_callbacks(
356                src,
357                mut_override(&callbacks.callbacks),
358                Box::into_raw(Box::new(callbacks)) as *mut _,
359                Some(destroy_callbacks),
360            );
361        }
362    }
363
364    #[doc(alias = "gst_app_src_set_latency")]
365    pub fn set_latency(
366        &self,
367        min: impl Into<Option<gst::ClockTime>>,
368        max: impl Into<Option<gst::ClockTime>>,
369    ) {
370        unsafe {
371            ffi::gst_app_src_set_latency(
372                self.to_glib_none().0,
373                min.into().into_glib(),
374                max.into().into_glib(),
375            );
376        }
377    }
378
379    #[doc(alias = "get_latency")]
380    #[doc(alias = "gst_app_src_get_latency")]
381    pub fn latency(&self) -> (Option<gst::ClockTime>, Option<gst::ClockTime>) {
382        unsafe {
383            let mut min = mem::MaybeUninit::uninit();
384            let mut max = mem::MaybeUninit::uninit();
385            ffi::gst_app_src_get_latency(self.to_glib_none().0, min.as_mut_ptr(), max.as_mut_ptr());
386            (from_glib(min.assume_init()), from_glib(max.assume_init()))
387        }
388    }
389
390    #[doc(alias = "do-timestamp")]
391    #[doc(alias = "gst_base_src_set_do_timestamp")]
392    pub fn set_do_timestamp(&self, timestamp: bool) {
393        unsafe {
394            gst_base::ffi::gst_base_src_set_do_timestamp(
395                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
396                timestamp.into_glib(),
397            );
398        }
399    }
400
401    #[doc(alias = "do-timestamp")]
402    #[doc(alias = "gst_base_src_get_do_timestamp")]
403    pub fn do_timestamp(&self) -> bool {
404        unsafe {
405            from_glib(gst_base::ffi::gst_base_src_get_do_timestamp(
406                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc
407            ))
408        }
409    }
410
411    #[doc(alias = "do-timestamp")]
412    pub fn connect_do_timestamp_notify<F: Fn(&Self) + Send + Sync + 'static>(
413        &self,
414        f: F,
415    ) -> glib::SignalHandlerId {
416        unsafe extern "C" fn notify_do_timestamp_trampoline<
417            F: Fn(&AppSrc) + Send + Sync + 'static,
418        >(
419            this: *mut ffi::GstAppSrc,
420            _param_spec: glib::ffi::gpointer,
421            f: glib::ffi::gpointer,
422        ) {
423            let f: &F = &*(f as *const F);
424            f(&AppSrc::from_glib_borrow(this))
425        }
426        unsafe {
427            let f: Box<F> = Box::new(f);
428            glib::signal::connect_raw(
429                self.as_ptr() as *mut _,
430                b"notify::do-timestamp\0".as_ptr() as *const _,
431                Some(mem::transmute::<*const (), unsafe extern "C" fn()>(
432                    notify_do_timestamp_trampoline::<F> as *const (),
433                )),
434                Box::into_raw(f),
435            )
436        }
437    }
438
439    #[doc(alias = "set-automatic-eos")]
440    #[doc(alias = "gst_base_src_set_automatic_eos")]
441    pub fn set_automatic_eos(&self, automatic_eos: bool) {
442        unsafe {
443            gst_base::ffi::gst_base_src_set_automatic_eos(
444                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
445                automatic_eos.into_glib(),
446            );
447        }
448    }
449
450    pub fn sink(&self) -> AppSrcSink {
451        AppSrcSink::new(self)
452    }
453}
454
455#[must_use = "The builder must be built to be used"]
460pub struct AppSrcBuilder<'a> {
461    builder: gst::gobject::GObjectBuilder<'a, AppSrc>,
462    callbacks: Option<AppSrcCallbacks>,
463    automatic_eos: Option<bool>,
464}
465
466impl<'a> AppSrcBuilder<'a> {
467    #[must_use = "Building the object from the builder is usually expensive and is not expected to have side effects"]
475    pub fn build(self) -> AppSrc {
476        let appsrc = self.builder.build().unwrap();
477
478        if let Some(callbacks) = self.callbacks {
479            appsrc.set_callbacks(callbacks);
480        }
481
482        if let Some(automatic_eos) = self.automatic_eos {
483            appsrc.set_automatic_eos(automatic_eos);
484        }
485
486        appsrc
487    }
488
489    pub fn automatic_eos(self, automatic_eos: bool) -> Self {
490        Self {
491            automatic_eos: Some(automatic_eos),
492            ..self
493        }
494    }
495
496    pub fn block(self, block: bool) -> Self {
497        Self {
498            builder: self.builder.property("block", block),
499            ..self
500        }
501    }
502
503    pub fn callbacks(self, callbacks: AppSrcCallbacks) -> Self {
504        Self {
505            callbacks: Some(callbacks),
506            ..self
507        }
508    }
509
510    pub fn caps(self, caps: &'a gst::Caps) -> Self {
511        Self {
512            builder: self.builder.property("caps", caps),
513            ..self
514        }
515    }
516
517    pub fn do_timestamp(self, do_timestamp: bool) -> Self {
518        Self {
519            builder: self.builder.property("do-timestamp", do_timestamp),
520            ..self
521        }
522    }
523
524    pub fn duration(self, duration: u64) -> Self {
525        Self {
526            builder: self.builder.property("duration", duration),
527            ..self
528        }
529    }
530
531    pub fn format(self, format: gst::Format) -> Self {
532        Self {
533            builder: self.builder.property("format", format),
534            ..self
535        }
536    }
537
538    #[cfg(feature = "v1_18")]
539    #[cfg_attr(docsrs, doc(cfg(feature = "v1_18")))]
540    pub fn handle_segment_change(self, handle_segment_change: bool) -> Self {
541        Self {
542            builder: self
543                .builder
544                .property("handle-segment-change", handle_segment_change),
545            ..self
546        }
547    }
548
549    pub fn is_live(self, is_live: bool) -> Self {
550        Self {
551            builder: self.builder.property("is-live", is_live),
552            ..self
553        }
554    }
555
556    #[cfg(feature = "v1_20")]
557    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
558    pub fn leaky_type(self, leaky_type: crate::AppLeakyType) -> Self {
559        Self {
560            builder: self.builder.property("leaky-type", leaky_type),
561            ..self
562        }
563    }
564
565    #[cfg(feature = "v1_20")]
566    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
567    pub fn max_buffers(self, max_buffers: u64) -> Self {
568        Self {
569            builder: self.builder.property("max-buffers", max_buffers),
570            ..self
571        }
572    }
573
574    pub fn max_bytes(self, max_bytes: u64) -> Self {
575        Self {
576            builder: self.builder.property("max-bytes", max_bytes),
577            ..self
578        }
579    }
580
581    pub fn max_latency(self, max_latency: i64) -> Self {
582        Self {
583            builder: self.builder.property("max-latency", max_latency),
584            ..self
585        }
586    }
587
588    #[cfg(feature = "v1_20")]
589    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
590    pub fn max_time(self, max_time: Option<gst::ClockTime>) -> Self {
591        Self {
592            builder: self.builder.property("max-time", max_time),
593            ..self
594        }
595    }
596
597    pub fn min_latency(self, min_latency: i64) -> Self {
598        Self {
599            builder: self.builder.property("min-latency", min_latency),
600            ..self
601        }
602    }
603
604    pub fn min_percent(self, min_percent: u32) -> Self {
605        Self {
606            builder: self.builder.property("min-percent", min_percent),
607            ..self
608        }
609    }
610
611    pub fn size(self, size: i64) -> Self {
612        Self {
613            builder: self.builder.property("size", size),
614            ..self
615        }
616    }
617
618    pub fn stream_type(self, stream_type: crate::AppStreamType) -> Self {
619        Self {
620            builder: self.builder.property("stream-type", stream_type),
621            ..self
622        }
623    }
624
625    #[cfg(feature = "v1_28")]
626    #[cfg_attr(docsrs, doc(cfg(feature = "v1_28")))]
627    pub fn silent(self, silent: bool) -> Self {
628        Self {
629            builder: self.builder.property("silent", silent),
630            ..self
631        }
632    }
633
634    #[inline]
639    pub fn property(self, name: &'a str, value: impl Into<glib::Value> + 'a) -> Self {
640        Self {
641            builder: self.builder.property(name, value),
642            ..self
643        }
644    }
645
646    #[inline]
649    pub fn property_from_str(self, name: &'a str, value: &'a str) -> Self {
650        Self {
651            builder: self.builder.property_from_str(name, value),
652            ..self
653        }
654    }
655
656    gst::impl_builder_gvalue_extra_setters!(property_and_name);
657}
658
659#[derive(Debug)]
660pub struct AppSrcSink {
661    app_src: glib::WeakRef<AppSrc>,
662    waker_reference: Arc<Mutex<Option<Waker>>>,
663}
664
665impl AppSrcSink {
666    fn new(app_src: &AppSrc) -> Self {
667        skip_assert_initialized!();
668
669        let waker_reference = Arc::new(Mutex::new(None as Option<Waker>));
670
671        app_src.set_callbacks(
672            AppSrcCallbacks::builder()
673                .need_data({
674                    let waker_reference = Arc::clone(&waker_reference);
675
676                    move |_, _| {
677                        if let Some(waker) = waker_reference.lock().unwrap().take() {
678                            waker.wake();
679                        }
680                    }
681                })
682                .build(),
683        );
684
685        Self {
686            app_src: app_src.downgrade(),
687            waker_reference,
688        }
689    }
690}
691
692impl Drop for AppSrcSink {
693    fn drop(&mut self) {
694        #[cfg(not(feature = "v1_18"))]
695        {
696            if gst::version() >= (1, 16, 3, 0) {
699                if let Some(app_src) = self.app_src.upgrade() {
700                    app_src.set_callbacks(AppSrcCallbacks::builder().build());
701                }
702            }
703        }
704    }
705}
706
707impl Sink<gst::Sample> for AppSrcSink {
708    type Error = gst::FlowError;
709
710    fn poll_ready(self: Pin<&mut Self>, context: &mut Context) -> Poll<Result<(), Self::Error>> {
711        let mut waker = self.waker_reference.lock().unwrap();
712
713        let Some(app_src) = self.app_src.upgrade() else {
714            return Poll::Ready(Err(gst::FlowError::Eos));
715        };
716
717        let current_level_bytes = app_src.current_level_bytes();
718        let max_bytes = app_src.max_bytes();
719
720        if current_level_bytes >= max_bytes && max_bytes != 0 {
721            waker.replace(context.waker().to_owned());
722
723            Poll::Pending
724        } else {
725            Poll::Ready(Ok(()))
726        }
727    }
728
729    fn start_send(self: Pin<&mut Self>, sample: gst::Sample) -> Result<(), Self::Error> {
730        let Some(app_src) = self.app_src.upgrade() else {
731            return Err(gst::FlowError::Eos);
732        };
733
734        app_src.push_sample(&sample)?;
735
736        Ok(())
737    }
738
739    fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
740        Poll::Ready(Ok(()))
741    }
742
743    fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
744        let Some(app_src) = self.app_src.upgrade() else {
745            return Poll::Ready(Ok(()));
746        };
747
748        app_src.end_of_stream()?;
749
750        Poll::Ready(Ok(()))
751    }
752}
753
754#[cfg(test)]
755mod tests {
756    use std::sync::atomic::{AtomicUsize, Ordering};
757
758    use futures_util::{sink::SinkExt, stream::StreamExt};
759    use gst::prelude::*;
760
761    use super::*;
762
763    #[test]
764    fn test_app_src_sink() {
765        gst::init().unwrap();
766
767        let appsrc = gst::ElementFactory::make("appsrc").build().unwrap();
768        let fakesink = gst::ElementFactory::make("fakesink")
769            .property("signal-handoffs", true)
770            .build()
771            .unwrap();
772
773        let pipeline = gst::Pipeline::new();
774        pipeline.add(&appsrc).unwrap();
775        pipeline.add(&fakesink).unwrap();
776
777        appsrc.link(&fakesink).unwrap();
778
779        let mut bus_stream = pipeline.bus().unwrap().stream();
780        let mut app_src_sink = appsrc.dynamic_cast::<AppSrc>().unwrap().sink();
781
782        let sample_quantity = 5;
783
784        let samples = (0..sample_quantity)
785            .map(|_| gst::Sample::builder().buffer(&gst::Buffer::new()).build())
786            .collect::<Vec<gst::Sample>>();
787
788        let mut sample_stream = futures_util::stream::iter(samples).map(Ok);
789
790        let handoff_count_reference = Arc::new(AtomicUsize::new(0));
791
792        fakesink.connect("handoff", false, {
793            let handoff_count_reference = Arc::clone(&handoff_count_reference);
794
795            move |_| {
796                handoff_count_reference.fetch_add(1, Ordering::AcqRel);
797
798                None
799            }
800        });
801
802        pipeline.set_state(gst::State::Playing).unwrap();
803
804        futures_executor::block_on(app_src_sink.send_all(&mut sample_stream)).unwrap();
805        futures_executor::block_on(app_src_sink.close()).unwrap();
806
807        while let Some(message) = futures_executor::block_on(bus_stream.next()) {
808            match message.view() {
809                gst::MessageView::Eos(_) => break,
810                gst::MessageView::Error(_) => unreachable!(),
811                _ => continue,
812            }
813        }
814
815        pipeline.set_state(gst::State::Null).unwrap();
816
817        assert_eq!(
818            handoff_count_reference.load(Ordering::Acquire),
819            sample_quantity
820        );
821    }
822
823    #[test]
824    fn builder_caps_lt() {
825        gst::init().unwrap();
826
827        let caps = &gst::Caps::new_any();
828        {
829            let stream_type = "random-access".to_owned();
830            let appsrc = AppSrc::builder()
831                .property_from_str("stream-type", &stream_type)
832                .caps(caps)
833                .build();
834            assert_eq!(
835                appsrc.property::<crate::AppStreamType>("stream-type"),
836                crate::AppStreamType::RandomAccess
837            );
838            assert!(appsrc.property::<gst::Caps>("caps").is_any());
839        }
840
841        let stream_type = &"random-access".to_owned();
842        {
843            let caps = &gst::Caps::new_any();
844            let appsrc = AppSrc::builder()
845                .property_from_str("stream-type", stream_type)
846                .caps(caps)
847                .build();
848            assert_eq!(
849                appsrc.property::<crate::AppStreamType>("stream-type"),
850                crate::AppStreamType::RandomAccess
851            );
852            assert!(appsrc.property::<gst::Caps>("caps").is_any());
853        }
854    }
855}