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}