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::{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 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 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#[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 #[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 #[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 #[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 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}