mirror of
https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git
synced 2025-03-10 15:31:34 +00:00
ts-jitterbuffer: add dedicated functions for state transitions + tests
fixes https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/issues/95
This commit is contained in:
François Laignel
parent
c50aa09034
commit
e2add3f2c8
2 changed files with 253 additions and 50 deletions
|
|
@ -1132,6 +1132,73 @@ impl JitterBuffer {
|
|||
state.clock_rate = -1;
|
||||
state.jbuf.borrow().reset_skew();
|
||||
}
|
||||
|
||||
async fn prepare(&self, element: &gst::Element) -> Result<(), gst::ErrorMessage> {
|
||||
let mut state = self.state.lock().await;
|
||||
gst_info!(CAT, obj: element, "Preparing");
|
||||
|
||||
let (context, latency) = {
|
||||
let settings = self.settings.lock().await;
|
||||
let context = Context::acquire(&settings.context, settings.context_wait).unwrap();
|
||||
let latency = settings.latency_ms as u64 * gst::MSECOND;
|
||||
(context, latency)
|
||||
};
|
||||
|
||||
state.src_pad_handler = JitterBufferPadSrcHandler::new(latency);
|
||||
|
||||
let _ = self
|
||||
.src_pad
|
||||
.prepare(context, &state.src_pad_handler)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
gst_error_msg!(
|
||||
gst::ResourceError::OpenRead,
|
||||
["Error preparing src_pad: {:?}", err]
|
||||
);
|
||||
gst::StateChangeError
|
||||
});
|
||||
|
||||
self.sink_pad.prepare(&JitterBufferPadSinkHandler).await;
|
||||
|
||||
gst_info!(CAT, obj: element, "Prepared");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn unprepare(&self, element: &gst::Element) -> Result<(), ()> {
|
||||
let mut state = self.state.lock().await;
|
||||
gst_debug!(CAT, obj: element, "Unpreparing");
|
||||
|
||||
self.sink_pad.unprepare().await;
|
||||
let _ = self.src_pad.unprepare().await;
|
||||
|
||||
state.jbuf.borrow().flush();
|
||||
|
||||
if let Some(wakeup_abort_handle) = state.wakeup_abort_handle.take() {
|
||||
wakeup_abort_handle.abort();
|
||||
}
|
||||
|
||||
gst_debug!(CAT, obj: element, "Unprepared");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn stop(&self, element: &gst::Element) -> Result<(), ()> {
|
||||
let mut state = self.state.lock().await;
|
||||
gst_debug!(CAT, obj: element, "Stopping");
|
||||
|
||||
if let Some(wakeup_abort_handle) = state.wakeup_abort_handle.take() {
|
||||
wakeup_abort_handle.abort();
|
||||
}
|
||||
|
||||
if let Some(abort_handle) = state.task_queue_abort_handle.take() {
|
||||
abort_handle.abort();
|
||||
}
|
||||
|
||||
gst_debug!(CAT, obj: element, "Stopped");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl ObjectSubclass for JitterBuffer {
|
||||
|
|
@ -1320,56 +1387,20 @@ impl ElementImpl for JitterBuffer {
|
|||
gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
|
||||
|
||||
match transition {
|
||||
gst::StateChange::NullToReady => runtime::executor::block_on(async {
|
||||
let mut state = self.state.lock().await;
|
||||
|
||||
let (context, latency) = {
|
||||
let settings = self.settings.lock().await;
|
||||
let context =
|
||||
Context::acquire(&settings.context, settings.context_wait).unwrap();
|
||||
let latency = settings.latency_ms as u64 * gst::MSECOND;
|
||||
(context, latency)
|
||||
};
|
||||
|
||||
state.src_pad_handler = JitterBufferPadSrcHandler::new(latency);
|
||||
|
||||
let _ = self
|
||||
.src_pad
|
||||
.prepare(context, &state.src_pad_handler)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
gst_error_msg!(
|
||||
gst::ResourceError::OpenRead,
|
||||
["Error preparing src_pad: {:?}", err]
|
||||
);
|
||||
gst::StateChange::NullToReady => {
|
||||
runtime::executor::block_on(self.prepare(element)).map_err(|err| {
|
||||
element.post_error_message(&err);
|
||||
gst::StateChangeError
|
||||
});
|
||||
|
||||
self.sink_pad.prepare(&JitterBufferPadSinkHandler).await;
|
||||
}),
|
||||
gst::StateChange::PausedToReady => runtime::executor::block_on(async {
|
||||
let mut state = self.state.lock().await;
|
||||
|
||||
if let Some(wakeup_abort_handle) = state.wakeup_abort_handle.take() {
|
||||
wakeup_abort_handle.abort();
|
||||
})?;
|
||||
}
|
||||
|
||||
if let Some(abort_handle) = state.task_queue_abort_handle.take() {
|
||||
abort_handle.abort();
|
||||
gst::StateChange::PausedToReady => {
|
||||
runtime::executor::block_on(self.stop(element))
|
||||
.map_err(|_| gst::StateChangeError)?;
|
||||
}
|
||||
}),
|
||||
gst::StateChange::ReadyToNull => runtime::executor::block_on(async {
|
||||
let mut state = self.state.lock().await;
|
||||
|
||||
self.sink_pad.unprepare().await;
|
||||
let _ = self.src_pad.unprepare().await;
|
||||
|
||||
state.jbuf.borrow().flush();
|
||||
|
||||
if let Some(wakeup_abort_handle) = state.wakeup_abort_handle.take() {
|
||||
wakeup_abort_handle.abort();
|
||||
gst::StateChange::ReadyToNull => {
|
||||
runtime::executor::block_on(self.unprepare(element))
|
||||
.map_err(|_| gst::StateChangeError)?;
|
||||
}
|
||||
}),
|
||||
_ => (),
|
||||
}
|
||||
|
||||
|
|
|
|||
172
gst-plugin-threadshare/tests/jitterbuffer.rs
Normal file
172
gst-plugin-threadshare/tests/jitterbuffer.rs
Normal file
|
|
@ -0,0 +1,172 @@
|
|||
// Copyright (C) 2020 François Laignel <fengalin@free.fr>
|
||||
//
|
||||
// This library is free software; you can redistribute it and/or
|
||||
// modify it under the terms of the GNU Library General Public
|
||||
// License as published by the Free Software Foundation; either
|
||||
// version 2 of the License, or (at your option) any later version.
|
||||
//
|
||||
// This library is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
||||
// Library General Public License for more details.
|
||||
//
|
||||
// You should have received a copy of the GNU Library General Public
|
||||
// License along with this library; if not, write to the
|
||||
// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
|
||||
// Boston, MA 02110-1335, USA.
|
||||
|
||||
use gst;
|
||||
use gst::gst_debug;
|
||||
use gst::prelude::*;
|
||||
|
||||
use lazy_static::lazy_static;
|
||||
|
||||
use std::sync::mpsc;
|
||||
|
||||
use gstthreadshare;
|
||||
|
||||
lazy_static! {
|
||||
static ref CAT: gst::DebugCategory = gst::DebugCategory::new(
|
||||
"ts-test",
|
||||
gst::DebugColorFlags::empty(),
|
||||
Some("Thread-sharing test"),
|
||||
);
|
||||
}
|
||||
|
||||
fn init() {
|
||||
use std::sync::Once;
|
||||
static INIT: Once = Once::new();
|
||||
|
||||
INIT.call_once(|| {
|
||||
gst::init().unwrap();
|
||||
gstthreadshare::plugin_register_static().expect("gstthreadshare jitterbuffer test");
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jb_pipeline() {
|
||||
init();
|
||||
|
||||
const CONTEXT_WAIT: u32 = 20;
|
||||
const LATENCY: u32 = 20;
|
||||
const BUFFER_NB: i32 = 3;
|
||||
|
||||
let pipeline = gst::Pipeline::new(None);
|
||||
|
||||
let src = gst::ElementFactory::make("audiotestsrc", Some("audiotestsrc")).unwrap();
|
||||
src.set_property("is-live", &true).unwrap();
|
||||
src.set_property("num-buffers", &BUFFER_NB).unwrap();
|
||||
|
||||
let enc = gst::ElementFactory::make("alawenc", Some("alawenc")).unwrap();
|
||||
let pay = gst::ElementFactory::make("rtppcmapay", Some("rtppcmapay")).unwrap();
|
||||
|
||||
let jb = gst::ElementFactory::make("ts-jitterbuffer", Some("ts-jitterbuffer")).unwrap();
|
||||
jb.set_property("context", &"jb_pipeline").unwrap();
|
||||
jb.set_property("context-wait", &CONTEXT_WAIT).unwrap();
|
||||
jb.set_property("latency", &LATENCY).unwrap();
|
||||
|
||||
let depay = gst::ElementFactory::make("rtppcmadepay", Some("rtppcmadepay")).unwrap();
|
||||
let dec = gst::ElementFactory::make("alawdec", Some("alawdec")).unwrap();
|
||||
|
||||
let sink = gst::ElementFactory::make("appsink", Some("appsink")).unwrap();
|
||||
sink.set_property("sync", &false).unwrap();
|
||||
sink.set_property("async", &false).unwrap();
|
||||
sink.set_property("emit-signals", &true).unwrap();
|
||||
|
||||
pipeline
|
||||
.add_many(&[&src, &enc, &pay, &jb, &depay, &dec, &sink])
|
||||
.unwrap();
|
||||
gst::Element::link_many(&[&src, &enc, &pay, &jb, &depay, &dec, &sink]).unwrap();
|
||||
|
||||
let appsink = sink.dynamic_cast::<gst_app::AppSink>().unwrap();
|
||||
let (sender, receiver) = mpsc::channel();
|
||||
appsink.connect_new_sample(move |appsink| {
|
||||
let _sample = appsink
|
||||
.emit("pull-sample", &[])
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.get::<gst::Sample>()
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
|
||||
sender.send(()).unwrap();
|
||||
Ok(gst::FlowSuccess::Ok)
|
||||
});
|
||||
|
||||
pipeline.set_state(gst::State::Playing).unwrap();
|
||||
|
||||
gst_debug!(CAT, "jb_pipeline: waiting for {} buffers", BUFFER_NB);
|
||||
for idx in 0..BUFFER_NB {
|
||||
receiver.recv().unwrap();
|
||||
gst_debug!(CAT, "jb_pipeline: received buffer #{}", idx);
|
||||
}
|
||||
|
||||
pipeline.set_state(gst::State::Null).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn jb_ts_pipeline() {
|
||||
init();
|
||||
|
||||
const CONTEXT_WAIT: u32 = 20;
|
||||
const LATENCY: u32 = 20;
|
||||
const BUFFER_NB: i32 = 3;
|
||||
|
||||
let pipeline = gst::Pipeline::new(None);
|
||||
|
||||
let src = gst::ElementFactory::make("audiotestsrc", Some("audiotestsrc")).unwrap();
|
||||
src.set_property("is-live", &true).unwrap();
|
||||
src.set_property("num-buffers", &BUFFER_NB).unwrap();
|
||||
|
||||
let queue = gst::ElementFactory::make("ts-queue", Some("ts-queue")).unwrap();
|
||||
queue
|
||||
.set_property("context", &"jb_ts_pipeline_queue")
|
||||
.unwrap();
|
||||
queue.set_property("context-wait", &CONTEXT_WAIT).unwrap();
|
||||
|
||||
let enc = gst::ElementFactory::make("alawenc", Some("alawenc")).unwrap();
|
||||
let pay = gst::ElementFactory::make("rtppcmapay", Some("rtppcmapay")).unwrap();
|
||||
|
||||
let jb = gst::ElementFactory::make("ts-jitterbuffer", Some("ts-jitterbuffer")).unwrap();
|
||||
jb.set_property("context", &"jb_ts_pipeline").unwrap();
|
||||
jb.set_property("context-wait", &CONTEXT_WAIT).unwrap();
|
||||
jb.set_property("latency", &LATENCY).unwrap();
|
||||
|
||||
let depay = gst::ElementFactory::make("rtppcmadepay", Some("rtppcmadepay")).unwrap();
|
||||
let dec = gst::ElementFactory::make("alawdec", Some("alawdec")).unwrap();
|
||||
|
||||
let sink = gst::ElementFactory::make("appsink", Some("appsink")).unwrap();
|
||||
sink.set_property("sync", &false).unwrap();
|
||||
sink.set_property("async", &false).unwrap();
|
||||
sink.set_property("emit-signals", &true).unwrap();
|
||||
|
||||
pipeline
|
||||
.add_many(&[&src, &queue, &enc, &pay, &jb, &depay, &dec, &sink])
|
||||
.unwrap();
|
||||
gst::Element::link_many(&[&src, &queue, &enc, &pay, &jb, &depay, &dec, &sink]).unwrap();
|
||||
|
||||
let appsink = sink.dynamic_cast::<gst_app::AppSink>().unwrap();
|
||||
let (sender, receiver) = mpsc::channel();
|
||||
appsink.connect_new_sample(move |appsink| {
|
||||
let _sample = appsink
|
||||
.emit("pull-sample", &[])
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.get::<gst::Sample>()
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
|
||||
sender.send(()).unwrap();
|
||||
Ok(gst::FlowSuccess::Ok)
|
||||
});
|
||||
|
||||
pipeline.set_state(gst::State::Playing).unwrap();
|
||||
|
||||
gst_debug!(CAT, "jb_ts_pipeline: waiting for {} buffers", BUFFER_NB);
|
||||
for idx in 0..BUFFER_NB {
|
||||
receiver.recv().unwrap();
|
||||
gst_debug!(CAT, "jb_ts_pipeline: received buffer #{}", idx);
|
||||
}
|
||||
|
||||
pipeline.set_state(gst::State::Null).unwrap();
|
||||
}
|
||||
Loading…
Reference in a new issue