2017-08-01 14:42:15 +00:00
|
|
|
// Copyright (C) 2017 Sebastian Dröge <sebastian@centricular.com>
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
|
|
|
|
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
|
|
|
|
// option. This file may not be copied, modified, or distributed
|
|
|
|
// except according to those terms.
|
|
|
|
|
2020-01-29 12:40:17 +00:00
|
|
|
use futures_sink::Sink;
|
2020-11-21 18:06:11 +00:00
|
|
|
use glib::ffi::{gboolean, gpointer};
|
2020-05-13 19:13:11 +00:00
|
|
|
use glib::prelude::*;
|
2018-04-01 08:30:03 +00:00
|
|
|
use glib::translate::*;
|
2020-03-05 08:31:48 +00:00
|
|
|
|
2020-11-21 18:06:11 +00:00
|
|
|
use crate::AppSrc;
|
2018-04-25 08:10:06 +00:00
|
|
|
use std::cell::RefCell;
|
2017-12-16 12:04:36 +00:00
|
|
|
use std::mem;
|
2020-03-05 08:31:48 +00:00
|
|
|
use std::panic;
|
2020-01-29 12:40:17 +00:00
|
|
|
use std::pin::Pin;
|
2018-04-01 08:30:03 +00:00
|
|
|
use std::ptr;
|
2020-03-05 08:31:48 +00:00
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
2020-01-29 15:51:47 +00:00
|
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use std::task::{Context, Poll, Waker};
|
2017-08-01 14:42:15 +00:00
|
|
|
|
2019-02-28 08:32:13 +00:00
|
|
|
#[allow(clippy::type_complexity)]
|
2017-08-01 14:42:15 +00:00
|
|
|
pub struct AppSrcCallbacks {
|
2019-06-06 06:09:34 +00:00
|
|
|
need_data: Option<RefCell<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>>,
|
|
|
|
enough_data: Option<Box<dyn Fn(&AppSrc) + Send + Sync + 'static>>,
|
|
|
|
seek_data: Option<Box<dyn Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>>,
|
2020-03-05 08:31:48 +00:00
|
|
|
panicked: AtomicBool,
|
2020-11-21 18:06:11 +00:00
|
|
|
callbacks: ffi::GstAppSrcCallbacks,
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|
|
|
|
|
2017-12-16 12:14:29 +00:00
|
|
|
unsafe impl Send for AppSrcCallbacks {}
|
|
|
|
unsafe impl Sync for AppSrcCallbacks {}
|
|
|
|
|
2017-12-16 09:37:00 +00:00
|
|
|
impl AppSrcCallbacks {
|
2020-06-25 16:22:25 +00:00
|
|
|
pub fn builder() -> AppSrcCallbacksBuilder {
|
2017-12-16 09:37:00 +00:00
|
|
|
skip_assert_initialized!();
|
|
|
|
|
|
|
|
AppSrcCallbacksBuilder {
|
|
|
|
need_data: None,
|
|
|
|
enough_data: None,
|
|
|
|
seek_data: None,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-02-28 08:32:13 +00:00
|
|
|
#[allow(clippy::type_complexity)]
|
2017-12-10 13:18:54 +00:00
|
|
|
pub struct AppSrcCallbacksBuilder {
|
2019-06-06 06:09:34 +00:00
|
|
|
need_data: Option<RefCell<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>>,
|
|
|
|
enough_data: Option<Box<dyn Fn(&AppSrc) + Send + Sync + 'static>>,
|
|
|
|
seek_data: Option<Box<dyn Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>>,
|
2017-12-10 13:18:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl AppSrcCallbacksBuilder {
|
2018-04-25 08:08:44 +00:00
|
|
|
pub fn need_data<F: FnMut(&AppSrc, u32) + Send + 'static>(self, need_data: F) -> Self {
|
2017-12-10 13:18:54 +00:00
|
|
|
Self {
|
2018-04-25 08:08:44 +00:00
|
|
|
need_data: Some(RefCell::new(Box::new(need_data))),
|
2017-12-10 13:18:54 +00:00
|
|
|
..self
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn enough_data<F: Fn(&AppSrc) + Send + Sync + 'static>(self, enough_data: F) -> Self {
|
|
|
|
Self {
|
|
|
|
enough_data: Some(Box::new(enough_data)),
|
|
|
|
..self
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-12-16 12:53:00 +00:00
|
|
|
pub fn seek_data<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
|
|
|
|
self,
|
|
|
|
seek_data: F,
|
|
|
|
) -> Self {
|
2017-12-10 13:18:54 +00:00
|
|
|
Self {
|
|
|
|
seek_data: Some(Box::new(seek_data)),
|
|
|
|
..self
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn build(self) -> AppSrcCallbacks {
|
|
|
|
let have_need_data = self.need_data.is_some();
|
|
|
|
let have_enough_data = self.enough_data.is_some();
|
|
|
|
let have_seek_data = self.seek_data.is_some();
|
|
|
|
|
2017-08-01 14:42:15 +00:00
|
|
|
AppSrcCallbacks {
|
2017-12-10 13:18:54 +00:00
|
|
|
need_data: self.need_data,
|
|
|
|
enough_data: self.enough_data,
|
|
|
|
seek_data: self.seek_data,
|
2020-03-05 08:31:48 +00:00
|
|
|
panicked: AtomicBool::new(false),
|
2020-11-21 18:06:11 +00:00
|
|
|
callbacks: ffi::GstAppSrcCallbacks {
|
2017-12-16 12:53:00 +00:00
|
|
|
need_data: if have_need_data {
|
|
|
|
Some(trampoline_need_data)
|
|
|
|
} else {
|
|
|
|
None
|
|
|
|
},
|
|
|
|
enough_data: if have_enough_data {
|
|
|
|
Some(trampoline_enough_data)
|
|
|
|
} else {
|
|
|
|
None
|
|
|
|
},
|
|
|
|
seek_data: if have_seek_data {
|
|
|
|
Some(trampoline_seek_data)
|
|
|
|
} else {
|
|
|
|
None
|
|
|
|
},
|
2017-08-01 14:42:15 +00:00
|
|
|
_gst_reserved: [
|
|
|
|
ptr::null_mut(),
|
|
|
|
ptr::null_mut(),
|
|
|
|
ptr::null_mut(),
|
|
|
|
ptr::null_mut(),
|
|
|
|
],
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-03-05 08:31:48 +00:00
|
|
|
fn post_panic_error_message(element: &AppSrc, err: &dyn std::any::Any) {
|
2020-03-22 14:18:47 +00:00
|
|
|
skip_assert_initialized!();
|
2020-03-05 08:31:48 +00:00
|
|
|
if let Some(cause) = err.downcast_ref::<&str>() {
|
2020-11-21 18:06:11 +00:00
|
|
|
gst::gst_element_error!(&element, gst::LibraryError::Failed, ["Panicked: {}", cause]);
|
2020-03-05 08:31:48 +00:00
|
|
|
} else if let Some(cause) = err.downcast_ref::<String>() {
|
2020-11-21 18:06:11 +00:00
|
|
|
gst::gst_element_error!(&element, gst::LibraryError::Failed, ["Panicked: {}", cause]);
|
2020-03-05 08:31:48 +00:00
|
|
|
} else {
|
2020-11-21 18:06:11 +00:00
|
|
|
gst::gst_element_error!(&element, gst::LibraryError::Failed, ["Panicked"]);
|
2020-03-05 08:31:48 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-08-01 14:42:15 +00:00
|
|
|
unsafe extern "C" fn trampoline_need_data(
|
2020-11-21 18:06:11 +00:00
|
|
|
appsrc: *mut ffi::GstAppSrc,
|
2017-08-01 14:42:15 +00:00
|
|
|
length: u32,
|
|
|
|
callbacks: gpointer,
|
|
|
|
) {
|
2017-08-02 17:09:00 +00:00
|
|
|
let callbacks = &*(callbacks as *const AppSrcCallbacks);
|
2020-04-05 14:52:56 +00:00
|
|
|
let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
|
2020-03-05 08:31:48 +00:00
|
|
|
|
|
|
|
if callbacks.panicked.load(Ordering::Relaxed) {
|
2020-04-05 14:52:56 +00:00
|
|
|
let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
|
2020-11-21 18:06:11 +00:00
|
|
|
gst::gst_element_error!(&element, gst::LibraryError::Failed, ["Panicked"]);
|
2020-03-05 08:31:48 +00:00
|
|
|
return;
|
|
|
|
}
|
2017-08-01 14:42:15 +00:00
|
|
|
|
2018-07-20 07:21:06 +00:00
|
|
|
if let Some(ref need_data) = callbacks.need_data {
|
2020-03-05 08:31:48 +00:00
|
|
|
let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
|
|
|
|
(&mut *need_data.borrow_mut())(&element, length)
|
|
|
|
}));
|
|
|
|
match result {
|
|
|
|
Ok(result) => result,
|
|
|
|
Err(err) => {
|
|
|
|
callbacks.panicked.store(true, Ordering::Relaxed);
|
|
|
|
post_panic_error_message(&element, &err);
|
|
|
|
}
|
|
|
|
}
|
2018-07-20 07:21:06 +00:00
|
|
|
}
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|
|
|
|
|
2020-11-21 18:06:11 +00:00
|
|
|
unsafe extern "C" fn trampoline_enough_data(appsrc: *mut ffi::GstAppSrc, callbacks: gpointer) {
|
2017-08-02 17:09:00 +00:00
|
|
|
let callbacks = &*(callbacks as *const AppSrcCallbacks);
|
2020-04-05 14:52:56 +00:00
|
|
|
let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
|
2020-03-05 08:31:48 +00:00
|
|
|
|
|
|
|
if callbacks.panicked.load(Ordering::Relaxed) {
|
2020-04-05 14:52:56 +00:00
|
|
|
let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
|
2020-11-21 18:06:11 +00:00
|
|
|
gst::gst_element_error!(&element, gst::LibraryError::Failed, ["Panicked"]);
|
2020-03-05 08:31:48 +00:00
|
|
|
return;
|
|
|
|
}
|
2017-08-01 14:42:15 +00:00
|
|
|
|
2018-07-20 07:21:06 +00:00
|
|
|
if let Some(ref enough_data) = callbacks.enough_data {
|
2020-03-05 08:31:48 +00:00
|
|
|
let result = panic::catch_unwind(panic::AssertUnwindSafe(|| (*enough_data)(&element)));
|
|
|
|
match result {
|
|
|
|
Ok(result) => result,
|
|
|
|
Err(err) => {
|
|
|
|
callbacks.panicked.store(true, Ordering::Relaxed);
|
|
|
|
post_panic_error_message(&element, &err);
|
|
|
|
}
|
|
|
|
}
|
2018-07-20 07:21:06 +00:00
|
|
|
}
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
unsafe extern "C" fn trampoline_seek_data(
|
2020-11-21 18:06:11 +00:00
|
|
|
appsrc: *mut ffi::GstAppSrc,
|
2017-08-01 14:42:15 +00:00
|
|
|
offset: u64,
|
|
|
|
callbacks: gpointer,
|
|
|
|
) -> gboolean {
|
2017-08-02 17:09:00 +00:00
|
|
|
let callbacks = &*(callbacks as *const AppSrcCallbacks);
|
2020-04-05 14:52:56 +00:00
|
|
|
let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
|
2020-03-05 08:31:48 +00:00
|
|
|
|
|
|
|
if callbacks.panicked.load(Ordering::Relaxed) {
|
2020-04-05 14:52:56 +00:00
|
|
|
let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
|
2020-11-21 18:06:11 +00:00
|
|
|
gst::gst_element_error!(&element, gst::LibraryError::Failed, ["Panicked"]);
|
2020-03-05 08:31:48 +00:00
|
|
|
return false.to_glib();
|
|
|
|
}
|
2017-08-01 14:42:15 +00:00
|
|
|
|
2018-07-20 07:21:06 +00:00
|
|
|
let ret = if let Some(ref seek_data) = callbacks.seek_data {
|
2020-03-05 08:31:48 +00:00
|
|
|
let result =
|
|
|
|
panic::catch_unwind(panic::AssertUnwindSafe(|| (*seek_data)(&element, offset)));
|
|
|
|
match result {
|
|
|
|
Ok(result) => result,
|
|
|
|
Err(err) => {
|
|
|
|
callbacks.panicked.store(true, Ordering::Relaxed);
|
|
|
|
post_panic_error_message(&element, &err);
|
|
|
|
|
|
|
|
false
|
|
|
|
}
|
|
|
|
}
|
2018-07-20 07:21:06 +00:00
|
|
|
} else {
|
|
|
|
false
|
|
|
|
};
|
|
|
|
|
|
|
|
ret.to_glib()
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
unsafe extern "C" fn destroy_callbacks(ptr: gpointer) {
|
2017-11-11 15:05:08 +00:00
|
|
|
Box::<AppSrcCallbacks>::from_raw(ptr as *mut _);
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl AppSrc {
|
2019-01-08 16:13:37 +00:00
|
|
|
pub fn end_of_stream(&self) -> Result<gst::FlowSuccess, gst::FlowError> {
|
2020-11-21 18:06:11 +00:00
|
|
|
let ret: gst::FlowReturn =
|
|
|
|
unsafe { from_glib(ffi::gst_app_src_end_of_stream(self.to_glib_none().0)) };
|
2019-01-08 16:13:37 +00:00
|
|
|
ret.into_result()
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn push_buffer(&self, buffer: gst::Buffer) -> Result<gst::FlowSuccess, gst::FlowError> {
|
|
|
|
let ret: gst::FlowReturn = unsafe {
|
2020-11-21 18:06:11 +00:00
|
|
|
from_glib(ffi::gst_app_src_push_buffer(
|
2017-08-01 14:42:15 +00:00
|
|
|
self.to_glib_none().0,
|
|
|
|
buffer.into_ptr(),
|
|
|
|
))
|
2019-01-08 16:13:37 +00:00
|
|
|
};
|
|
|
|
ret.into_result()
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|
|
|
|
|
2020-11-25 14:54:02 +00:00
|
|
|
#[cfg(any(feature = "v1_14", all(not(doctest), doc)))]
|
|
|
|
#[cfg_attr(all(not(doctest), doc), doc(cfg(feature = "v1_14")))]
|
2019-01-08 16:13:37 +00:00
|
|
|
pub fn push_buffer_list(
|
|
|
|
&self,
|
|
|
|
list: gst::BufferList,
|
|
|
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
|
|
|
let ret: gst::FlowReturn = unsafe {
|
2020-11-21 18:06:11 +00:00
|
|
|
from_glib(ffi::gst_app_src_push_buffer_list(
|
2018-03-15 07:52:36 +00:00
|
|
|
self.to_glib_none().0,
|
|
|
|
list.into_ptr(),
|
|
|
|
))
|
2019-01-08 16:13:37 +00:00
|
|
|
};
|
|
|
|
ret.into_result()
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn push_sample(&self, sample: &gst::Sample) -> Result<gst::FlowSuccess, gst::FlowError> {
|
|
|
|
let ret: gst::FlowReturn = unsafe {
|
2020-11-21 18:06:11 +00:00
|
|
|
from_glib(ffi::gst_app_src_push_sample(
|
2019-01-08 16:13:37 +00:00
|
|
|
self.to_glib_none().0,
|
|
|
|
sample.to_glib_none().0,
|
|
|
|
))
|
|
|
|
};
|
|
|
|
ret.into_result()
|
2018-03-15 07:52:36 +00:00
|
|
|
}
|
|
|
|
|
2017-08-01 14:42:15 +00:00
|
|
|
pub fn set_callbacks(&self, callbacks: AppSrcCallbacks) {
|
2020-02-09 19:30:53 +00:00
|
|
|
use once_cell::sync::Lazy;
|
|
|
|
static SET_ONCE_QUARK: Lazy<glib::Quark> =
|
|
|
|
Lazy::new(|| glib::Quark::from_string("gstreamer-rs-app-src-callbacks"));
|
|
|
|
|
2017-08-01 14:42:15 +00:00
|
|
|
unsafe {
|
2020-02-09 19:30:53 +00:00
|
|
|
let src = self.to_glib_none().0;
|
2020-02-15 09:05:25 +00:00
|
|
|
// This is not thread-safe before 1.16.3, see
|
|
|
|
// https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
|
|
|
|
if gst::version() < (1, 16, 3, 0) {
|
2020-11-21 18:06:11 +00:00
|
|
|
if !glib::gobject_ffi::g_object_get_qdata(src as *mut _, SET_ONCE_QUARK.to_glib())
|
2020-02-15 09:05:25 +00:00
|
|
|
.is_null()
|
|
|
|
{
|
|
|
|
panic!("AppSrc callbacks can only be set once");
|
|
|
|
}
|
|
|
|
|
2020-11-21 18:06:11 +00:00
|
|
|
glib::gobject_ffi::g_object_set_qdata(
|
2020-02-15 09:05:25 +00:00
|
|
|
src as *mut _,
|
|
|
|
SET_ONCE_QUARK.to_glib(),
|
|
|
|
1 as *mut _,
|
|
|
|
);
|
2020-02-09 19:30:53 +00:00
|
|
|
}
|
|
|
|
|
2020-11-21 18:06:11 +00:00
|
|
|
ffi::gst_app_src_set_callbacks(
|
2020-02-09 19:30:53 +00:00
|
|
|
src,
|
2017-08-01 14:42:15 +00:00
|
|
|
mut_override(&callbacks.callbacks),
|
|
|
|
Box::into_raw(Box::new(callbacks)) as *mut _,
|
|
|
|
Some(destroy_callbacks),
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|
2017-12-16 12:04:36 +00:00
|
|
|
|
|
|
|
pub fn set_latency(&self, min: gst::ClockTime, max: gst::ClockTime) {
|
|
|
|
unsafe {
|
2020-11-21 18:06:11 +00:00
|
|
|
ffi::gst_app_src_set_latency(self.to_glib_none().0, min.to_glib(), max.to_glib());
|
2017-12-16 12:04:36 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn get_latency(&self) -> (gst::ClockTime, gst::ClockTime) {
|
|
|
|
unsafe {
|
2019-07-11 12:34:28 +00:00
|
|
|
let mut min = mem::MaybeUninit::uninit();
|
|
|
|
let mut max = mem::MaybeUninit::uninit();
|
2020-11-21 18:06:11 +00:00
|
|
|
ffi::gst_app_src_get_latency(self.to_glib_none().0, min.as_mut_ptr(), max.as_mut_ptr());
|
2019-07-11 12:34:28 +00:00
|
|
|
(from_glib(min.assume_init()), from_glib(max.assume_init()))
|
2017-12-16 12:04:36 +00:00
|
|
|
}
|
|
|
|
}
|
2020-01-29 12:40:17 +00:00
|
|
|
|
|
|
|
pub fn sink(&self) -> AppSrcSink {
|
|
|
|
AppSrcSink::new(self)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Debug)]
|
|
|
|
pub struct AppSrcSink {
|
2020-05-13 19:13:11 +00:00
|
|
|
app_src: glib::WeakRef<AppSrc>,
|
2020-01-29 15:51:47 +00:00
|
|
|
waker_reference: Arc<Mutex<Option<Waker>>>,
|
2020-01-29 12:40:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl AppSrcSink {
|
|
|
|
fn new(app_src: &AppSrc) -> Self {
|
|
|
|
skip_assert_initialized!();
|
|
|
|
|
2020-01-29 15:51:47 +00:00
|
|
|
let waker_reference = Arc::new(Mutex::new(None as Option<Waker>));
|
|
|
|
|
|
|
|
app_src.set_callbacks(
|
2020-06-25 16:22:25 +00:00
|
|
|
AppSrcCallbacks::builder()
|
2020-01-29 15:51:47 +00:00
|
|
|
.need_data({
|
|
|
|
let waker_reference = Arc::clone(&waker_reference);
|
|
|
|
|
|
|
|
move |_, _| {
|
|
|
|
if let Some(waker) = waker_reference.lock().unwrap().take() {
|
|
|
|
waker.wake();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
})
|
|
|
|
.build(),
|
|
|
|
);
|
2020-01-29 12:40:17 +00:00
|
|
|
|
2020-01-29 15:51:47 +00:00
|
|
|
Self {
|
2020-05-13 19:13:11 +00:00
|
|
|
app_src: app_src.downgrade(),
|
2020-01-29 15:51:47 +00:00
|
|
|
waker_reference,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-02-15 09:05:25 +00:00
|
|
|
impl Drop for AppSrcSink {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
// This is not thread-safe before 1.16.3, see
|
|
|
|
// https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
|
|
|
|
if gst::version() >= (1, 16, 3, 0) {
|
2020-05-13 19:13:11 +00:00
|
|
|
if let Some(app_src) = self.app_src.upgrade() {
|
2020-06-25 16:22:25 +00:00
|
|
|
app_src.set_callbacks(AppSrcCallbacks::builder().build());
|
2020-05-13 19:13:11 +00:00
|
|
|
}
|
2020-02-15 09:05:25 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-01-29 12:40:17 +00:00
|
|
|
impl Sink<gst::Sample> for AppSrcSink {
|
|
|
|
type Error = gst::FlowError;
|
|
|
|
|
2020-01-29 15:51:47 +00:00
|
|
|
fn poll_ready(self: Pin<&mut Self>, context: &mut Context) -> Poll<Result<(), Self::Error>> {
|
|
|
|
let mut waker = self.waker_reference.lock().unwrap();
|
|
|
|
|
2020-05-13 19:13:11 +00:00
|
|
|
let app_src = match self.app_src.upgrade() {
|
|
|
|
Some(app_src) => app_src,
|
|
|
|
None => return Poll::Ready(Err(gst::FlowError::Eos)),
|
|
|
|
};
|
|
|
|
|
|
|
|
let current_level_bytes = app_src.get_current_level_bytes();
|
|
|
|
let max_bytes = app_src.get_max_bytes();
|
2020-01-29 15:51:47 +00:00
|
|
|
|
|
|
|
if current_level_bytes >= max_bytes && max_bytes != 0 {
|
|
|
|
waker.replace(context.waker().to_owned());
|
|
|
|
|
|
|
|
Poll::Pending
|
|
|
|
} else {
|
|
|
|
Poll::Ready(Ok(()))
|
|
|
|
}
|
2020-01-29 12:40:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
fn start_send(self: Pin<&mut Self>, sample: gst::Sample) -> Result<(), Self::Error> {
|
2020-05-13 19:13:11 +00:00
|
|
|
let app_src = match self.app_src.upgrade() {
|
|
|
|
Some(app_src) => app_src,
|
|
|
|
None => return Err(gst::FlowError::Eos),
|
|
|
|
};
|
|
|
|
|
|
|
|
app_src.push_sample(&sample)?;
|
2020-01-29 12:40:17 +00:00
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
|
|
|
|
Poll::Ready(Ok(()))
|
|
|
|
}
|
|
|
|
|
|
|
|
fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Self::Error>> {
|
2020-05-13 19:13:11 +00:00
|
|
|
let app_src = match self.app_src.upgrade() {
|
|
|
|
Some(app_src) => app_src,
|
|
|
|
None => return Poll::Ready(Ok(())),
|
|
|
|
};
|
|
|
|
|
|
|
|
app_src.end_of_stream()?;
|
2020-01-29 12:40:17 +00:00
|
|
|
|
|
|
|
Poll::Ready(Ok(()))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
mod tests {
|
|
|
|
use super::*;
|
2020-01-29 15:51:47 +00:00
|
|
|
use futures_util::{sink::SinkExt, stream::StreamExt};
|
|
|
|
use gst::prelude::*;
|
|
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
2020-01-29 12:40:17 +00:00
|
|
|
|
|
|
|
#[test]
|
|
|
|
fn test_app_src_sink() {
|
|
|
|
gst::init().unwrap();
|
|
|
|
|
2020-01-29 15:51:47 +00:00
|
|
|
let appsrc = gst::ElementFactory::make("appsrc", None).unwrap();
|
|
|
|
let fakesink = gst::ElementFactory::make("fakesink", None).unwrap();
|
|
|
|
|
|
|
|
fakesink.set_property("signal-handoffs", &true).unwrap();
|
|
|
|
|
|
|
|
let pipeline = gst::Pipeline::new(None);
|
|
|
|
pipeline.add(&appsrc).unwrap();
|
|
|
|
pipeline.add(&fakesink).unwrap();
|
|
|
|
|
|
|
|
appsrc.link(&fakesink).unwrap();
|
|
|
|
|
|
|
|
let mut bus_stream = pipeline.get_bus().unwrap().stream();
|
|
|
|
let mut app_src_sink = appsrc.dynamic_cast::<AppSrc>().unwrap().sink();
|
|
|
|
|
|
|
|
let sample_quantity = 5;
|
|
|
|
|
|
|
|
let samples = (0..sample_quantity)
|
2020-06-25 16:22:25 +00:00
|
|
|
.map(|_| gst::Sample::builder().buffer(&gst::Buffer::new()).build())
|
2020-01-29 15:51:47 +00:00
|
|
|
.collect::<Vec<gst::Sample>>();
|
|
|
|
|
|
|
|
let mut sample_stream = futures_util::stream::iter(samples).map(Ok);
|
|
|
|
|
|
|
|
let handoff_count_reference = Arc::new(AtomicUsize::new(0));
|
|
|
|
|
|
|
|
fakesink
|
|
|
|
.connect("handoff", false, {
|
|
|
|
let handoff_count_reference = Arc::clone(&handoff_count_reference);
|
|
|
|
|
|
|
|
move |_| {
|
|
|
|
handoff_count_reference.fetch_add(1, Ordering::AcqRel);
|
|
|
|
|
|
|
|
None
|
|
|
|
}
|
|
|
|
})
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
pipeline.set_state(gst::State::Playing).unwrap();
|
|
|
|
|
|
|
|
futures_executor::block_on(app_src_sink.send_all(&mut sample_stream)).unwrap();
|
|
|
|
futures_executor::block_on(app_src_sink.close()).unwrap();
|
|
|
|
|
|
|
|
while let Some(message) = futures_executor::block_on(bus_stream.next()) {
|
|
|
|
match message.view() {
|
|
|
|
gst::MessageView::Eos(_) => break,
|
|
|
|
gst::MessageView::Error(_) => unreachable!(),
|
|
|
|
_ => continue,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pipeline.set_state(gst::State::Null).unwrap();
|
|
|
|
|
|
|
|
assert_eq!(
|
|
|
|
handoff_count_reference.load(Ordering::Acquire),
|
|
|
|
sample_quantity
|
|
|
|
);
|
2020-01-29 12:40:17 +00:00
|
|
|
}
|
2017-08-01 14:42:15 +00:00
|
|
|
}
|