gstreamer_app/
app_src.rs

Help
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::{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    // rustdoc-stripper-ignore-next
311    /// Creates a new builder-pattern struct instance to construct [`AppSrc`] objects.
312    ///
313    /// This method returns an instance of [`AppSrcBuilder`](crate::builders::AppSrcBuilder) which can be used to create [`AppSrc`] objects.
314    pub fn builder() -> AppSrcBuilder {
315        assert_initialized_main_thread!();
316        AppSrcBuilder::new()
317    }
318
319    /// Set callbacks which will be executed when data is needed, enough data has
320    /// been collected or when a seek should be performed.
321    /// This is an alternative to using the signals, it has lower overhead and is thus
322    /// less expensive, but also less flexible.
323    ///
324    /// If callbacks are installed, no signals will be emitted for performance
325    /// reasons.
326    ///
327    /// Before 1.16.3 it was not possible to change the callbacks in a thread-safe
328    /// way.
329    /// ## `callbacks`
330    /// the callbacks
331    /// ## `notify`
332    /// a destroy notify function
333    #[doc(alias = "gst_app_src_set_callbacks")]
334    pub fn set_callbacks(&self, callbacks: AppSrcCallbacks) {
335        unsafe {
336            let src = self.to_glib_none().0;
337            #[cfg(not(feature = "v1_18"))]
338            {
339                static SET_ONCE_QUARK: std::sync::OnceLock<glib::Quark> =
340                    std::sync::OnceLock::new();
341
342                let set_once_quark = SET_ONCE_QUARK
343                    .get_or_init(|| glib::Quark::from_str("gstreamer-rs-app-src-callbacks"));
344
345                // This is not thread-safe before 1.16.3, see
346                // https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
347                if gst::version() < (1, 16, 3, 0) {
348                    if !glib::gobject_ffi::g_object_get_qdata(
349                        src as *mut _,
350                        set_once_quark.into_glib(),
351                    )
352                    .is_null()
353                    {
354                        panic!("AppSrc callbacks can only be set once");
355                    }
356
357                    glib::gobject_ffi::g_object_set_qdata(
358                        src as *mut _,
359                        set_once_quark.into_glib(),
360                        1 as *mut _,
361                    );
362                }
363            }
364
365            ffi::gst_app_src_set_callbacks(
366                src,
367                mut_override(&callbacks.callbacks),
368                Box::into_raw(Box::new(callbacks)) as *mut _,
369                Some(destroy_callbacks),
370            );
371        }
372    }
373
374    /// Configure the `min` and `max` latency in `src`. If `min` is set to -1, the
375    /// default latency calculations for pseudo-live sources will be used.
376    /// ## `min`
377    /// the min latency
378    /// ## `max`
379    /// the max latency
380    #[doc(alias = "gst_app_src_set_latency")]
381    pub fn set_latency(
382        &self,
383        min: impl Into<Option<gst::ClockTime>>,
384        max: impl Into<Option<gst::ClockTime>>,
385    ) {
386        unsafe {
387            ffi::gst_app_src_set_latency(
388                self.to_glib_none().0,
389                min.into().into_glib(),
390                max.into().into_glib(),
391            );
392        }
393    }
394
395    /// Retrieve the min and max latencies in `min` and `max` respectively.
396    ///
397    /// # Returns
398    ///
399    ///
400    /// ## `min`
401    /// the min latency
402    ///
403    /// ## `max`
404    /// the max latency
405    #[doc(alias = "get_latency")]
406    #[doc(alias = "gst_app_src_get_latency")]
407    pub fn latency(&self) -> (Option<gst::ClockTime>, Option<gst::ClockTime>) {
408        unsafe {
409            let mut min = mem::MaybeUninit::uninit();
410            let mut max = mem::MaybeUninit::uninit();
411            ffi::gst_app_src_get_latency(self.to_glib_none().0, min.as_mut_ptr(), max.as_mut_ptr());
412            (from_glib(min.assume_init()), from_glib(max.assume_init()))
413        }
414    }
415
416    #[doc(alias = "do-timestamp")]
417    #[doc(alias = "gst_base_src_set_do_timestamp")]
418    pub fn set_do_timestamp(&self, timestamp: bool) {
419        unsafe {
420            gst_base::ffi::gst_base_src_set_do_timestamp(
421                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
422                timestamp.into_glib(),
423            );
424        }
425    }
426
427    #[doc(alias = "do-timestamp")]
428    #[doc(alias = "gst_base_src_get_do_timestamp")]
429    pub fn do_timestamp(&self) -> bool {
430        unsafe {
431            from_glib(gst_base::ffi::gst_base_src_get_do_timestamp(
432                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc
433            ))
434        }
435    }
436
437    #[doc(alias = "do-timestamp")]
438    pub fn connect_do_timestamp_notify<F: Fn(&Self) + Send + Sync + 'static>(
439        &self,
440        f: F,
441    ) -> glib::SignalHandlerId {
442        unsafe extern "C" fn notify_do_timestamp_trampoline<
443            F: Fn(&AppSrc) + Send + Sync + 'static,
444        >(
445            this: *mut ffi::GstAppSrc,
446            _param_spec: glib::ffi::gpointer,
447            f: glib::ffi::gpointer,
448        ) {
449            let f: &F = &*(f as *const F);
450            f(&AppSrc::from_glib_borrow(this))
451        }
452        unsafe {
453            let f: Box<F> = Box::new(f);
454            glib::signal::connect_raw(
455                self.as_ptr() as *mut _,
456                b"notify::do-timestamp\0".as_ptr() as *const _,
457                Some(mem::transmute::<*const (), unsafe extern "C" fn()>(
458                    notify_do_timestamp_trampoline::<F> as *const (),
459                )),
460                Box::into_raw(f),
461            )
462        }
463    }
464
465    #[doc(alias = "set-automatic-eos")]
466    #[doc(alias = "gst_base_src_set_automatic_eos")]
467    pub fn set_automatic_eos(&self, automatic_eos: bool) {
468        unsafe {
469            gst_base::ffi::gst_base_src_set_automatic_eos(
470                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
471                automatic_eos.into_glib(),
472            );
473        }
474    }
475
476    pub fn sink(&self) -> AppSrcSink {
477        AppSrcSink::new(self)
478    }
479}
480
481// rustdoc-stripper-ignore-next
482/// A [builder-pattern] type to construct [`AppSrc`] objects.
483///
484/// [builder-pattern]: https://doc.rust-lang.org/1.0.0/style/ownership/builders.html
485#[must_use = "The builder must be built to be used"]
486pub struct AppSrcBuilder {
487    builder: glib::object::ObjectBuilder<'static, AppSrc>,
488    callbacks: Option<AppSrcCallbacks>,
489    automatic_eos: Option<bool>,
490}
491
492impl AppSrcBuilder {
493    fn new() -> Self {
494        Self {
495            builder: glib::Object::builder(),
496            callbacks: None,
497            automatic_eos: None,
498        }
499    }
500
501    // rustdoc-stripper-ignore-next
502    /// Build the [`AppSrc`].
503    #[must_use = "Building the object from the builder is usually expensive and is not expected to have side effects"]
504    pub fn build(self) -> AppSrc {
505        let appsrc = self.builder.build();
506
507        if let Some(callbacks) = self.callbacks {
508            appsrc.set_callbacks(callbacks);
509        }
510
511        if let Some(automatic_eos) = self.automatic_eos {
512            appsrc.set_automatic_eos(automatic_eos);
513        }
514
515        appsrc
516    }
517
518    pub fn automatic_eos(self, automatic_eos: bool) -> Self {
519        Self {
520            automatic_eos: Some(automatic_eos),
521            ..self
522        }
523    }
524
525    pub fn block(self, block: bool) -> Self {
526        Self {
527            builder: self.builder.property("block", block),
528            ..self
529        }
530    }
531
532    pub fn callbacks(self, callbacks: AppSrcCallbacks) -> Self {
533        Self {
534            callbacks: Some(callbacks),
535            ..self
536        }
537    }
538
539    pub fn caps(self, caps: &gst::Caps) -> Self {
540        Self {
541            builder: self.builder.property("caps", caps),
542            ..self
543        }
544    }
545
546    pub fn do_timestamp(self, do_timestamp: bool) -> Self {
547        Self {
548            builder: self.builder.property("do-timestamp", do_timestamp),
549            ..self
550        }
551    }
552
553    pub fn duration(self, duration: u64) -> Self {
554        Self {
555            builder: self.builder.property("duration", duration),
556            ..self
557        }
558    }
559
560    pub fn format(self, format: gst::Format) -> Self {
561        Self {
562            builder: self.builder.property("format", format),
563            ..self
564        }
565    }
566
567    #[cfg(feature = "v1_18")]
568    #[cfg_attr(docsrs, doc(cfg(feature = "v1_18")))]
569    pub fn handle_segment_change(self, handle_segment_change: bool) -> Self {
570        Self {
571            builder: self
572                .builder
573                .property("handle-segment-change", handle_segment_change),
574            ..self
575        }
576    }
577
578    pub fn is_live(self, is_live: bool) -> Self {
579        Self {
580            builder: self.builder.property("is-live", is_live),
581            ..self
582        }
583    }
584
585    #[cfg(feature = "v1_20")]
586    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
587    pub fn leaky_type(self, leaky_type: crate::AppLeakyType) -> Self {
588        Self {
589            builder: self.builder.property("leaky-type", leaky_type),
590            ..self
591        }
592    }
593
594    #[cfg(feature = "v1_20")]
595    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
596    pub fn max_buffers(self, max_buffers: u64) -> Self {
597        Self {
598            builder: self.builder.property("max-buffers", max_buffers),
599            ..self
600        }
601    }
602
603    pub fn max_bytes(self, max_bytes: u64) -> Self {
604        Self {
605            builder: self.builder.property("max-bytes", max_bytes),
606            ..self
607        }
608    }
609
610    pub fn max_latency(self, max_latency: i64) -> Self {
611        Self {
612            builder: self.builder.property("max-latency", max_latency),
613            ..self
614        }
615    }
616
617    #[cfg(feature = "v1_20")]
618    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
619    pub fn max_time(self, max_time: Option<gst::ClockTime>) -> Self {
620        Self {
621            builder: self.builder.property("max-time", max_time),
622            ..self
623        }
624    }
625
626    pub fn min_latency(self, min_latency: i64) -> Self {
627        Self {
628            builder: self.builder.property("min-latency", min_latency),
629            ..self
630        }
631    }
632
633    pub fn min_percent(self, min_percent: u32) -> Self {
634        Self {
635            builder: self.builder.property("min-percent", min_percent),
636            ..self
637        }
638    }
639
640    pub fn size(self, size: i64) -> Self {
641        Self {
642            builder: self.builder.property("size", size),
643            ..self
644        }
645    }
646
647    pub fn stream_type(self, stream_type: crate::AppStreamType) -> Self {
648        Self {
649            builder: self.builder.property("stream-type", stream_type),
650            ..self
651        }
652    }
653
654    pub fn name(self, name: impl Into<glib::GString>) -> Self {
655        Self {
656            builder: self.builder.property("name", name.into()),
657            ..self
658        }
659    }
660}
661
662#[derive(Debug)]
663pub struct AppSrcSink {
664    app_src: glib::WeakRef<AppSrc>,
665    waker_reference: Arc<Mutex<Option<Waker>>>,
666}
667
668impl AppSrcSink {
669    fn new(app_src: &AppSrc) -> Self {
670        skip_assert_initialized!();
671
672        let waker_reference = Arc::new(Mutex::new(None as Option<Waker>));
673
674        app_src.set_callbacks(
675            AppSrcCallbacks::builder()
676                .need_data({
677                    let waker_reference = Arc::clone(&waker_reference);
678
679                    move |_, _| {
680                        if let Some(waker) = waker_reference.lock().unwrap().take() {
681                            waker.wake();
682                        }
683                    }
684                })
685                .build(),
686        );
687
688        Self {
689            app_src: app_src.downgrade(),
690            waker_reference,
691        }
692    }
693}
694
695impl Drop for AppSrcSink {
696    fn drop(&mut self) {
697        #[cfg(not(feature = "v1_18"))]
698        {
699            // This is not thread-safe before 1.16.3, see
700            // https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
701            if gst::version() >= (1, 16, 3, 0) {
702                if let Some(app_src) = self.app_src.upgrade() {
703                    app_src.set_callbacks(AppSrcCallbacks::builder().build());
704                }
705            }
706        }
707    }
708}
709
710impl Sink<gst::Sample> for AppSrcSink {
711    type Error = gst::FlowError;
712
713    fn poll_ready(self: Pin<&mut Self>, context: &mut Context) -> Poll<Result<(), Self::Error>> {
714        let mut waker = self.waker_reference.lock().unwrap();
715
716        let Some(app_src) = self.app_src.upgrade() else {
717            return Poll::Ready(Err(gst::FlowError::Eos));
718        };
719
720        let current_level_bytes = app_src.current_level_bytes();
721        let max_bytes = app_src.max_bytes();
722
723        if current_level_bytes >= max_bytes && max_bytes != 0 {
724            waker.replace(context.waker().to_owned());
725
726            Poll::Pending
727        } else {
728            Poll::Ready(Ok(()))
729        }
730    }
731
732    fn start_send(self: Pin<&mut Self>, sample: gst::Sample) -> Result<(), Self::Error> {
733        let Some(app_src) = self.app_src.upgrade() else {
734            return Err(gst::FlowError::Eos);
735        };
736
737        app_src.push_sample(&sample)?;
738
739        Ok(())
740    }
741
742    fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
743        Poll::Ready(Ok(()))
744    }
745
746    fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
747        let Some(app_src) = self.app_src.upgrade() else {
748            return Poll::Ready(Ok(()));
749        };
750
751        app_src.end_of_stream()?;
752
753        Poll::Ready(Ok(()))
754    }
755}
756
757#[cfg(test)]
758mod tests {
759    use std::sync::atomic::{AtomicUsize, Ordering};
760
761    use futures_util::{sink::SinkExt, stream::StreamExt};
762    use gst::prelude::*;
763
764    use super::*;
765
766    #[test]
767    fn test_app_src_sink() {
768        gst::init().unwrap();
769
770        let appsrc = gst::ElementFactory::make("appsrc").build().unwrap();
771        let fakesink = gst::ElementFactory::make("fakesink")
772            .property("signal-handoffs", true)
773            .build()
774            .unwrap();
775
776        let pipeline = gst::Pipeline::new();
777        pipeline.add(&appsrc).unwrap();
778        pipeline.add(&fakesink).unwrap();
779
780        appsrc.link(&fakesink).unwrap();
781
782        let mut bus_stream = pipeline.bus().unwrap().stream();
783        let mut app_src_sink = appsrc.dynamic_cast::<AppSrc>().unwrap().sink();
784
785        let sample_quantity = 5;
786
787        let samples = (0..sample_quantity)
788            .map(|_| gst::Sample::builder().buffer(&gst::Buffer::new()).build())
789            .collect::<Vec<gst::Sample>>();
790
791        let mut sample_stream = futures_util::stream::iter(samples).map(Ok);
792
793        let handoff_count_reference = Arc::new(AtomicUsize::new(0));
794
795        fakesink.connect("handoff", false, {
796            let handoff_count_reference = Arc::clone(&handoff_count_reference);
797
798            move |_| {
799                handoff_count_reference.fetch_add(1, Ordering::AcqRel);
800
801                None
802            }
803        });
804
805        pipeline.set_state(gst::State::Playing).unwrap();
806
807        futures_executor::block_on(app_src_sink.send_all(&mut sample_stream)).unwrap();
808        futures_executor::block_on(app_src_sink.close()).unwrap();
809
810        while let Some(message) = futures_executor::block_on(bus_stream.next()) {
811            match message.view() {
812                gst::MessageView::Eos(_) => break,
813                gst::MessageView::Error(_) => unreachable!(),
814                _ => continue,
815            }
816        }
817
818        pipeline.set_state(gst::State::Null).unwrap();
819
820        assert_eq!(
821            handoff_count_reference.load(Ordering::Acquire),
822            sample_quantity
823        );
824    }
825}
gstreamer_app/app_src.rs

gstreamer_app/
app_src.rs
