mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2024-12-22 18:16:28 +00:00
Code Issues Projects Releases Wiki Activity

Add livesync plugin

Browse source 
It attempts to produce a (nearly) gapless live stream by synchronizing
its output to the running time and forwarding the next input buffer if
its start is (nearly) flush with the end of the last output buffer.

If the input buffer is missing or too far in the future, it duplicates
the last output buffer with adjusted timestamps. If it is operating on a
raw audio stream, it will fill duplicate buffers with silence.

If an input buffer arrives too late, it is thrown away. If the last
input buffer was accepted too long ago (according to `late-threshold`),
a late input buffer is accepted anyway, but immediately considered a
duplicate. Due to the silence-filling, this has no effect on audio, but
video gets a "slideshow" effect instead of freezing completely.

The "many-repeats" property will be notified when this element has
recently duplicated a lot of buffers or recovered from such a state.

Co-authored-by: Vivia Nikolaidou <vivia@ahiru.eu>
Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/708>
This commit is contained in:
Jan Alexander Steffens (heftig) 2022-10-12 15:35:11 +02:00 committed by Vivia Nikolaidou
parent 473e7d951b
commit 42385c81be
12 changed files with 1843 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.toml
View file

@ -36,6 +36,7 @@ members = [
"text/wrap",
"utils/fallbackswitch",
"utils/livesync",
"utils/togglerecord",
"utils/tracers",
"utils/uriplaylistbin",
@ -84,6 +85,7 @@ default-members = [
"text/wrap",
"utils/fallbackswitch",
"utils/livesync",
"utils/togglerecord",
"utils/tracers",
"utils/uriplaylistbin",

3
README.md
View file

@ -127,6 +127,9 @@ You will find the following plugins in this repository:
configuring a fallback audio/video if there are problems with the main
source.
- `livesync`: Element to maintain a continuous live stream from a
potentially unstable source.
- `togglerecord`: Element to enable starting and stopping multiple streams together.
- `tracers`: Plugin with multiple tracers:

136
docs/plugins/gst_plugins_cache.json
View file

@ -2497,6 +2497,142 @@
"tracers": {},
"url": "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs"
},
"livesync": {
"description": "Livesync Plugin",
"elements": {
"livesync": {
"author": "Jan Alexander Steffens (heftig) <jan.steffens@ltnglobal.com>",
"description": "Outputs livestream, inserting gap frames when input lags",
"hierarchy": [
"GstLiveSync",
"GstElement",
"GstObject",
"GInitiallyUnowned",
"GObject"
],
"klass": "Filter",
"long-name": "Live Synchronizer",
"pad-templates": {
"sink": {
"caps": "ANY",
"direction": "sink",
"presence": "always"
},
"src": {
"caps": "ANY",
"direction": "src",
"presence": "always"
}
},
"properties": {
"drop": {
"blurb": "Number of incoming frames dropped",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "0",
"max": "18446744073709551615",
"min": "0",
"mutable": "null",
"readable": true,
"type": "guint64",
"writable": false
},
"duplicate": {
"blurb": "Number of outgoing frames duplicated",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "0",
"max": "18446744073709551615",
"min": "0",
"mutable": "null",
"readable": true,
"type": "guint64",
"writable": false
},
"in": {
"blurb": "Number of incoming frames accepted",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "0",
"max": "18446744073709551615",
"min": "0",
"mutable": "null",
"readable": true,
"type": "guint64",
"writable": false
},
"late-threshold": {
"blurb": "Maximum time spent (in nanoseconds) before accepting one late buffer; -1 = never",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "2000000000",
"max": "18446744073709551615",
"min": "1000000000",
"mutable": "playing",
"readable": true,
"type": "guint64",
"writable": true
},
"latency": {
"blurb": "Additional latency to allow upstream to take longer to produce buffers for the current position (in nanoseconds)",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "0",
"max": "9223372036854775807",
"min": "0",
"mutable": "playing",
"readable": true,
"type": "guint64",
"writable": true
},
"out": {
"blurb": "Number of outgoing frames produced",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "0",
"max": "18446744073709551615",
"min": "0",
"mutable": "null",
"readable": true,
"type": "guint64",
"writable": false
},
"single-segment": {
"blurb": "Timestamp buffers and eat segments so as to appear as one segment",
"conditionally-available": false,
"construct": false,
"construct-only": false,
"controllable": false,
"default": "false",
"mutable": "ready",
"readable": true,
"type": "gboolean",
"writable": true
}
},
"rank": "none"
}
},
"filename": "gstlivesync",
"license": "MPL",
"other-types": {},
"package": "gst-plugin-livesync",
"source": "gst-plugin-livesync",
"tracers": {},
"url": "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs"
},
"mp4": {
"description": "GStreamer Rust MP4 Plugin",
"elements": {

1
meson.build
View file

@ -67,6 +67,7 @@ plugins = {
'gst-plugin-textwrap': 'libgsttextwrap',
'gst-plugin-fallbackswitch': 'libgstfallbackswitch',
'gst-plugin-livesync': 'libgstlivesync',
'gst-plugin-togglerecord': 'libgsttogglerecord',
'gst-plugin-tracers': 'libgstrstracers',
'gst-plugin-uriplaylistbin': 'libgsturiplaylistbin',

58
utils/livesync/Cargo.toml Normal file
View file

@ -0,0 +1,58 @@
[package]
name = "gst-plugin-livesync"
version = "0.9.0-alpha.1"
authors = ["Jan Alexander Steffens (heftig) <jan.steffens@ltnglobal.com>"]
license = "MPL-2.0"
description = "Livesync Plugin"
repository = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs"
edition = "2021"
rust-version = "1.63"
[dependencies]
gio = { git = "https://github.com/gtk-rs/gtk-rs-core", optional = true }
gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
gst-audio = { package = "gstreamer-audio", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
gst-plugin-gtk4 = { path = "../../video/gtk4", optional = true }
gtk = { package = "gtk4", git = "https://github.com/gtk-rs/gtk4-rs", optional = true }
muldiv = "1.0"
num-rational = { version = "0.4", default-features = false, features = [] }
once_cell = "1.0"
parking_lot = "0.12"
[dev-dependencies]
gst-check = { package = "gstreamer-check", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
[lib]
name = "gstlivesync"
crate-type = ["cdylib", "rlib"]
path = "src/lib.rs"
[[example]]
name = "gtk-livesync"
path = "examples/gtk_livesync.rs"
required-features = ["gtk", "gio", "gst-plugin-gtk4"]
[[test]]
name = "livesync"
path = "tests/livesync.rs"
[build-dependencies]
gst-plugin-version-helper = { path="../../version-helper" }
[features]
static = []
capi = []
doc = ["gst/v1_18"]
[package.metadata.capi]
min_version = "0.8.0"
[package.metadata.capi.header]
enabled = false
[package.metadata.capi.library]
install_subdir = "gstreamer-1.0"
versioning = false
[package.metadata.capi.pkg_config]
requires_private = "gstreamer-1.0, gstreamer-audio-1.0, gobject-2.0, glib-2.0, gmodule-2.0"

1
utils/livesync/LICENSE-MPL-2.0 Symbolic link
View file

@ -0,0 +1 @@
../../LICENSE-MPL-2.0

3
utils/livesync/build.rs Normal file
View file

@ -0,0 +1,3 @@
fn main() {
gst_plugin_version_helper::info()
}

175
utils/livesync/examples/gtk_livesync.rs Normal file
View file

@ -0,0 +1,175 @@
// Copyright (C) 2022 LTN Global Communications, Inc.
// Contact: Jan Alexander Steffens (heftig) <jan.steffens@ltnglobal.com>
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
// If a copy of the MPL was not distributed with this file, You can obtain one at
// <https://mozilla.org/MPL/2.0/>.
//
// SPDX-License-Identifier: MPL-2.0
use gio::prelude::*;
use gst::{glib, prelude::*};
use gtk::prelude::*;
use std::cell::Cell;
struct DroppingProbe(glib::WeakRef<gst::Pad>, Option<gst::PadProbeId>);
impl DroppingProbe {
fn install(pad: &gst::Pad) -> Self {
let probe_id = pad
.add_probe(gst::PadProbeType::BUFFER, |_, _| gst::PadProbeReturn::Drop)
.unwrap();
Self(pad.downgrade(), Some(probe_id))
}
}
impl Drop for DroppingProbe {
fn drop(&mut self) {
if let Some((pad, probe_id)) = self.0.upgrade().zip(self.1.take()) {
pad.remove_probe(probe_id);
}
}
}
fn create_pipeline() -> gst::Pipeline {
gst::parse_launch(
r#"videotestsrc name=vsrc is-live=1
! video/x-raw,framerate=60/1,width=800,height=600
! timeoverlay text="Pre:"
! queue
! livesync latency=50000000
! videorate
! timeoverlay text="Post:" halignment=right
! queue
! gtk4paintablesink name=vsink
audiotestsrc name=asrc is-live=1
! audio/x-raw,channels=2
! queue
! livesync latency=50000000
! audiorate
! queue
! autoaudiosink
"#,
)
.expect("Failed to create pipeline")
.downcast()
.unwrap()
}
fn create_window(app: &gtk::Application) {
let pipeline = create_pipeline();
let video_src_pad = pipeline.by_name("vsrc").unwrap().static_pad("src").unwrap();
let audio_src_pad = pipeline.by_name("asrc").unwrap().static_pad("src").unwrap();
let window = gtk::ApplicationWindow::new(app);
window.set_default_size(800, 684);
let vbox = gtk::Box::new(gtk::Orientation::Vertical, 0);
let picture = gtk::Picture::new();
let paintable = pipeline
.by_name("vsink")
.unwrap()
.property::<gtk::gdk::Paintable>("paintable");
picture.set_paintable(Some(&paintable));
vbox.append(&picture);
let action_bar = gtk::ActionBar::new();
vbox.append(&action_bar);
let offset_spin = gtk::SpinButton::with_range(0.0, 500.0, 100.0);
action_bar.pack_start(&offset_spin);
{
let video_src_pad = video_src_pad.clone();
let audio_src_pad = audio_src_pad.clone();
offset_spin.connect_value_notify(move |offset_spin| {
const MSECOND: f64 = gst::ClockTime::MSECOND.nseconds() as _;
let offset = (offset_spin.value() * -MSECOND) as i64;
video_src_pad.set_offset(offset);
audio_src_pad.set_offset(offset);
});
}
let drop_button = gtk::ToggleButton::with_label("Drop Signal");
action_bar.pack_end(&drop_button);
let drop_ids = Cell::new(None);
drop_button.connect_toggled(move |drop_button| {
if drop_button.is_active() {
let video_probe = DroppingProbe::install(&video_src_pad);
let audio_probe = DroppingProbe::install(&audio_src_pad);
drop_ids.set(Some((video_probe, audio_probe)));
} else {
drop_ids.set(None);
}
});
{
let bus = pipeline.bus().unwrap();
let window = window.downgrade();
bus.add_watch_local(move |_, msg| {
use gst::MessageView;
match msg.view() {
MessageView::Eos(..) => {
if let Some(window) = window.upgrade() {
window.close();
}
}
MessageView::Error(err) => {
eprintln!(
"Error from {}: {} ({:?})",
msg.src().map(|s| s.path_string()).as_deref().unwrap_or(""),
err.error(),
err.debug(),
);
if let Some(window) = window.upgrade() {
window.close();
}
}
_ => (),
};
glib::Continue(true)
})
.unwrap();
}
{
let pipeline = pipeline.clone();
window.connect_realize(move |_| {
pipeline
.set_state(gst::State::Playing)
.expect("Failed to start pipeline");
});
}
window.connect_unrealize(move |_| {
pipeline
.set_state(gst::State::Null)
.expect("Failed to stop pipeline");
});
window.set_child(Some(&vbox));
window.show();
}
fn main() {
let app = gtk::Application::new(
Some("gtk-plugins-rs.gtk-livesync"),
gio::ApplicationFlags::FLAGS_NONE,
);
app.connect_startup(move |_app| {
gst::init().expect("Failed to initialize GStreamer");
gstlivesync::plugin_register_static().expect("Failed to register livesync plugin");
gstgtk4::plugin_register_static().expect("Failed to register gstgtk4 plugin");
});
app.connect_activate(create_window);
app.run();
}

36
utils/livesync/src/lib.rs Normal file
View file

@ -0,0 +1,36 @@
// Copyright (C) 2022 LTN Global Communications, Inc.
// Contact: Jan Alexander Steffens (heftig) <jan.steffens@ltnglobal.com>
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
// If a copy of the MPL was not distributed with this file, You can obtain one at
// <https://mozilla.org/MPL/2.0/>.
//
// SPDX-License-Identifier: MPL-2.0
#![allow(clippy::non_send_fields_in_send_ty, unused_doc_comments)]
/**
* plugin-livesync:
*
* Since: plugins-rs-0.9.0
*/
use gst::glib;
mod livesync;
fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
livesync::register(plugin)
}
gst::plugin_define!(
livesync,
env!("CARGO_PKG_DESCRIPTION"),
plugin_init,
concat!(env!("CARGO_PKG_VERSION"), "-", env!("COMMIT_ID")),
// FIXME: MPL-2.0 is only allowed since 1.18.3 (as unknown) and 1.20 (as known)
"MPL",
env!("CARGO_PKG_NAME"),
env!("CARGO_PKG_NAME"),
env!("CARGO_PKG_REPOSITORY"),
env!("BUILD_REL_DATE")
);

1202
utils/livesync/src/livesync/imp.rs Normal file
View file

File diff suppressed because it is too large Load diff

28
utils/livesync/src/livesync/mod.rs Normal file
View file

@ -0,0 +1,28 @@
// Copyright (C) 2022 LTN Global Communications, Inc.
// Contact: Jan Alexander Steffens (heftig) <jan.steffens@ltnglobal.com>
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
// If a copy of the MPL was not distributed with this file, You can obtain one at
// <https://mozilla.org/MPL/2.0/>.
//
// SPDX-License-Identifier: MPL-2.0
use gst::{glib, prelude::*};
mod imp;
glib::wrapper! {
pub struct LiveSync(ObjectSubclass<imp::LiveSync>)
@extends gst::Element, gst::Object;
}
pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
gst::Element::register(
Some(plugin),
"livesync",
gst::Rank::None,
LiveSync::static_type(),
)?;
Ok(())
}

198
utils/livesync/tests/livesync.rs Normal file
View file

@ -0,0 +1,198 @@
// Copyright (C) 2022 LTN Global Communications, Inc.
// Contact: Jan Alexander Steffens (heftig) <jan.steffens@ltnglobal.com>
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
// If a copy of the MPL was not distributed with this file, You can obtain one at
// <https://mozilla.org/MPL/2.0/>.
//
// SPDX-License-Identifier: MPL-2.0
use gst::prelude::*;
fn init() {
use std::sync::Once;
static INIT: Once = Once::new();
INIT.call_once(|| {
gst::init().unwrap();
gstlivesync::plugin_register_static().expect("Failed to register livesync plugin");
});
}
const DURATION: gst::ClockTime = gst::ClockTime::from_mseconds(100);
const LATENCY: gst::ClockTime = gst::ClockTime::from_mseconds(200);
const SEGMENT_OFFSET: gst::ClockTime = gst::ClockTime::from_seconds(60 * 60 * 1000);
fn crank_pull(harness: &mut gst_check::Harness) -> gst::Buffer {
harness.crank_single_clock_wait().unwrap();
harness.pull().unwrap()
}
#[track_caller]
fn assert_buf(
buf: &gst::BufferRef,
offset: u64,
pts: gst::ClockTime,
duration: gst::ClockTime,
flags: gst::BufferFlags,
) {
assert_eq!(buf.offset(), offset, "Bad offset");
assert_eq!(buf.pts(), Some(pts), "Bad PTS");
assert_eq!(buf.duration(), Some(duration), "Bad duration");
assert_eq!(
buf.flags() - gst::BufferFlags::TAG_MEMORY,
flags,
"Bad flags",
);
}
#[track_caller]
fn assert_crank_pull(
harness: &mut gst_check::Harness,
offset_per_buffer: u64,
src_buffer_number: u64,
sink_buffer_number: u64,
flags: gst::BufferFlags,
singlesegment: bool,
) {
let pts = if singlesegment {
LATENCY + DURATION * sink_buffer_number + SEGMENT_OFFSET
} else {
DURATION * sink_buffer_number
};
assert_buf(
&crank_pull(harness),
offset_per_buffer * src_buffer_number,
pts,
DURATION,
flags,
);
}
#[test]
fn test_video_singlesegment() {
test_video(true);
}
#[test]
fn test_audio_singlesegment() {
test_audio(true);
}
#[test]
fn test_video_nonsinglesegment() {
test_video(false);
}
#[test]
fn test_audio_nonsinglesegment() {
test_audio(false);
}
fn test_video(singlesegment: bool) {
init();
let mut h = gst_check::Harness::new("livesync");
h.add_src_parse(
r"videotestsrc is-live=1
! capsfilter caps=video/x-raw,framerate=10/1
",
true,
);
let element = h.element().unwrap();
element.set_property("latency", LATENCY);
element.set_property("single-segment", singlesegment);
test_livesync(&mut h, 1, singlesegment);
}
fn test_audio(singlesegment: bool) {
init();
let mut h = gst_check::Harness::new("livesync");
h.add_src_parse(
r"audiotestsrc is-live=1 samplesperbuffer=4800
! capsfilter caps=audio/x-raw,rate=48000
",
true,
);
let element = h.element().unwrap();
element.set_property("latency", LATENCY);
element.set_property("single-segment", singlesegment);
test_livesync(&mut h, 4800, singlesegment);
}
fn test_livesync(h: &mut gst_check::Harness, o: u64, singlesegment: bool) {
// Normal operation ------------------------------
// Push frames 0-1, pull frame 0
h.push_from_src().unwrap();
h.push_from_src().unwrap();
assert_eq!(h.pull_event().unwrap().type_(), gst::EventType::StreamStart);
assert_eq!(h.pull_event().unwrap().type_(), gst::EventType::Caps);
assert_eq!(h.pull_event().unwrap().type_(), gst::EventType::Segment);
assert_crank_pull(h, o, 0, 0, gst::BufferFlags::DISCONT, singlesegment);
// Push frames 2-10, pull frames 1-9
for i in 1..=9 {
h.push_from_src().unwrap();
assert_crank_pull(h, o, i, i, gst::BufferFlags::empty(), singlesegment);
}
// Pull frame 10
assert_crank_pull(h, o, 10, 10, gst::BufferFlags::empty(), singlesegment);
// Bridging gap ----------------------------------
// Pull frames 11-19
for i in 11..=19 {
assert_crank_pull(h, o, 10, i, gst::BufferFlags::GAP, singlesegment);
}
// Push frames 11-19
for _ in 11..=19 {
h.push_from_src().unwrap();
}
// Normal operation ------------------------------
// Push frames 20-21, pull frame 20
for _ in 1..=2 {
let mut src_h = h.src_harness_mut().unwrap();
src_h.crank_single_clock_wait().unwrap();
let mut buf = src_h.pull().unwrap();
let buf_mut = buf.make_mut();
buf_mut.set_flags(gst::BufferFlags::MARKER);
h.push(buf).unwrap();
}
assert_crank_pull(h, o, 10, 20, gst::BufferFlags::GAP, singlesegment);
// Push frame 22, pull frame 21
h.push_from_src().unwrap();
assert_crank_pull(
h,
o,
21,
21,
gst::BufferFlags::DISCONT | gst::BufferFlags::MARKER,
singlesegment,
);
// Push frames 23-30, pull frames 22-29
for i in 22..=29 {
h.push_from_src().unwrap();
assert_crank_pull(h, o, i, i, gst::BufferFlags::empty(), singlesegment);
}
// EOS -------------------------------------------
assert!(h.push_event(gst::event::Eos::new()));
// Pull frame 30
assert_crank_pull(h, o, 30, 30, gst::BufferFlags::empty(), singlesegment);
assert_eq!(h.pull_event().unwrap().type_(), gst::EventType::Eos);
assert_eq!(h.try_pull(), None);
}