I am trying to record a live webrtc stream of vp8 rtp packets into a webm file. I am facing two issues 1)The output webm video has significant artificats in between at regular intervals. 2) The pipeline takes a while to get to playing state, like it sometimes takes 10-15 seconds after the webrtc video call has started. I am new to gstreamer and most likely i am making a mistake in setting up the pipeline. I went through the documentation of each of the elements i am using below on the gstreamer website and tried a bunch of properties and settings, but not able to solve these 2 issues. This is my current setup. Please help!
My gstreamer pipeline setup
gst::init()?;
let pipeline = gst::Pipeline::default();
let appsrc = AppSrc::builder().build();
appsrc.set_property("is-live", true);
appsrc.set_property("format", gst::Format::Time);
appsrc.set_property("max-bytes", 50_000_000u64);
appsrc.set_property("max-buffers", 50_000u64);
appsrc.set_property("max-time", 10_000_000_000u64);
let caps_str = "application/x-rtp,media=video,encoding-name=VP8,payload=96,clock-rate=90000";
let caps = gst::Caps::from_str(caps_str)?;
appsrc.set_caps(Some(&caps));
let jitter_buffer = gst::ElementFactory::make("rtpjitterbuffer").name("jitter-buffer").build().expect("Could not create jitter buffer");
jitter_buffer.set_property("latency", 1000u32);
jitter_buffer.set_property("drop-on-latency", false);
jitter_buffer.set_property("do-lost", true);
jitter_buffer.set_property("max-misorder-time", 3000u32);
jitter_buffer.set_property("max-dropout-time", 120000u32);
let depay = gst::ElementFactory::make("rtpvp8depay").name("depay").build().expect("Could not create depay element");
let muxer = gst::ElementFactory::make("webmmux").name("muxer").build().expect("Could not create muxer element");;
muxer.set_property("streamable", true);
muxer.set_property("max-cluster-duration", 1_000_000_000i64);
let queue1 = gst::ElementFactory::make("queue").name("queue1").build().expect("Could not create queue element");
queue1.set_property("leaky", &leaky_value);
queue1.set_property("max-size-buffers", 0u32);
queue1.set_property("max-size-bytes", 0u32);
queue1.set_property("max-size-time", 0u64);
queue1.set_property("notify-levels", true);
queue1.set_property("flush-on-eos", true);
let queue2 = gst::ElementFactory::make("queue").name("queue2").build().expect("Could not create queue element");
queue2.set_property("leaky", &leaky_value);
queue2.set_property("max-size-buffers", 0u32);
queue2.set_property("max-size-bytes", 0u32);
queue2.set_property("max-size-time", 0u64);
queue2.set_property("notify-levels", true);
queue2.set_property("flush-on-eos", true);
let video_sink = AppSink::builder().build();
video_sink.set_property("emit-signals", true);
video_sink.set_property("sync", false);
video_sink.set_property("max-buffers", 0u32);
video_sink.set_property("drop", false);
pipeline.add_many(&[
appsrc.upcast_ref::<gst::Element>(),
jitter_buffer.upcast_ref::<gst::Element>(),
depay.upcast_ref::<gst::Element>(),
queue1.upcast_ref::<gst::Element>(),
muxer.upcast_ref::<gst::Element>(),
queue2.upcast_ref::<gst::Element>(),
video_sink.upcast_ref::<gst::Element>(),
])?;
gst::Element::link_many(&[
appsrc.upcast_ref::<gst::Element>(),
jitter_buffer.upcast_ref::<gst::Element>(),
depay.upcast_ref::<gst::Element>(),
queue1.upcast_ref::<gst::Element>(),
muxer.upcast_ref::<gst::Element>(),
queue2.upcast_ref::<gst::Element>(),
video_sink.upcast_ref::<gst::Element>(),
])?;
pipeline.use_clock(Some(&gst::SystemClock::obtain()));
pipeline.set_start_time(gst::ClockTime::NONE);
pipeline.set_base_time(gst::ClockTime::ZERO);
video_sink.set_callbacks(
AppSinkCallbacks::builder()
.new_sample(move |appsink| {
if stopped_for_cb_clone.load(Ordering::SeqCst) {
return Ok(gst::FlowSuccess::Ok);
}
let sample = match appsink.pull_sample() {
Ok(s) => s
,
Err(e) if e.to_string().contains("EOS") => {
return Ok(gst::FlowSuccess::Ok);
}
Err(e) => {
return Err(gst::FlowError::Error);
}
};
if let Some(buffer) = sample.buffer() {
// push to s3 to save webm recording
);
This is how i push live webrtc rtp packets to appsrc buffer.
loop {
tokio::select! {
maybe_data = async {
let mut rx = rx_clone.lock().await;
rx.recv().await
} => {
match maybe_data {
Some(rtp_packet) => {
if let Some(pipeline) = pipeline_weak.upgrade() {
let timestamp = rtp_packet.header.timestamp;
if first_timestamp.is_none() {
first_timestamp = Some(timestamp as u64);
let mut base_time = base_time_clone.lock().unwrap();
*base_time = Some(timestamp as u64);
}
let relative_timestamp = timestamp as u64 - first_timestamp.unwrap();
let duration = gst::ClockTime::SECOND
.mul_div_floor(1, u64::from(FPS))
.expect("Duration calculation overflow");
let pts = gst::ClockTime::SECOND
.mul_div_floor(relative_timestamp as u64, u64::from(SAMPLE_RATE))
.expect("Timestamp calculation overflow");
let csrc_count = rtp_packet.header.csrc.len().min(15) as u8;
let pad_len = if rtp_packet.header.padding { 4 } else { 0 };
let mut buffer = gst::Buffer::new_rtp_with_sizes(
rtp_packet.payload.len() as u32,
pad_len as u8,
csrc_count
).expect("Failed to create RTP buffer");
{
let buffer_mut = buffer.get_mut().unwrap();
buffer_mut.set_pts(pts);
buffer_mut.set_dts(pts);
buffer_mut.set_duration(duration);
if first_timestamp.unwrap() == timestamp as u64 {
buffer_mut.set_flags(gst::BufferFlags::DISCONT);
}
}
{
let buffer_mut = buffer.get_mut().unwrap();
let mut rtp_buffer = gstreamer_rtp::RTPBuffer::from_buffer_writable(buffer_mut)
.expect("Failed to map RTP buffer");
let payload = rtp_buffer.payload_mut()
.expect("Failed to get payload buffer");
payload.copy_from_slice(&rtp_packet.payload);
rtp_buffer.set_ssrc(rtp_packet.header.ssrc);
rtp_buffer.set_seq(rtp_packet.header.sequence_number);
rtp_buffer.set_timestamp(rtp_packet.header.timestamp);
rtp_buffer.set_payload_type(rtp_packet.header.payload_type);
rtp_buffer.set_marker(rtp_packet.header.marker);
rtp_buffer.set_padding(rtp_packet.header.padding);
rtp_buffer.set_extension(rtp_packet.header.extension);
for (idx, csrc) in rtp_packet.header.csrc.iter().enumerate() {
if idx > 15 { break; } // Maximum of 15 CSRC identifiers
rtp_buffer.set_csrc(idx as u8, *csrc);
}
if rtp_packet.header.extension {
for ext in &rtp_packet.header.extensions {
rtp_buffer.add_extension_twobytes_header(0, ext.id, &ext.payload)
.expect("Failed to add extension header");
}
}
}
let flow_ret = appsrc_clone.push_buffer(buffer);[![enter image description here][1]][1]
