gstreamer: Add accessors for PadProbeDatas on PadProbeInfo

And make use of it in examples and other code.

This allows to simplify usage a bit in most cases.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/-/merge_requests/1324>
This commit is contained in:
Sebastian Dröge 2023-10-16 11:28:18 +03:00
parent 4c8d16d09e
commit 62f58620b7
6 changed files with 144 additions and 77 deletions

View file

@ -75,12 +75,18 @@ fn example_main() {
// Add a pad probe on the sink pad and catch the custom event we sent, then send // Add a pad probe on the sink pad and catch the custom event we sent, then send
// an EOS event on the pipeline. // an EOS event on the pipeline.
sinkpad.add_probe(gst::PadProbeType::EVENT_DOWNSTREAM, move |_, probe_info| { sinkpad.add_probe(gst::PadProbeType::EVENT_DOWNSTREAM, move |_, probe_info| {
match probe_info.data { let Some(event) = probe_info.event() else {
Some(gst::PadProbeData::Event(ref ev)) return gst::PadProbeReturn::Ok;
if ev.type_() == gst::EventType::CustomDownstream => };
{
if let Some(custom_event) = ExampleCustomEvent::parse(ev) { let Some(custom_event) = ExampleCustomEvent::parse(event) else {
if let Some(pipeline) = pipeline_weak.upgrade() { return gst::PadProbeReturn::Ok;
};
let Some(pipeline) = pipeline_weak.upgrade() else {
return gst::PadProbeReturn::Ok;
};
if custom_event.send_eos { if custom_event.send_eos {
/* Send EOS event to shut down the pipeline, but from an async callback, as we're /* Send EOS event to shut down the pipeline, but from an async callback, as we're
* in a pad probe blocking the stream thread here... */ * in a pad probe blocking the stream thread here... */
@ -95,11 +101,7 @@ fn example_main() {
} else { } else {
println!("Got custom event, with send_eos=false. Ignoring"); println!("Got custom event, with send_eos=false. Ignoring");
} }
}
}
}
_ => (),
}
gst::PadProbeReturn::Ok gst::PadProbeReturn::Ok
}); });

View file

@ -305,7 +305,6 @@ fn main() -> Result<()> {
let sinkpad = videosink.static_pad("sink").unwrap(); let sinkpad = videosink.static_pad("sink").unwrap();
let overlay_context_weak = Arc::downgrade(&overlay_context); let overlay_context_weak = Arc::downgrade(&overlay_context);
sinkpad.add_probe(gst::PadProbeType::BUFFER, move |_, probe_info| { sinkpad.add_probe(gst::PadProbeType::BUFFER, move |_, probe_info| {
if let Some(gst::PadProbeData::Buffer(_)) = probe_info.data {
let overlay_context = overlay_context_weak.upgrade().unwrap(); let overlay_context = overlay_context_weak.upgrade().unwrap();
let mut context = overlay_context.lock().unwrap(); let mut context = overlay_context.lock().unwrap();
context.timestamp_queue.push_back(SystemTime::now()); context.timestamp_queue.push_back(SystemTime::now());
@ -317,7 +316,7 @@ fn main() -> Result<()> {
context.avg_fps = 1000f32 * (context.timestamp_queue.len() - 1) as f32 / duration; context.avg_fps = 1000f32 * (context.timestamp_queue.len() - 1) as f32 / duration;
context.timestamp_queue.clear(); context.timestamp_queue.clear();
} }
}
gst::PadProbeReturn::Ok gst::PadProbeReturn::Ok
}); });

View file

@ -38,7 +38,10 @@ fn example_main() {
// This handler gets called for every buffer that passes the pad we probe. // This handler gets called for every buffer that passes the pad we probe.
src_pad.add_probe(gst::PadProbeType::BUFFER, |_, probe_info| { src_pad.add_probe(gst::PadProbeType::BUFFER, |_, probe_info| {
// Interpret the data sent over the pad as one buffer // Interpret the data sent over the pad as one buffer
if let Some(gst::PadProbeData::Buffer(ref buffer)) = probe_info.data { let Some(buffer) = probe_info.buffer() else {
return gst::PadProbeReturn::Ok;
};
// At this point, buffer is only a reference to an existing memory region somewhere. // At this point, buffer is only a reference to an existing memory region somewhere.
// When we want to access its content, we have to map it while requesting the required // When we want to access its content, we have to map it while requesting the required
// mode of access (read, read/write). // mode of access (read, read/write).
@ -67,7 +70,6 @@ fn example_main() {
.sum(); .sum();
let rms = (sum / (samples.len() as f64)).sqrt(); let rms = (sum / (samples.len() as f64)).sqrt();
println!("rms: {rms}"); println!("rms: {rms}");
}
gst::PadProbeReturn::Ok gst::PadProbeReturn::Ok
}); });

View file

@ -96,7 +96,7 @@ fn example_main() {
mixer_src_pad.add_probe(gst::PadProbeType::EVENT_UPSTREAM, move |_, probe_info| { mixer_src_pad.add_probe(gst::PadProbeType::EVENT_UPSTREAM, move |_, probe_info| {
let mixer_sink_pad = mixer_sink_pad_weak.upgrade().unwrap(); let mixer_sink_pad = mixer_sink_pad_weak.upgrade().unwrap();
let Some(gst::PadProbeData::Event(ref ev)) = probe_info.data else { let Some(ev) = probe_info.event() else {
return gst::PadProbeReturn::Ok; return gst::PadProbeReturn::Ok;
}; };

View file

@ -196,11 +196,13 @@ impl StreamProducer {
.add_probe( .add_probe(
gst::PadProbeType::EVENT_UPSTREAM, gst::PadProbeType::EVENT_UPSTREAM,
glib::clone!(@weak appsink, @weak consumer => @default-panic, move |_pad, info| { glib::clone!(@weak appsink, @weak consumer => @default-panic, move |_pad, info| {
if let Some(gst::PadProbeData::Event(ref ev)) = info.data { let Some(event) = info.event() else {
if gst_video::UpstreamForceKeyUnitEvent::parse(ev).is_ok() { return gst::PadProbeReturn::Ok;
};
if gst_video::UpstreamForceKeyUnitEvent::parse(event).is_ok() {
gst::debug!(CAT, obj: &appsink, "Requesting keyframe"); gst::debug!(CAT, obj: &appsink, "Requesting keyframe");
let _ = appsink.send_event(ev.clone()); let _ = appsink.send_event(event.clone());
}
} }
gst::PadProbeReturn::Ok gst::PadProbeReturn::Ok
@ -486,14 +488,18 @@ impl<'a> From<&'a gst_app::AppSink> for StreamProducer {
sinkpad.add_probe( sinkpad.add_probe(
gst::PadProbeType::EVENT_UPSTREAM, gst::PadProbeType::EVENT_UPSTREAM,
glib::clone!(@strong consumers => move |_pad, info| { glib::clone!(@strong consumers => move |_pad, info| {
if let Some(gst::PadProbeData::Event(ref ev)) = info.data { let Some(event) = info.event() else {
if let gst::EventView::Latency(ev) = ev.view() { return gst::PadProbeReturn::Ok;
let latency = ev.latency(); };
let gst::EventView::Latency(event) = event.view() else {
return gst::PadProbeReturn::Ok;
};
let latency = event.latency();
let mut consumers = consumers.lock().unwrap(); let mut consumers = consumers.lock().unwrap();
consumers.current_latency = Some(latency); consumers.current_latency = Some(latency);
consumers.latency_updated = true;
}
}
gst::PadProbeReturn::Ok gst::PadProbeReturn::Ok
}), }),
); );

View file

@ -56,6 +56,64 @@ pub struct PadProbeInfo<'a> {
pub flow_res: Result<FlowSuccess, FlowError>, pub flow_res: Result<FlowSuccess, FlowError>,
} }
impl<'a> PadProbeInfo<'a> {
pub fn buffer(&self) -> Option<&Buffer> {
match self.data {
Some(PadProbeData::Buffer(ref buffer)) => Some(buffer),
_ => None,
}
}
pub fn buffer_mut(&mut self) -> Option<&mut Buffer> {
match self.data {
Some(PadProbeData::Buffer(ref mut buffer)) => Some(buffer),
_ => None,
}
}
pub fn buffer_list(&self) -> Option<&BufferList> {
match self.data {
Some(PadProbeData::BufferList(ref buffer_list)) => Some(buffer_list),
_ => None,
}
}
pub fn buffer_list_mut(&mut self) -> Option<&mut BufferList> {
match self.data {
Some(PadProbeData::BufferList(ref mut buffer_list)) => Some(buffer_list),
_ => None,
}
}
pub fn query(&self) -> Option<&QueryRef> {
match self.data {
Some(PadProbeData::Query(ref query)) => Some(*query),
_ => None,
}
}
pub fn query_mut(&mut self) -> Option<&mut QueryRef> {
match self.data {
Some(PadProbeData::Query(ref mut query)) => Some(*query),
_ => None,
}
}
pub fn event(&self) -> Option<&Event> {
match self.data {
Some(PadProbeData::Event(ref event)) => Some(event),
_ => None,
}
}
pub fn event_mut(&mut self) -> Option<&mut Event> {
match self.data {
Some(PadProbeData::Event(ref mut event)) => Some(event),
_ => None,
}
}
}
#[derive(Debug)] #[derive(Debug)]
pub enum PadProbeData<'a> { pub enum PadProbeData<'a> {
Buffer(Buffer), Buffer(Buffer),