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-11-22 11:30:59 +00:00
Code Issues Projects Releases Wiki Activity

Remove & for obj in log macros

Browse source 
This is no longer necessary.

See https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/-/merge_requests/1137
This commit is contained in:
François Laignel 2022-10-23 20:46:18 +02:00
parent fe8e0a8bf8
commit 86776be58c
26 changed files with 294 additions and 298 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
audio/audiofx/src/ebur128level/imp.rs
View file

@ -130,7 +130,7 @@ impl ObjectImpl for EbuR128Level {
let this = args[0].get::<super::EbuR128Level>().unwrap(); let this = args[0].get::<super::EbuR128Level>().unwrap();
let imp = this.imp(); let imp = this.imp();
gst::info!(CAT, obj: &this, "Resetting measurements",); gst::info!(CAT, obj: this, "Resetting measurements",);
imp.reset.store(true, atomic::Ordering::SeqCst); imp.reset.store(true, atomic::Ordering::SeqCst);
None None

6
generic/sodium/src/decrypter/imp.rs
View file

@ -464,13 +464,13 @@ impl Decrypter {
self.sinkpad.pull_range(pull_offset, total_size).map_err(|err| { self.sinkpad.pull_range(pull_offset, total_size).map_err(|err| {
match err { match err {
gst::FlowError::Flushing => { gst::FlowError::Flushing => {
gst::debug!(CAT, obj: &self.sinkpad, "Pausing after pulling buffer, reason: flushing"); gst::debug!(CAT, obj: self.sinkpad, "Pausing after pulling buffer, reason: flushing");
} }
gst::FlowError::Eos => { gst::FlowError::Eos => {
gst::debug!(CAT, obj: &self.sinkpad, "Eos"); gst::debug!(CAT, obj: self.sinkpad, "Eos");
} }
flow => { flow => {
gst::error!(CAT, obj: &self.sinkpad, "Failed to pull, reason: {:?}", flow); gst::error!(CAT, obj: self.sinkpad, "Failed to pull, reason: {:?}", flow);
} }
}; };

42
generic/threadshare/examples/standalone/sink/imp.rs
View file

@ -84,7 +84,7 @@ impl PadSinkHandler for TestSinkPadHandler {
let sender = elem.imp().clone_item_sender(); let sender = elem.imp().clone_item_sender();
async move { async move {
if sender.send_async(StreamItem::Buffer(buffer)).await.is_err() { if sender.send_async(StreamItem::Buffer(buffer)).await.is_err() {
gst::debug!(CAT, obj: &elem, "Flushing"); gst::debug!(CAT, obj: elem, "Flushing");
return Err(gst::FlowError::Flushing); return Err(gst::FlowError::Flushing);
} }
@ -103,7 +103,7 @@ impl PadSinkHandler for TestSinkPadHandler {
async move { async move {
for buffer in list.iter_owned() { for buffer in list.iter_owned() {
if sender.send_async(StreamItem::Buffer(buffer)).await.is_err() { if sender.send_async(StreamItem::Buffer(buffer)).await.is_err() {
gst::debug!(CAT, obj: &elem, "Flushing"); gst::debug!(CAT, obj: elem, "Flushing");
return Err(gst::FlowError::Flushing); return Err(gst::FlowError::Flushing);
} }
} }
@ -125,7 +125,7 @@ impl PadSinkHandler for TestSinkPadHandler {
let imp = elem.imp(); let imp = elem.imp();
return imp.task.flush_stop().await_maybe_on_context().is_ok(); return imp.task.flush_stop().await_maybe_on_context().is_ok();
} else if sender.send_async(StreamItem::Event(event)).await.is_err() { } else if sender.send_async(StreamItem::Event(event)).await.is_err() {
gst::debug!(CAT, obj: &elem, "Flushing"); gst::debug!(CAT, obj: elem, "Flushing");
} }
true true
@ -441,9 +441,9 @@ impl TaskImpl for TestSinkTask {
self.raise_log_level = settings.raise_log_level; self.raise_log_level = settings.raise_log_level;
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Preparing Task"); gst::log!(CAT, obj: self.element, "Preparing Task");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Preparing Task"); gst::trace!(CAT, obj: self.element, "Preparing Task");
} }
self.stats.must_log = settings.logs_stats; self.stats.must_log = settings.logs_stats;
@ -458,9 +458,9 @@ impl TaskImpl for TestSinkTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async { async {
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Starting Task"); gst::log!(CAT, obj: self.element, "Starting Task");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Starting Task"); gst::trace!(CAT, obj: self.element, "Starting Task");
} }
self.last_dts = None; self.last_dts = None;
@ -473,9 +473,9 @@ impl TaskImpl for TestSinkTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async { async {
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Stopping Task"); gst::log!(CAT, obj: self.element, "Stopping Task");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Stopping Task"); gst::trace!(CAT, obj: self.element, "Stopping Task");
} }
self.flush().await; self.flush().await;
@ -489,9 +489,9 @@ impl TaskImpl for TestSinkTask {
let item = self.item_receiver.next().await.unwrap(); let item = self.item_receiver.next().await.unwrap();
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Popped item"); gst::log!(CAT, obj: self.element, "Popped item");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Popped item"); gst::trace!(CAT, obj: self.element, "Popped item");
} }
Ok(item) Ok(item)
@ -502,9 +502,9 @@ impl TaskImpl for TestSinkTask {
fn handle_item(&mut self, item: StreamItem) -> BoxFuture<'_, Result<(), gst::FlowError>> { fn handle_item(&mut self, item: StreamItem) -> BoxFuture<'_, Result<(), gst::FlowError>> {
async move { async move {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "Received {:?}", item); gst::debug!(CAT, obj: self.element, "Received {:?}", item);
} else { } else {
gst::trace!(CAT, obj: &self.element, "Received {:?}", item); gst::trace!(CAT, obj: self.element, "Received {:?}", item);
} }
match item { match item {
@ -527,28 +527,28 @@ impl TaskImpl for TestSinkTask {
self.stats.add_buffer(latency, interval); self.stats.add_buffer(latency, interval);
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "o latency {:.2?}", latency); gst::debug!(CAT, obj: self.element, "o latency {:.2?}", latency);
gst::debug!(CAT, obj: &self.element, "o interval {:.2?}", interval); gst::debug!(CAT, obj: self.element, "o interval {:.2?}", interval);
} else { } else {
gst::trace!(CAT, obj: &self.element, "o latency {:.2?}", latency); gst::trace!(CAT, obj: self.element, "o latency {:.2?}", latency);
gst::trace!(CAT, obj: &self.element, "o interval {:.2?}", interval); gst::trace!(CAT, obj: self.element, "o interval {:.2?}", interval);
} }
} }
self.last_dts = Some(dts); self.last_dts = Some(dts);
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Buffer processed"); gst::log!(CAT, obj: self.element, "Buffer processed");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Buffer processed"); gst::trace!(CAT, obj: self.element, "Buffer processed");
} }
} }
StreamItem::Event(event) => match event.view() { StreamItem::Event(event) => match event.view() {
EventView::Eos(_) => { EventView::Eos(_) => {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "EOS"); gst::debug!(CAT, obj: self.element, "EOS");
} else { } else {
gst::trace!(CAT, obj: &self.element, "EOS"); gst::trace!(CAT, obj: self.element, "EOS");
} }
let elem = self.element.clone(); let elem = self.element.clone();

54
generic/threadshare/examples/standalone/src/imp.rs
View file

@ -107,9 +107,9 @@ impl TaskImpl for SrcTask {
self.raise_log_level = settings.raise_log_level; self.raise_log_level = settings.raise_log_level;
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Preparing Task"); gst::log!(CAT, obj: self.element, "Preparing Task");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Preparing Task"); gst::trace!(CAT, obj: self.element, "Preparing Task");
} }
self.push_period = settings.push_period; self.push_period = settings.push_period;
@ -123,9 +123,9 @@ impl TaskImpl for SrcTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async { async {
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Starting Task"); gst::log!(CAT, obj: self.element, "Starting Task");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Starting Task"); gst::trace!(CAT, obj: self.element, "Starting Task");
} }
self.timer = Some( self.timer = Some(
@ -146,9 +146,9 @@ impl TaskImpl for SrcTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Stopping Task"); gst::log!(CAT, obj: self.element, "Stopping Task");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Stopping Task"); gst::trace!(CAT, obj: self.element, "Stopping Task");
} }
self.buffer_pool.set_active(false).unwrap(); self.buffer_pool.set_active(false).unwrap();
@ -164,17 +164,17 @@ impl TaskImpl for SrcTask {
fn try_next(&mut self) -> BoxFuture<'_, Result<gst::Buffer, gst::FlowError>> { fn try_next(&mut self) -> BoxFuture<'_, Result<gst::Buffer, gst::FlowError>> {
async move { async move {
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Awaiting timer"); gst::log!(CAT, obj: self.element, "Awaiting timer");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Awaiting timer"); gst::trace!(CAT, obj: self.element, "Awaiting timer");
} }
self.timer.as_mut().unwrap().next().await; self.timer.as_mut().unwrap().next().await;
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Timer ticked"); gst::log!(CAT, obj: self.element, "Timer ticked");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Timer ticked"); gst::trace!(CAT, obj: self.element, "Timer ticked");
} }
self.buffer_pool self.buffer_pool
@ -188,7 +188,7 @@ impl TaskImpl for SrcTask {
buffer buffer
}) })
.map_err(|err| { .map_err(|err| {
gst::error!(CAT, obj: &self.element, "Failed to acquire buffer {}", err); gst::error!(CAT, obj: self.element, "Failed to acquire buffer {}", err);
err err
}) })
} }
@ -201,16 +201,16 @@ impl TaskImpl for SrcTask {
match res { match res {
Ok(_) => { Ok(_) => {
if self.raise_log_level { if self.raise_log_level {
gst::log!(CAT, obj: &self.element, "Successfully pushed buffer"); gst::log!(CAT, obj: self.element, "Successfully pushed buffer");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Successfully pushed buffer"); gst::trace!(CAT, obj: self.element, "Successfully pushed buffer");
} }
} }
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "EOS"); gst::debug!(CAT, obj: self.element, "EOS");
} else { } else {
gst::trace!(CAT, obj: &self.element, "EOS"); gst::trace!(CAT, obj: self.element, "EOS");
} }
let test_src = self.element.imp(); let test_src = self.element.imp();
test_src.src_pad.push_event(gst::event::Eos::new()).await; test_src.src_pad.push_event(gst::event::Eos::new()).await;
@ -219,13 +219,13 @@ impl TaskImpl for SrcTask {
} }
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "Flushing"); gst::debug!(CAT, obj: self.element, "Flushing");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Flushing"); gst::trace!(CAT, obj: self.element, "Flushing");
} }
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Got error {}", err); gst::error!(CAT, obj: self.element, "Got error {}", err);
gst::element_error!( gst::element_error!(
&self.element, &self.element,
gst::StreamError::Failed, gst::StreamError::Failed,
@ -244,18 +244,18 @@ impl TaskImpl for SrcTask {
impl SrcTask { impl SrcTask {
async fn push(&mut self, buffer: gst::Buffer) -> Result<gst::FlowSuccess, gst::FlowError> { async fn push(&mut self, buffer: gst::Buffer) -> Result<gst::FlowSuccess, gst::FlowError> {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "Pushing {:?}", buffer); gst::debug!(CAT, obj: self.element, "Pushing {:?}", buffer);
} else { } else {
gst::trace!(CAT, obj: &self.element, "Pushing {:?}", buffer); gst::trace!(CAT, obj: self.element, "Pushing {:?}", buffer);
} }
let test_src = self.element.imp(); let test_src = self.element.imp();
if self.need_initial_events { if self.need_initial_events {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "Pushing initial events"); gst::debug!(CAT, obj: self.element, "Pushing initial events");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Pushing initial events"); gst::trace!(CAT, obj: self.element, "Pushing initial events");
} }
let stream_id = format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>()); let stream_id = format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>());
@ -283,9 +283,9 @@ impl SrcTask {
} }
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "Forwarding buffer"); gst::debug!(CAT, obj: self.element, "Forwarding buffer");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Forwarding buffer"); gst::trace!(CAT, obj: self.element, "Forwarding buffer");
} }
let ok = test_src.src_pad.push(buffer).await?; let ok = test_src.src_pad.push(buffer).await?;
@ -294,14 +294,14 @@ impl SrcTask {
if self.num_buffers.opt_eq(self.buffer_count).unwrap_or(false) { if self.num_buffers.opt_eq(self.buffer_count).unwrap_or(false) {
if self.raise_log_level { if self.raise_log_level {
gst::debug!(CAT, obj: &self.element, "Pushing EOS"); gst::debug!(CAT, obj: self.element, "Pushing EOS");
} else { } else {
gst::trace!(CAT, obj: &self.element, "Pushing EOS"); gst::trace!(CAT, obj: self.element, "Pushing EOS");
} }
let test_src = self.element.imp(); let test_src = self.element.imp();
if !test_src.src_pad.push_event(gst::event::Eos::new()).await { if !test_src.src_pad.push_event(gst::event::Eos::new()).await {
gst::error!(CAT, obj: &self.element, "Error pushing EOS"); gst::error!(CAT, obj: self.element, "Error pushing EOS");
} }
return Err(gst::FlowError::Eos); return Err(gst::FlowError::Eos);
} }

24
generic/threadshare/src/appsrc/imp.rs
View file

@ -170,11 +170,11 @@ impl AppSrcTask {
} }
async fn push_item(&mut self, item: StreamItem) -> Result<gst::FlowSuccess, gst::FlowError> { async fn push_item(&mut self, item: StreamItem) -> Result<gst::FlowSuccess, gst::FlowError> {
gst::log!(CAT, obj: &self.element, "Handling {:?}", item); gst::log!(CAT, obj: self.element, "Handling {:?}", item);
let appsrc = self.element.imp(); let appsrc = self.element.imp();
if self.need_initial_events { if self.need_initial_events {
gst::debug!(CAT, obj: &self.element, "Pushing initial events"); gst::debug!(CAT, obj: self.element, "Pushing initial events");
let stream_id = format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>()); let stream_id = format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>());
let stream_start_evt = gst::event::StreamStart::builder(&stream_id) let stream_start_evt = gst::event::StreamStart::builder(&stream_id)
@ -204,7 +204,7 @@ impl AppSrcTask {
match item { match item {
StreamItem::Buffer(buffer) => { StreamItem::Buffer(buffer) => {
gst::log!(CAT, obj: &self.element, "Forwarding {:?}", buffer); gst::log!(CAT, obj: self.element, "Forwarding {:?}", buffer);
appsrc.src_pad.push(buffer).await appsrc.src_pad.push(buffer).await
} }
StreamItem::Event(event) => { StreamItem::Event(event) => {
@ -214,7 +214,7 @@ impl AppSrcTask {
Err(gst::FlowError::Eos) Err(gst::FlowError::Eos)
} }
_ => { _ => {
gst::log!(CAT, obj: &self.element, "Forwarding {:?}", event); gst::log!(CAT, obj: self.element, "Forwarding {:?}", event);
appsrc.src_pad.push_event(event).await; appsrc.src_pad.push_event(event).await;
Ok(gst::FlowSuccess::Ok) Ok(gst::FlowSuccess::Ok)
} }
@ -242,18 +242,18 @@ impl TaskImpl for AppSrcTask {
let res = self.push_item(item).await; let res = self.push_item(item).await;
match res { match res {
Ok(_) => { Ok(_) => {
gst::log!(CAT, obj: &self.element, "Successfully pushed item"); gst::log!(CAT, obj: self.element, "Successfully pushed item");
} }
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
gst::debug!(CAT, obj: &self.element, "EOS"); gst::debug!(CAT, obj: self.element, "EOS");
let appsrc = self.element.imp(); let appsrc = self.element.imp();
appsrc.src_pad.push_event(gst::event::Eos::new()).await; appsrc.src_pad.push_event(gst::event::Eos::new()).await;
} }
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(CAT, obj: &self.element, "Flushing"); gst::debug!(CAT, obj: self.element, "Flushing");
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Got error {}", err); gst::error!(CAT, obj: self.element, "Got error {}", err);
gst::element_error!( gst::element_error!(
&self.element, &self.element,
gst::StreamError::Failed, gst::StreamError::Failed,
@ -270,13 +270,13 @@ impl TaskImpl for AppSrcTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task"); gst::log!(CAT, obj: self.element, "Stopping task");
self.flush(); self.flush();
self.need_initial_events = true; self.need_initial_events = true;
self.need_segment = true; self.need_segment = true;
gst::log!(CAT, obj: &self.element, "Task stopped"); gst::log!(CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -284,12 +284,12 @@ impl TaskImpl for AppSrcTask {
fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Starting task flush"); gst::log!(CAT, obj: self.element, "Starting task flush");
self.flush(); self.flush();
self.need_segment = true; self.need_segment = true;
gst::log!(CAT, obj: &self.element, "Task flush started"); gst::log!(CAT, obj: self.element, "Task flush started");
Ok(()) Ok(())
} }
.boxed() .boxed()

28
generic/threadshare/src/dataqueue.rs
View file

@ -127,10 +127,10 @@ impl DataQueue {
pub fn start(&self) { pub fn start(&self) {
let mut inner = self.0.lock().unwrap(); let mut inner = self.0.lock().unwrap();
if inner.state == DataQueueState::Started { if inner.state == DataQueueState::Started {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Data queue already Started"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Data queue already Started");
return; return;
} }
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Starting data queue"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Starting data queue");
inner.state = DataQueueState::Started; inner.state = DataQueueState::Started;
inner.wake(); inner.wake();
} }
@ -138,10 +138,10 @@ impl DataQueue {
pub fn stop(&self) { pub fn stop(&self) {
let mut inner = self.0.lock().unwrap(); let mut inner = self.0.lock().unwrap();
if inner.state == DataQueueState::Stopped { if inner.state == DataQueueState::Stopped {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Data queue already Stopped"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Data queue already Stopped");
return; return;
} }
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Stopping data queue"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Stopping data queue");
inner.state = DataQueueState::Stopped; inner.state = DataQueueState::Stopped;
inner.wake(); inner.wake();
} }
@ -149,7 +149,7 @@ impl DataQueue {
pub fn clear(&self) { pub fn clear(&self) {
let mut inner = self.0.lock().unwrap(); let mut inner = self.0.lock().unwrap();
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Clearing data queue"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Clearing data queue");
let src_pad = inner.src_pad.clone(); let src_pad = inner.src_pad.clone();
for item in inner.queue.drain(..) { for item in inner.queue.drain(..) {
@ -163,7 +163,7 @@ impl DataQueue {
} }
} }
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Data queue cleared"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Data queue cleared");
} }
pub fn push(&self, item: DataQueueItem) -> Result<(), DataQueueItem> { pub fn push(&self, item: DataQueueItem) -> Result<(), DataQueueItem> {
@ -172,7 +172,7 @@ impl DataQueue {
if inner.state == DataQueueState::Stopped { if inner.state == DataQueueState::Stopped {
gst::debug!( gst::debug!(
DATA_QUEUE_CAT, DATA_QUEUE_CAT,
obj: &inner.element, obj: inner.element,
"Rejecting item {:?} in state {:?}", "Rejecting item {:?} in state {:?}",
item, item,
inner.state inner.state
@ -180,7 +180,7 @@ impl DataQueue {
return Err(item); return Err(item);
} }
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Pushing item {:?}", item); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Pushing item {:?}", item);
let (count, bytes) = item.size(); let (count, bytes) = item.size();
let queue_ts = inner.queue.iter().filter_map(|i| i.timestamp()).next(); let queue_ts = inner.queue.iter().filter_map(|i| i.timestamp()).next();
@ -188,14 +188,14 @@ impl DataQueue {
if let Some(max) = inner.max_size_buffers { if let Some(max) = inner.max_size_buffers {
if max <= inner.cur_size_buffers { if max <= inner.cur_size_buffers {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Queue is full (buffers): {} <= {}", max, inner.cur_size_buffers); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Queue is full (buffers): {} <= {}", max, inner.cur_size_buffers);
return Err(item); return Err(item);
} }
} }
if let Some(max) = inner.max_size_bytes { if let Some(max) = inner.max_size_bytes {
if max <= inner.cur_size_bytes { if max <= inner.cur_size_bytes {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Queue is full (bytes): {} <= {}", max, inner.cur_size_bytes); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Queue is full (bytes): {} <= {}", max, inner.cur_size_bytes);
return Err(item); return Err(item);
} }
} }
@ -209,7 +209,7 @@ impl DataQueue {
}; };
if max <= level { if max <= level {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Queue is full (time): {} <= {}", max, level); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Queue is full (time): {} <= {}", max, level);
return Err(item); return Err(item);
} }
} }
@ -232,10 +232,10 @@ impl DataQueue {
match inner.state { match inner.state {
DataQueueState::Started => match inner.queue.pop_front() { DataQueueState::Started => match inner.queue.pop_front() {
None => { None => {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Data queue is empty"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Data queue is empty");
} }
Some(item) => { Some(item) => {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Popped item {:?}", item); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Popped item {:?}", item);
let (count, bytes) = item.size(); let (count, bytes) = item.size();
inner.cur_size_buffers -= count; inner.cur_size_buffers -= count;
@ -245,7 +245,7 @@ impl DataQueue {
} }
}, },
DataQueueState::Stopped => { DataQueueState::Stopped => {
gst::debug!(DATA_QUEUE_CAT, obj: &inner.element, "Data queue Stopped"); gst::debug!(DATA_QUEUE_CAT, obj: inner.element, "Data queue Stopped");
return None; return None;
} }
} }

20
generic/threadshare/src/jitterbuffer/imp.rs
View file

@ -1033,7 +1033,7 @@ impl TaskImpl for JitterBufferTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Starting task"); gst::log!(CAT, obj: self.element, "Starting task");
self.src_pad_handler.clear(); self.src_pad_handler.clear();
self.sink_pad_handler.clear(); self.sink_pad_handler.clear();
@ -1046,7 +1046,7 @@ impl TaskImpl for JitterBufferTask {
state.jbuf.set_delay(latency); state.jbuf.set_delay(latency);
*jb.state.lock().unwrap() = state; *jb.state.lock().unwrap() = state;
gst::log!(CAT, obj: &self.element, "Task started"); gst::log!(CAT, obj: self.element, "Task started");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -1106,9 +1106,9 @@ impl TaskImpl for JitterBufferTask {
// Got aborted, reschedule if needed // Got aborted, reschedule if needed
if let Some(delay_fut) = delay_fut { if let Some(delay_fut) = delay_fut {
gst::debug!(CAT, obj: &self.element, "Waiting"); gst::debug!(CAT, obj: self.element, "Waiting");
if let Err(Aborted) = delay_fut.await { if let Err(Aborted) = delay_fut.await {
gst::debug!(CAT, obj: &self.element, "Waiting aborted"); gst::debug!(CAT, obj: self.element, "Waiting aborted");
return Ok(()); return Ok(());
} }
} }
@ -1126,7 +1126,7 @@ impl TaskImpl for JitterBufferTask {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &self.element, obj: self.element,
"Woke up at {}, earliest_pts {}", "Woke up at {}, earliest_pts {}",
now.display(), now.display(),
state.earliest_pts.display() state.earliest_pts.display()
@ -1182,13 +1182,13 @@ impl TaskImpl for JitterBufferTask {
if let Err(err) = res { if let Err(err) = res {
match err { match err {
gst::FlowError::Eos => { gst::FlowError::Eos => {
gst::debug!(CAT, obj: &self.element, "Pushing EOS event"); gst::debug!(CAT, obj: self.element, "Pushing EOS event");
let _ = jb.src_pad.push_event(gst::event::Eos::new()).await; let _ = jb.src_pad.push_event(gst::event::Eos::new()).await;
} }
gst::FlowError::Flushing => { gst::FlowError::Flushing => {
gst::debug!(CAT, obj: &self.element, "Flushing") gst::debug!(CAT, obj: self.element, "Flushing")
} }
err => gst::error!(CAT, obj: &self.element, "Error {}", err), err => gst::error!(CAT, obj: self.element, "Error {}", err),
} }
return Err(err); return Err(err);
@ -1204,7 +1204,7 @@ impl TaskImpl for JitterBufferTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task"); gst::log!(CAT, obj: self.element, "Stopping task");
let jb = self.element.imp(); let jb = self.element.imp();
let mut jb_state = jb.state.lock().unwrap(); let mut jb_state = jb.state.lock().unwrap();
@ -1218,7 +1218,7 @@ impl TaskImpl for JitterBufferTask {
*jb_state = State::default(); *jb_state = State::default();
gst::log!(CAT, obj: &self.element, "Task stopped"); gst::log!(CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()

26
generic/threadshare/src/proxy/imp.rs
View file

@ -784,15 +784,15 @@ impl ProxySrcTask {
match item { match item {
DataQueueItem::Buffer(buffer) => { DataQueueItem::Buffer(buffer) => {
gst::log!(SRC_CAT, obj: &self.element, "Forwarding {:?}", buffer); gst::log!(SRC_CAT, obj: self.element, "Forwarding {:?}", buffer);
proxysrc.src_pad.push(buffer).await.map(drop) proxysrc.src_pad.push(buffer).await.map(drop)
} }
DataQueueItem::BufferList(list) => { DataQueueItem::BufferList(list) => {
gst::log!(SRC_CAT, obj: &self.element, "Forwarding {:?}", list); gst::log!(SRC_CAT, obj: self.element, "Forwarding {:?}", list);
proxysrc.src_pad.push_list(list).await.map(drop) proxysrc.src_pad.push_list(list).await.map(drop)
} }
DataQueueItem::Event(event) => { DataQueueItem::Event(event) => {
gst::log!(SRC_CAT, obj: &self.element, "Forwarding {:?}", event); gst::log!(SRC_CAT, obj: self.element, "Forwarding {:?}", event);
proxysrc.src_pad.push_event(event).await; proxysrc.src_pad.push_event(event).await;
Ok(()) Ok(())
} }
@ -805,7 +805,7 @@ impl TaskImpl for ProxySrcTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(SRC_CAT, obj: &self.element, "Starting task"); gst::log!(SRC_CAT, obj: self.element, "Starting task");
let proxysrc = self.element.imp(); let proxysrc = self.element.imp();
let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap(); let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap();
@ -819,7 +819,7 @@ impl TaskImpl for ProxySrcTask {
self.dataqueue.start(); self.dataqueue.start();
gst::log!(SRC_CAT, obj: &self.element, "Task started"); gst::log!(SRC_CAT, obj: self.element, "Task started");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -841,25 +841,25 @@ impl TaskImpl for ProxySrcTask {
let proxysrc = self.element.imp(); let proxysrc = self.element.imp();
match res { match res {
Ok(()) => { Ok(()) => {
gst::log!(SRC_CAT, obj: &self.element, "Successfully pushed item"); gst::log!(SRC_CAT, obj: self.element, "Successfully pushed item");
let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap(); let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap();
let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared(); let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared();
shared_ctx.last_res = Ok(gst::FlowSuccess::Ok); shared_ctx.last_res = Ok(gst::FlowSuccess::Ok);
} }
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(SRC_CAT, obj: &self.element, "Flushing"); gst::debug!(SRC_CAT, obj: self.element, "Flushing");
let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap(); let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap();
let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared(); let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared();
shared_ctx.last_res = Err(gst::FlowError::Flushing); shared_ctx.last_res = Err(gst::FlowError::Flushing);
} }
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
gst::debug!(SRC_CAT, obj: &self.element, "EOS"); gst::debug!(SRC_CAT, obj: self.element, "EOS");
let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap(); let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap();
let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared(); let mut shared_ctx = proxy_ctx.as_ref().unwrap().lock_shared();
shared_ctx.last_res = Err(gst::FlowError::Eos); shared_ctx.last_res = Err(gst::FlowError::Eos);
} }
Err(err) => { Err(err) => {
gst::error!(SRC_CAT, obj: &self.element, "Got error {}", err); gst::error!(SRC_CAT, obj: self.element, "Got error {}", err);
gst::element_error!( gst::element_error!(
&self.element, &self.element,
gst::StreamError::Failed, gst::StreamError::Failed,
@ -879,7 +879,7 @@ impl TaskImpl for ProxySrcTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(SRC_CAT, obj: &self.element, "Stopping task"); gst::log!(SRC_CAT, obj: self.element, "Stopping task");
let proxysrc = self.element.imp(); let proxysrc = self.element.imp();
let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap(); let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap();
@ -894,7 +894,7 @@ impl TaskImpl for ProxySrcTask {
pending_queue.notify_more_queue_space(); pending_queue.notify_more_queue_space();
} }
gst::log!(SRC_CAT, obj: &self.element, "Task stopped"); gst::log!(SRC_CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -902,7 +902,7 @@ impl TaskImpl for ProxySrcTask {
fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(SRC_CAT, obj: &self.element, "Starting task flush"); gst::log!(SRC_CAT, obj: self.element, "Starting task flush");
let proxysrc = self.element.imp(); let proxysrc = self.element.imp();
let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap(); let proxy_ctx = proxysrc.proxy_ctx.lock().unwrap();
@ -912,7 +912,7 @@ impl TaskImpl for ProxySrcTask {
shared_ctx.last_res = Err(gst::FlowError::Flushing); shared_ctx.last_res = Err(gst::FlowError::Flushing);
gst::log!(SRC_CAT, obj: &self.element, "Task flush started"); gst::log!(SRC_CAT, obj: self.element, "Task flush started");
Ok(()) Ok(())
} }
.boxed() .boxed()

26
generic/threadshare/src/queue/imp.rs
View file

@ -264,15 +264,15 @@ impl QueueTask {
match item { match item {
DataQueueItem::Buffer(buffer) => { DataQueueItem::Buffer(buffer) => {
gst::log!(CAT, obj: &self.element, "Forwarding {:?}", buffer); gst::log!(CAT, obj: self.element, "Forwarding {:?}", buffer);
queue.src_pad.push(buffer).await.map(drop) queue.src_pad.push(buffer).await.map(drop)
} }
DataQueueItem::BufferList(list) => { DataQueueItem::BufferList(list) => {
gst::log!(CAT, obj: &self.element, "Forwarding {:?}", list); gst::log!(CAT, obj: self.element, "Forwarding {:?}", list);
queue.src_pad.push_list(list).await.map(drop) queue.src_pad.push_list(list).await.map(drop)
} }
DataQueueItem::Event(event) => { DataQueueItem::Event(event) => {
gst::log!(CAT, obj: &self.element, "Forwarding {:?}", event); gst::log!(CAT, obj: self.element, "Forwarding {:?}", event);
queue.src_pad.push_event(event).await; queue.src_pad.push_event(event).await;
Ok(()) Ok(())
} }
@ -285,7 +285,7 @@ impl TaskImpl for QueueTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Starting task"); gst::log!(CAT, obj: self.element, "Starting task");
let queue = self.element.imp(); let queue = self.element.imp();
let mut last_res = queue.last_res.lock().unwrap(); let mut last_res = queue.last_res.lock().unwrap();
@ -294,7 +294,7 @@ impl TaskImpl for QueueTask {
*last_res = Ok(gst::FlowSuccess::Ok); *last_res = Ok(gst::FlowSuccess::Ok);
gst::log!(CAT, obj: &self.element, "Task started"); gst::log!(CAT, obj: self.element, "Task started");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -316,20 +316,20 @@ impl TaskImpl for QueueTask {
let queue = self.element.imp(); let queue = self.element.imp();
match res { match res {
Ok(()) => { Ok(()) => {
gst::log!(CAT, obj: &self.element, "Successfully pushed item"); gst::log!(CAT, obj: self.element, "Successfully pushed item");
*queue.last_res.lock().unwrap() = Ok(gst::FlowSuccess::Ok); *queue.last_res.lock().unwrap() = Ok(gst::FlowSuccess::Ok);
} }
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(CAT, obj: &self.element, "Flushing"); gst::debug!(CAT, obj: self.element, "Flushing");
*queue.last_res.lock().unwrap() = Err(gst::FlowError::Flushing); *queue.last_res.lock().unwrap() = Err(gst::FlowError::Flushing);
} }
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
gst::debug!(CAT, obj: &self.element, "EOS"); gst::debug!(CAT, obj: self.element, "EOS");
*queue.last_res.lock().unwrap() = Err(gst::FlowError::Eos); *queue.last_res.lock().unwrap() = Err(gst::FlowError::Eos);
queue.src_pad.push_event(gst::event::Eos::new()).await; queue.src_pad.push_event(gst::event::Eos::new()).await;
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Got error {}", err); gst::error!(CAT, obj: self.element, "Got error {}", err);
gst::element_error!( gst::element_error!(
&self.element, &self.element,
gst::StreamError::Failed, gst::StreamError::Failed,
@ -347,7 +347,7 @@ impl TaskImpl for QueueTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task"); gst::log!(CAT, obj: self.element, "Stopping task");
let queue = self.element.imp(); let queue = self.element.imp();
let mut last_res = queue.last_res.lock().unwrap(); let mut last_res = queue.last_res.lock().unwrap();
@ -361,7 +361,7 @@ impl TaskImpl for QueueTask {
*last_res = Err(gst::FlowError::Flushing); *last_res = Err(gst::FlowError::Flushing);
gst::log!(CAT, obj: &self.element, "Task stopped"); gst::log!(CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -369,7 +369,7 @@ impl TaskImpl for QueueTask {
fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Starting task flush"); gst::log!(CAT, obj: self.element, "Starting task flush");
let queue = self.element.imp(); let queue = self.element.imp();
let mut last_res = queue.last_res.lock().unwrap(); let mut last_res = queue.last_res.lock().unwrap();
@ -382,7 +382,7 @@ impl TaskImpl for QueueTask {
*last_res = Err(gst::FlowError::Flushing); *last_res = Err(gst::FlowError::Flushing);
gst::log!(CAT, obj: &self.element, "Task flush started"); gst::log!(CAT, obj: self.element, "Task flush started");
Ok(()) Ok(())
} }
.boxed() .boxed()

8
generic/threadshare/src/runtime/pad.rs
View file

@ -241,7 +241,7 @@ impl PadSrcInner {
err err
})?; })?;
gst::log!(RUNTIME_CAT, obj: &self.gst_pad, "Processing any pending sub tasks"); gst::log!(RUNTIME_CAT, obj: self.gst_pad, "Processing any pending sub tasks");
Context::drain_sub_tasks().await?; Context::drain_sub_tasks().await?;
Ok(success) Ok(success)
@ -260,18 +260,18 @@ impl PadSrcInner {
err err
})?; })?;
gst::log!(RUNTIME_CAT, obj: &self.gst_pad, "Processing any pending sub tasks"); gst::log!(RUNTIME_CAT, obj: self.gst_pad, "Processing any pending sub tasks");
Context::drain_sub_tasks().await?; Context::drain_sub_tasks().await?;
Ok(success) Ok(success)
} }
pub async fn push_event(&self, event: gst::Event) -> bool { pub async fn push_event(&self, event: gst::Event) -> bool {
gst::log!(RUNTIME_CAT, obj: &self.gst_pad, "Pushing {:?}", event); gst::log!(RUNTIME_CAT, obj: self.gst_pad, "Pushing {:?}", event);
let was_handled = self.gst_pad().push_event(event); let was_handled = self.gst_pad().push_event(event);
gst::log!(RUNTIME_CAT, obj: &self.gst_pad, "Processing any pending sub tasks"); gst::log!(RUNTIME_CAT, obj: self.gst_pad, "Processing any pending sub tasks");
if Context::drain_sub_tasks().await.is_err() { if Context::drain_sub_tasks().await.is_err() {
return false; return false;
} }

14
generic/threadshare/src/socket.rs
View file

@ -76,7 +76,7 @@ impl<T: SocketRead> Socket<T> {
buffer_pool.set_active(true).map_err(|err| { buffer_pool.set_active(true).map_err(|err| {
gst::error!( gst::error!(
SOCKET_CAT, SOCKET_CAT,
obj: &element, obj: element,
"Failed to prepare socket: {}", "Failed to prepare socket: {}",
err err
); );
@ -124,7 +124,7 @@ impl<T: SocketRead> Socket<T> {
pub async fn try_next( pub async fn try_next(
&mut self, &mut self,
) -> Result<(gst::Buffer, Option<std::net::SocketAddr>), SocketError> { ) -> Result<(gst::Buffer, Option<std::net::SocketAddr>), SocketError> {
gst::log!(SOCKET_CAT, obj: &self.element, "Trying to read data"); gst::log!(SOCKET_CAT, obj: self.element, "Trying to read data");
if self.mapped_buffer.is_none() { if self.mapped_buffer.is_none() {
match self.buffer_pool.acquire_buffer(None) { match self.buffer_pool.acquire_buffer(None) {
@ -132,7 +132,7 @@ impl<T: SocketRead> Socket<T> {
self.mapped_buffer = Some(buffer.into_mapped_buffer_writable().unwrap()); self.mapped_buffer = Some(buffer.into_mapped_buffer_writable().unwrap());
} }
Err(err) => { Err(err) => {
gst::debug!(SOCKET_CAT, obj: &self.element, "Failed to acquire buffer {:?}", err); gst::debug!(SOCKET_CAT, obj: self.element, "Failed to acquire buffer {:?}", err);
return Err(SocketError::Gst(err)); return Err(SocketError::Gst(err));
} }
} }
@ -151,7 +151,7 @@ impl<T: SocketRead> Socket<T> {
// so as to display another message // so as to display another message
gst::debug!( gst::debug!(
SOCKET_CAT, SOCKET_CAT,
obj: &self.element, obj: self.element,
"Read {} bytes at {} (clock {})", "Read {} bytes at {} (clock {})",
len, len,
running_time.display(), running_time.display(),
@ -159,7 +159,7 @@ impl<T: SocketRead> Socket<T> {
); );
running_time running_time
} else { } else {
gst::debug!(SOCKET_CAT, obj: &self.element, "Read {} bytes", len); gst::debug!(SOCKET_CAT, obj: self.element, "Read {} bytes", len);
gst::ClockTime::NONE gst::ClockTime::NONE
}; };
@ -175,7 +175,7 @@ impl<T: SocketRead> Socket<T> {
Ok((buffer, saddr)) Ok((buffer, saddr))
} }
Err(err) => { Err(err) => {
gst::debug!(SOCKET_CAT, obj: &self.element, "Read error {:?}", err); gst::debug!(SOCKET_CAT, obj: self.element, "Read error {:?}", err);
Err(SocketError::Io(err)) Err(SocketError::Io(err))
} }
@ -186,7 +186,7 @@ impl<T: SocketRead> Socket<T> {
impl<T: SocketRead> Drop for Socket<T> { impl<T: SocketRead> Drop for Socket<T> {
fn drop(&mut self) { fn drop(&mut self) {
if let Err(err) = self.buffer_pool.set_active(false) { if let Err(err) = self.buffer_pool.set_active(false) {
gst::error!(SOCKET_CAT, obj: &self.element, "Failed to unprepare socket: {}", err); gst::error!(SOCKET_CAT, obj: self.element, "Failed to unprepare socket: {}", err);
} }
} }
} }

26
generic/threadshare/src/tcpclientsrc/imp.rs
View file

@ -184,12 +184,12 @@ impl TcpClientSrcTask {
&mut self, &mut self,
buffer: gst::Buffer, buffer: gst::Buffer,
) -> Result<gst::FlowSuccess, gst::FlowError> { ) -> Result<gst::FlowSuccess, gst::FlowError> {
gst::log!(CAT, obj: &self.element, "Handling {:?}", buffer); gst::log!(CAT, obj: self.element, "Handling {:?}", buffer);
let tcpclientsrc = self.element.imp(); let tcpclientsrc = self.element.imp();
if self.need_initial_events { if self.need_initial_events {
gst::debug!(CAT, obj: &self.element, "Pushing initial events"); gst::debug!(CAT, obj: self.element, "Pushing initial events");
let stream_id = format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>()); let stream_id = format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>());
let stream_start_evt = gst::event::StreamStart::builder(&stream_id) let stream_start_evt = gst::event::StreamStart::builder(&stream_id)
@ -228,20 +228,20 @@ impl TcpClientSrcTask {
let res = tcpclientsrc.src_pad.push(buffer).await; let res = tcpclientsrc.src_pad.push(buffer).await;
match res { match res {
Ok(_) => { Ok(_) => {
gst::log!(CAT, obj: &self.element, "Successfully pushed buffer"); gst::log!(CAT, obj: self.element, "Successfully pushed buffer");
} }
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(CAT, obj: &self.element, "Flushing"); gst::debug!(CAT, obj: self.element, "Flushing");
} }
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
gst::debug!(CAT, obj: &self.element, "EOS"); gst::debug!(CAT, obj: self.element, "EOS");
tcpclientsrc tcpclientsrc
.src_pad .src_pad
.push_event(gst::event::Eos::new()) .push_event(gst::event::Eos::new())
.await; .await;
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Got error {}", err); gst::error!(CAT, obj: self.element, "Got error {}", err);
gst::element_error!( gst::element_error!(
self.element, self.element,
gst::StreamError::Failed, gst::StreamError::Failed,
@ -260,7 +260,7 @@ impl TaskImpl for TcpClientSrcTask {
fn prepare(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn prepare(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Preparing task connecting to {:?}", self.saddr); gst::log!(CAT, obj: self.element, "Preparing task connecting to {:?}", self.saddr);
let socket = Async::<TcpStream>::connect(self.saddr) let socket = Async::<TcpStream>::connect(self.saddr)
.await .await
@ -285,7 +285,7 @@ impl TaskImpl for TcpClientSrcTask {
})?, })?,
); );
gst::log!(CAT, obj: &self.element, "Task prepared"); gst::log!(CAT, obj: self.element, "Task prepared");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -320,7 +320,7 @@ impl TaskImpl for TcpClientSrcTask {
.await .await
.map(|(buffer, _saddr)| buffer) .map(|(buffer, _saddr)| buffer)
.map_err(|err| { .map_err(|err| {
gst::error!(CAT, obj: &self.element, "Got error {:?}", err); gst::error!(CAT, obj: self.element, "Got error {:?}", err);
match err { match err {
SocketError::Gst(err) => { SocketError::Gst(err) => {
gst::element_error!( gst::element_error!(
@ -351,9 +351,9 @@ impl TaskImpl for TcpClientSrcTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task"); gst::log!(CAT, obj: self.element, "Stopping task");
self.need_initial_events = true; self.need_initial_events = true;
gst::log!(CAT, obj: &self.element, "Task stopped"); gst::log!(CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -361,9 +361,9 @@ impl TaskImpl for TcpClientSrcTask {
fn flush_stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task flush"); gst::log!(CAT, obj: self.element, "Stopping task flush");
self.need_initial_events = true; self.need_initial_events = true;
gst::log!(CAT, obj: &self.element, "Task flush stopped"); gst::log!(CAT, obj: self.element, "Task flush stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()

52
generic/threadshare/src/udpsink/imp.rs
View file

@ -141,7 +141,7 @@ impl PadSinkHandler for UdpSinkPadHandler {
let sender = elem.imp().clone_item_sender(); let sender = elem.imp().clone_item_sender();
async move { async move {
if sender.send_async(TaskItem::Buffer(buffer)).await.is_err() { if sender.send_async(TaskItem::Buffer(buffer)).await.is_err() {
gst::debug!(CAT, obj: &elem, "Flushing"); gst::debug!(CAT, obj: elem, "Flushing");
return Err(gst::FlowError::Flushing); return Err(gst::FlowError::Flushing);
} }
@ -160,7 +160,7 @@ impl PadSinkHandler for UdpSinkPadHandler {
async move { async move {
for buffer in list.iter_owned() { for buffer in list.iter_owned() {
if sender.send_async(TaskItem::Buffer(buffer)).await.is_err() { if sender.send_async(TaskItem::Buffer(buffer)).await.is_err() {
gst::debug!(CAT, obj: &elem, "Flushing"); gst::debug!(CAT, obj: elem, "Flushing");
return Err(gst::FlowError::Flushing); return Err(gst::FlowError::Flushing);
} }
} }
@ -182,7 +182,7 @@ impl PadSinkHandler for UdpSinkPadHandler {
let imp = elem.imp(); let imp = elem.imp();
return imp.task.flush_stop().await_maybe_on_context().is_ok(); return imp.task.flush_stop().await_maybe_on_context().is_ok();
} else if sender.send_async(TaskItem::Event(event)).await.is_err() { } else if sender.send_async(TaskItem::Event(event)).await.is_err() {
gst::debug!(CAT, obj: &elem, "Flushing"); gst::debug!(CAT, obj: elem, "Flushing");
} }
true true
@ -306,7 +306,7 @@ impl UdpSinkTask {
}; };
let saddr = SocketAddr::new(bind_addr, bind_port as u16); let saddr = SocketAddr::new(bind_addr, bind_port as u16);
gst::debug!(CAT, obj: &self.element, "Binding to {:?}", saddr); gst::debug!(CAT, obj: self.element, "Binding to {:?}", saddr);
let socket = match family { let socket = match family {
SocketFamily::Ipv4 => socket2::Socket::new( SocketFamily::Ipv4 => socket2::Socket::new(
@ -326,7 +326,7 @@ impl UdpSinkTask {
Err(err) => { Err(err) => {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &self.element, obj: self.element,
"Failed to create {} socket: {}", "Failed to create {} socket: {}",
match family { match family {
SocketFamily::Ipv4 => "IPv4", SocketFamily::Ipv4 => "IPv4",
@ -378,7 +378,7 @@ impl UdpSinkTask {
fn add_client(&mut self, addr: SocketAddr) { fn add_client(&mut self, addr: SocketAddr) {
if self.clients.contains(&addr) { if self.clients.contains(&addr) {
gst::warning!(CAT, obj: &self.element, "Not adding client {:?} again", &addr); gst::warning!(CAT, obj: self.element, "Not adding client {:?} again", &addr);
return; return;
} }
@ -386,11 +386,11 @@ impl UdpSinkTask {
let mut settings = udpsink.settings.lock().unwrap(); let mut settings = udpsink.settings.lock().unwrap();
match self.configure_client(&settings, &addr) { match self.configure_client(&settings, &addr) {
Ok(()) => { Ok(()) => {
gst::info!(CAT, obj: &self.element, "Added client {:?}", addr); gst::info!(CAT, obj: self.element, "Added client {:?}", addr);
self.clients.insert(addr); self.clients.insert(addr);
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Failed to add client {:?}: {}", addr, err); gst::error!(CAT, obj: self.element, "Failed to add client {:?}: {}", addr, err);
settings.clients = self.clients.clone(); settings.clients = self.clients.clone();
self.element.post_error_message(err); self.element.post_error_message(err);
} }
@ -399,7 +399,7 @@ impl UdpSinkTask {
fn remove_client(&mut self, addr: &SocketAddr) { fn remove_client(&mut self, addr: &SocketAddr) {
if self.clients.take(addr).is_none() { if self.clients.take(addr).is_none() {
gst::warning!(CAT, obj: &self.element, "Not removing unknown client {:?}", &addr); gst::warning!(CAT, obj: self.element, "Not removing unknown client {:?}", &addr);
return; return;
} }
@ -407,10 +407,10 @@ impl UdpSinkTask {
let mut settings = udpsink.settings.lock().unwrap(); let mut settings = udpsink.settings.lock().unwrap();
match self.unconfigure_client(&settings, addr) { match self.unconfigure_client(&settings, addr) {
Ok(()) => { Ok(()) => {
gst::info!(CAT, obj: &self.element, "Removed client {:?}", addr); gst::info!(CAT, obj: self.element, "Removed client {:?}", addr);
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Failed to remove client {:?}: {}", addr, err); gst::error!(CAT, obj: self.element, "Failed to remove client {:?}: {}", addr, err);
settings.clients = self.clients.clone(); settings.clients = self.clients.clone();
self.element.post_error_message(err); self.element.post_error_message(err);
} }
@ -419,9 +419,9 @@ impl UdpSinkTask {
fn replace_with_clients(&mut self, mut clients_to_add: BTreeSet<SocketAddr>) { fn replace_with_clients(&mut self, mut clients_to_add: BTreeSet<SocketAddr>) {
if clients_to_add.is_empty() { if clients_to_add.is_empty() {
gst::info!(CAT, obj: &self.element, "Clearing clients"); gst::info!(CAT, obj: self.element, "Clearing clients");
} else { } else {
gst::info!(CAT, obj: &self.element, "Replacing clients"); gst::info!(CAT, obj: self.element, "Replacing clients");
} }
let old_clients = std::mem::take(&mut self.clients); let old_clients = std::mem::take(&mut self.clients);
@ -435,19 +435,19 @@ impl UdpSinkTask {
// client is already configured // client is already configured
self.clients.insert(*addr); self.clients.insert(*addr);
} else if let Err(err) = self.unconfigure_client(&settings, addr) { } else if let Err(err) = self.unconfigure_client(&settings, addr) {
gst::error!(CAT, obj: &self.element, "Failed to remove client {:?}: {}", addr, err); gst::error!(CAT, obj: self.element, "Failed to remove client {:?}: {}", addr, err);
res = Err(err); res = Err(err);
} else { } else {
gst::info!(CAT, obj: &self.element, "Removed client {:?}", addr); gst::info!(CAT, obj: self.element, "Removed client {:?}", addr);
} }
} }
for addr in clients_to_add.into_iter() { for addr in clients_to_add.into_iter() {
if let Err(err) = self.configure_client(&settings, &addr) { if let Err(err) = self.configure_client(&settings, &addr) {
gst::error!(CAT, obj: &self.element, "Failed to add client {:?}: {}", addr, err); gst::error!(CAT, obj: self.element, "Failed to add client {:?}: {}", addr, err);
res = Err(err); res = Err(err);
} else { } else {
gst::info!(CAT, obj: &self.element, "Added client {:?}", addr); gst::info!(CAT, obj: self.element, "Added client {:?}", addr);
self.clients.insert(addr); self.clients.insert(addr);
} }
} }
@ -627,7 +627,7 @@ impl UdpSinkTask {
}; };
if let Some(socket) = socket.as_mut() { if let Some(socket) = socket.as_mut() {
gst::log!(CAT, obj: &self.element, "Sending to {:?}", &client); gst::log!(CAT, obj: self.element, "Sending to {:?}", &client);
socket.send_to(&data, *client).await.map_err(|err| { socket.send_to(&data, *client).await.map_err(|err| {
element_error!( element_error!(
self.element, self.element,
@ -650,7 +650,7 @@ impl UdpSinkTask {
gst::log!( gst::log!(
CAT, CAT,
obj: &self.element, obj: self.element,
"Sent buffer {:?} to all clients", "Sent buffer {:?} to all clients",
&buffer &buffer
); );
@ -663,7 +663,7 @@ impl UdpSinkTask {
let now = self.element.current_running_time(); let now = self.element.current_running_time();
if let Ok(Some(delay)) = running_time.opt_checked_sub(now) { if let Ok(Some(delay)) = running_time.opt_checked_sub(now) {
gst::trace!(CAT, obj: &self.element, "sync: waiting {}", delay); gst::trace!(CAT, obj: self.element, "sync: waiting {}", delay);
runtime::timer::delay_for(delay.into()).await; runtime::timer::delay_for(delay.into()).await;
} }
} }
@ -674,7 +674,7 @@ impl TaskImpl for UdpSinkTask {
fn prepare(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn prepare(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::info!(CAT, obj: &self.element, "Preparing Task"); gst::info!(CAT, obj: self.element, "Preparing Task");
assert!(self.clients.is_empty()); assert!(self.clients.is_empty());
let clients = { let clients = {
let udpsink = self.element.imp(); let udpsink = self.element.imp();
@ -695,7 +695,7 @@ impl TaskImpl for UdpSinkTask {
fn unprepare(&mut self) -> BoxFuture<'_, ()> { fn unprepare(&mut self) -> BoxFuture<'_, ()> {
async move { async move {
gst::info!(CAT, obj: &self.element, "Unpreparing Task"); gst::info!(CAT, obj: self.element, "Unpreparing Task");
let udpsink = self.element.imp(); let udpsink = self.element.imp();
let settings = udpsink.settings.lock().unwrap(); let settings = udpsink.settings.lock().unwrap();
@ -709,7 +709,7 @@ impl TaskImpl for UdpSinkTask {
fn try_next(&mut self) -> BoxFuture<'_, Result<TaskItem, gst::FlowError>> { fn try_next(&mut self) -> BoxFuture<'_, Result<TaskItem, gst::FlowError>> {
async move { async move {
loop { loop {
gst::info!(CAT, obj: &self.element, "Awaiting next item or command"); gst::info!(CAT, obj: self.element, "Awaiting next item or command");
futures::select_biased! { futures::select_biased! {
cmd = self.cmd_receiver.recv_async() => { cmd = self.cmd_receiver.recv_async() => {
self.process_command(cmd.unwrap()); self.process_command(cmd.unwrap());
@ -751,7 +751,7 @@ impl TaskImpl for UdpSinkTask {
fn handle_item(&mut self, item: TaskItem) -> BoxFuture<'_, Result<(), gst::FlowError>> { fn handle_item(&mut self, item: TaskItem) -> BoxFuture<'_, Result<(), gst::FlowError>> {
async move { async move {
gst::info!(CAT, obj: &self.element, "Handling {:?}", item); gst::info!(CAT, obj: self.element, "Handling {:?}", item);
match item { match item {
TaskItem::Buffer(buffer) => self.render(buffer).await.map_err(|err| { TaskItem::Buffer(buffer) => self.render(buffer).await.map_err(|err| {
@ -785,7 +785,7 @@ impl TaskImpl for UdpSinkTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async { async {
gst::info!(CAT, obj: &self.element, "Stopping Task"); gst::info!(CAT, obj: self.element, "Stopping Task");
self.flush().await; self.flush().await;
Ok(()) Ok(())
} }
@ -794,7 +794,7 @@ impl TaskImpl for UdpSinkTask {
fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async { async {
gst::info!(CAT, obj: &self.element, "Starting Task Flush"); gst::info!(CAT, obj: self.element, "Starting Task Flush");
self.flush().await; self.flush().await;
Ok(()) Ok(())
} }

34
generic/threadshare/src/udpsrc/imp.rs
View file

@ -204,7 +204,7 @@ impl TaskImpl for UdpSrcTask {
let udpsrc = self.element.imp(); let udpsrc = self.element.imp();
let mut settings = udpsrc.settings.lock().unwrap(); let mut settings = udpsrc.settings.lock().unwrap();
gst::debug!(CAT, obj: &self.element, "Preparing Task"); gst::debug!(CAT, obj: self.element, "Preparing Task");
self.retrieve_sender_address = settings.retrieve_sender_address; self.retrieve_sender_address = settings.retrieve_sender_address;
@ -261,7 +261,7 @@ impl TaskImpl for UdpSrcTask {
let saddr = SocketAddr::new(bind_addr, port as u16); let saddr = SocketAddr::new(bind_addr, port as u16);
gst::debug!( gst::debug!(
CAT, CAT,
obj: &self.element, obj: self.element,
"Binding to {:?} for multicast group {:?}", "Binding to {:?} for multicast group {:?}",
saddr, saddr,
addr addr
@ -270,7 +270,7 @@ impl TaskImpl for UdpSrcTask {
saddr saddr
} else { } else {
let saddr = SocketAddr::new(addr, port as u16); let saddr = SocketAddr::new(addr, port as u16);
gst::debug!(CAT, obj: &self.element, "Binding to {:?}", saddr); gst::debug!(CAT, obj: self.element, "Binding to {:?}", saddr);
saddr saddr
}; };
@ -398,7 +398,7 @@ impl TaskImpl for UdpSrcTask {
fn unprepare(&mut self) -> BoxFuture<'_, ()> { fn unprepare(&mut self) -> BoxFuture<'_, ()> {
async move { async move {
gst::debug!(CAT, obj: &self.element, "Unpreparing Task"); gst::debug!(CAT, obj: self.element, "Unpreparing Task");
let udpsrc = self.element.imp(); let udpsrc = self.element.imp();
udpsrc.settings.lock().unwrap().used_socket = None; udpsrc.settings.lock().unwrap().used_socket = None;
self.element.notify("used-socket"); self.element.notify("used-socket");
@ -408,12 +408,12 @@ impl TaskImpl for UdpSrcTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Starting task"); gst::log!(CAT, obj: self.element, "Starting task");
self.socket self.socket
.as_mut() .as_mut()
.unwrap() .unwrap()
.set_clock(self.element.clock(), self.element.base_time()); .set_clock(self.element.clock(), self.element.base_time());
gst::log!(CAT, obj: &self.element, "Task started"); gst::log!(CAT, obj: self.element, "Task started");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -438,7 +438,7 @@ impl TaskImpl for UdpSrcTask {
buffer buffer
}) })
.map_err(|err| { .map_err(|err| {
gst::error!(CAT, obj: &self.element, "Got error {:?}", err); gst::error!(CAT, obj: self.element, "Got error {:?}", err);
match err { match err {
SocketError::Gst(err) => { SocketError::Gst(err) => {
gst::element_error!( gst::element_error!(
@ -465,11 +465,11 @@ impl TaskImpl for UdpSrcTask {
fn handle_item(&mut self, buffer: gst::Buffer) -> BoxFuture<'_, Result<(), gst::FlowError>> { fn handle_item(&mut self, buffer: gst::Buffer) -> BoxFuture<'_, Result<(), gst::FlowError>> {
async { async {
gst::log!(CAT, obj: &self.element, "Handling {:?}", buffer); gst::log!(CAT, obj: self.element, "Handling {:?}", buffer);
let udpsrc = self.element.imp(); let udpsrc = self.element.imp();
if self.need_initial_events { if self.need_initial_events {
gst::debug!(CAT, obj: &self.element, "Pushing initial events"); gst::debug!(CAT, obj: self.element, "Pushing initial events");
let stream_id = let stream_id =
format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>()); format!("{:08x}{:08x}", rand::random::<u32>(), rand::random::<u32>());
@ -500,14 +500,14 @@ impl TaskImpl for UdpSrcTask {
let res = udpsrc.src_pad.push(buffer).await.map(drop); let res = udpsrc.src_pad.push(buffer).await.map(drop);
match res { match res {
Ok(_) => gst::log!(CAT, obj: &self.element, "Successfully pushed buffer"), Ok(_) => gst::log!(CAT, obj: self.element, "Successfully pushed buffer"),
Err(gst::FlowError::Flushing) => gst::debug!(CAT, obj: &self.element, "Flushing"), Err(gst::FlowError::Flushing) => gst::debug!(CAT, obj: self.element, "Flushing"),
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
gst::debug!(CAT, obj: &self.element, "EOS"); gst::debug!(CAT, obj: self.element, "EOS");
udpsrc.src_pad.push_event(gst::event::Eos::new()).await; udpsrc.src_pad.push_event(gst::event::Eos::new()).await;
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &self.element, "Got error {}", err); gst::error!(CAT, obj: self.element, "Got error {}", err);
gst::element_error!( gst::element_error!(
self.element, self.element,
gst::StreamError::Failed, gst::StreamError::Failed,
@ -524,10 +524,10 @@ impl TaskImpl for UdpSrcTask {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task"); gst::log!(CAT, obj: self.element, "Stopping task");
self.need_initial_events = true; self.need_initial_events = true;
self.need_segment = true; self.need_segment = true;
gst::log!(CAT, obj: &self.element, "Task stopped"); gst::log!(CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -535,9 +535,9 @@ impl TaskImpl for UdpSrcTask {
fn flush_stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(CAT, obj: &self.element, "Stopping task flush"); gst::log!(CAT, obj: self.element, "Stopping task flush");
self.need_segment = true; self.need_segment = true;
gst::log!(CAT, obj: &self.element, "Stopped task flush"); gst::log!(CAT, obj: self.element, "Stopped task flush");
Ok(()) Ok(())
} }
.boxed() .boxed()

16
generic/threadshare/tests/pad.rs
View file

@ -129,7 +129,7 @@ mod imp_src {
while let Ok(Some(_item)) = self.receiver.try_next() {} while let Ok(Some(_item)) = self.receiver.try_next() {}
} }
async fn push_item(&self, item: Item) -> Result<gst::FlowSuccess, gst::FlowError> { async fn push_item(&self, item: Item) -> Result<gst::FlowSuccess, gst::FlowError> {
gst::debug!(SRC_CAT, obj: &self.element, "Handling {:?}", item); gst::debug!(SRC_CAT, obj: self.element, "Handling {:?}", item);
let elementsrctest = self.element.imp(); let elementsrctest = self.element.imp();
match item { match item {
@ -150,7 +150,7 @@ mod imp_src {
fn try_next(&mut self) -> BoxFuture<'_, Result<Item, gst::FlowError>> { fn try_next(&mut self) -> BoxFuture<'_, Result<Item, gst::FlowError>> {
async move { async move {
self.receiver.next().await.ok_or_else(|| { self.receiver.next().await.ok_or_else(|| {
gst::log!(SRC_CAT, obj: &self.element, "SrcPad channel aborted"); gst::log!(SRC_CAT, obj: self.element, "SrcPad channel aborted");
gst::FlowError::Eos gst::FlowError::Eos
}) })
} }
@ -161,9 +161,9 @@ mod imp_src {
async move { async move {
let res = self.push_item(item).await.map(drop); let res = self.push_item(item).await.map(drop);
match res { match res {
Ok(_) => gst::log!(SRC_CAT, obj: &self.element, "Successfully pushed item"), Ok(_) => gst::log!(SRC_CAT, obj: self.element, "Successfully pushed item"),
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(SRC_CAT, obj: &self.element, "Flushing") gst::debug!(SRC_CAT, obj: self.element, "Flushing")
} }
Err(err) => panic!("Got error {}", err), Err(err) => panic!("Got error {}", err),
} }
@ -175,9 +175,9 @@ mod imp_src {
fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn stop(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(SRC_CAT, obj: &self.element, "Stopping task"); gst::log!(SRC_CAT, obj: self.element, "Stopping task");
self.flush(); self.flush();
gst::log!(SRC_CAT, obj: &self.element, "Task stopped"); gst::log!(SRC_CAT, obj: self.element, "Task stopped");
Ok(()) Ok(())
} }
.boxed() .boxed()
@ -185,9 +185,9 @@ mod imp_src {
fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> { fn flush_start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move { async move {
gst::log!(SRC_CAT, obj: &self.element, "Starting task flush"); gst::log!(SRC_CAT, obj: self.element, "Starting task flush");
self.flush(); self.flush();
gst::log!(SRC_CAT, obj: &self.element, "Task flush started"); gst::log!(SRC_CAT, obj: self.element, "Task flush started");
Ok(()) Ok(())
} }
.boxed() .boxed()

124
mux/fmp4/src/fmp4mux/imp.rs
View file

@ -193,17 +193,17 @@ impl FMP4Mux {
Some(buffer) => buffer, Some(buffer) => buffer,
None => { None => {
if stream.sinkpad.is_eos() { if stream.sinkpad.is_eos() {
gst::trace!(CAT, obj: &stream.sinkpad, "Stream is EOS"); gst::trace!(CAT, obj: stream.sinkpad, "Stream is EOS");
} else { } else {
all_have_data_or_eos = false; all_have_data_or_eos = false;
gst::trace!(CAT, obj: &stream.sinkpad, "Stream has no buffer"); gst::trace!(CAT, obj: stream.sinkpad, "Stream has no buffer");
} }
continue; continue;
} }
}; };
if stream.fragment_filled { if stream.fragment_filled {
gst::trace!(CAT, obj: &stream.sinkpad, "Stream has current fragment filled"); gst::trace!(CAT, obj: stream.sinkpad, "Stream has current fragment filled");
continue; continue;
} }
@ -216,7 +216,7 @@ impl FMP4Mux {
{ {
Some(segment) => segment, Some(segment) => segment,
None => { None => {
gst::error!(CAT, obj: &stream.sinkpad, "Got buffer before segment"); gst::error!(CAT, obj: stream.sinkpad, "Got buffer before segment");
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
} }
}; };
@ -224,7 +224,7 @@ impl FMP4Mux {
// If the stream has no valid running time, assume it's before everything else. // If the stream has no valid running time, assume it's before everything else.
let running_time = match segment.to_running_time(buffer.dts_or_pts()) { let running_time = match segment.to_running_time(buffer.dts_or_pts()) {
None => { None => {
gst::trace!(CAT, obj: &stream.sinkpad, "Stream has no valid running time"); gst::trace!(CAT, obj: stream.sinkpad, "Stream has no valid running time");
if earliest_stream if earliest_stream
.as_ref() .as_ref()
.map_or(true, |(_, _, earliest_running_time)| { .map_or(true, |(_, _, earliest_running_time)| {
@ -238,7 +238,7 @@ impl FMP4Mux {
Some(running_time) => running_time, Some(running_time) => running_time,
}; };
gst::trace!(CAT, obj: &stream.sinkpad, "Stream has running time {} queued", running_time); gst::trace!(CAT, obj: stream.sinkpad, "Stream has running time {} queued", running_time);
if earliest_stream if earliest_stream
.as_ref() .as_ref()
@ -284,22 +284,22 @@ impl FMP4Mux {
assert!(!stream.fragment_filled); assert!(!stream.fragment_filled);
gst::trace!(CAT, obj: &stream.sinkpad, "Handling buffer {:?}", buffer); gst::trace!(CAT, obj: stream.sinkpad, "Handling buffer {:?}", buffer);
let intra_only = stream.intra_only; let intra_only = stream.intra_only;
if !intra_only && buffer.dts().is_none() { if !intra_only && buffer.dts().is_none() {
gst::error!(CAT, obj: &stream.sinkpad, "Require DTS for video streams"); gst::error!(CAT, obj: stream.sinkpad, "Require DTS for video streams");
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
} }
if intra_only && buffer.flags().contains(gst::BufferFlags::DELTA_UNIT) { if intra_only && buffer.flags().contains(gst::BufferFlags::DELTA_UNIT) {
gst::error!(CAT, obj: &stream.sinkpad, "Intra-only stream with delta units"); gst::error!(CAT, obj: stream.sinkpad, "Intra-only stream with delta units");
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
} }
let pts_position = buffer.pts().ok_or_else(|| { let pts_position = buffer.pts().ok_or_else(|| {
gst::error!(CAT, obj: &stream.sinkpad, "Require timestamped buffers"); gst::error!(CAT, obj: stream.sinkpad, "Require timestamped buffers");
gst::FlowError::Error gst::FlowError::Error
})?; })?;
let duration = buffer.duration(); let duration = buffer.duration();
@ -308,22 +308,22 @@ impl FMP4Mux {
let mut pts = segment let mut pts = segment
.to_running_time_full(pts_position) .to_running_time_full(pts_position)
.ok_or_else(|| { .ok_or_else(|| {
gst::error!(CAT, obj: &stream.sinkpad, "Couldn't convert PTS to running time"); gst::error!(CAT, obj: stream.sinkpad, "Couldn't convert PTS to running time");
gst::FlowError::Error gst::FlowError::Error
})? })?
.positive_or_else(|_| { .positive_or_else(|_| {
gst::error!(CAT, obj: &stream.sinkpad, "Negative PTSs are not supported"); gst::error!(CAT, obj: stream.sinkpad, "Negative PTSs are not supported");
gst::FlowError::Error gst::FlowError::Error
})?; })?;
let mut end_pts = segment let mut end_pts = segment
.to_running_time_full(end_pts_position) .to_running_time_full(end_pts_position)
.ok_or_else(|| { .ok_or_else(|| {
gst::error!(CAT, obj: &stream.sinkpad, "Couldn't convert end PTS to running time"); gst::error!(CAT, obj: stream.sinkpad, "Couldn't convert end PTS to running time");
gst::FlowError::Error gst::FlowError::Error
})? })?
.positive_or_else(|_| { .positive_or_else(|_| {
gst::error!(CAT, obj: &stream.sinkpad, "Negative PTSs are not supported"); gst::error!(CAT, obj: stream.sinkpad, "Negative PTSs are not supported");
gst::FlowError::Error gst::FlowError::Error
})?; })?;
@ -332,7 +332,7 @@ impl FMP4Mux {
if pts < stream.current_position { if pts < stream.current_position {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Decreasing PTS {} < {} for intra-only stream", "Decreasing PTS {} < {} for intra-only stream",
pts, pts,
stream.current_position, stream.current_position,
@ -353,7 +353,7 @@ impl FMP4Mux {
let end_dts_position = duration.opt_add(dts_position).unwrap_or(dts_position); let end_dts_position = duration.opt_add(dts_position).unwrap_or(dts_position);
let signed_dts = segment.to_running_time_full(dts_position).ok_or_else(|| { let signed_dts = segment.to_running_time_full(dts_position).ok_or_else(|| {
gst::error!(CAT, obj: &stream.sinkpad, "Couldn't convert DTS to running time"); gst::error!(CAT, obj: stream.sinkpad, "Couldn't convert DTS to running time");
gst::FlowError::Error gst::FlowError::Error
})?; })?;
let mut dts = match signed_dts { let mut dts = match signed_dts {
@ -371,7 +371,7 @@ impl FMP4Mux {
let dts_offset = stream.dts_offset.unwrap(); let dts_offset = stream.dts_offset.unwrap();
if dts > dts_offset { if dts > dts_offset {
gst::warning!(CAT, obj: &stream.sinkpad, "DTS before first DTS"); gst::warning!(CAT, obj: stream.sinkpad, "DTS before first DTS");
gst::ClockTime::ZERO gst::ClockTime::ZERO
} else { } else {
dts_offset - dts dts_offset - dts
@ -385,7 +385,7 @@ impl FMP4Mux {
.ok_or_else(|| { .ok_or_else(|| {
gst::error!( gst::error!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Couldn't convert end DTS to running time" "Couldn't convert end DTS to running time"
); );
gst::FlowError::Error gst::FlowError::Error
@ -405,7 +405,7 @@ impl FMP4Mux {
let dts_offset = stream.dts_offset.unwrap(); let dts_offset = stream.dts_offset.unwrap();
if dts > dts_offset { if dts > dts_offset {
gst::warning!(CAT, obj: &stream.sinkpad, "End DTS before first DTS"); gst::warning!(CAT, obj: stream.sinkpad, "End DTS before first DTS");
gst::ClockTime::ZERO gst::ClockTime::ZERO
} else { } else {
dts_offset - dts dts_offset - dts
@ -419,7 +419,7 @@ impl FMP4Mux {
if dts < stream.current_position { if dts < stream.current_position {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Decreasing DTS {} < {}", "Decreasing DTS {} < {}",
dts, dts,
stream.current_position, stream.current_position,
@ -456,7 +456,7 @@ impl FMP4Mux {
if !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT) { if !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT) {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Starting new GOP at PTS {} DTS {} (DTS offset {})", "Starting new GOP at PTS {} DTS {} (DTS offset {})",
pts, pts,
dts.display(), dts.display(),
@ -480,7 +480,7 @@ impl FMP4Mux {
if let Some(prev_gop) = stream.queued_gops.get_mut(1) { if let Some(prev_gop) = stream.queued_gops.get_mut(1) {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Updating previous GOP starting at PTS {} to end PTS {} DTS {}", "Updating previous GOP starting at PTS {} to end PTS {} DTS {}",
prev_gop.earliest_pts, prev_gop.earliest_pts,
pts, pts,
@ -500,7 +500,7 @@ impl FMP4Mux {
if !intra_only { if !intra_only {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Previous GOP has final earliest PTS at {}", "Previous GOP has final earliest PTS at {}",
prev_gop.earliest_pts prev_gop.earliest_pts
); );
@ -530,7 +530,7 @@ impl FMP4Mux {
if gop.earliest_pts > pts && !gop.final_earliest_pts { if gop.earliest_pts > pts && !gop.final_earliest_pts {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Updating current GOP earliest PTS from {} to {}", "Updating current GOP earliest PTS from {} to {}",
gop.earliest_pts, gop.earliest_pts,
pts pts
@ -542,7 +542,7 @@ impl FMP4Mux {
if prev_gop.end_pts < pts { if prev_gop.end_pts < pts {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Updating previous GOP starting PTS {} end time from {} to {}", "Updating previous GOP starting PTS {} end time from {} to {}",
pts, pts,
prev_gop.end_pts, prev_gop.end_pts,
@ -562,7 +562,7 @@ impl FMP4Mux {
if gop.start_pts <= dts && !gop.final_earliest_pts { if gop.start_pts <= dts && !gop.final_earliest_pts {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"GOP has final earliest PTS at {}", "GOP has final earliest PTS at {}",
gop.earliest_pts gop.earliest_pts
); );
@ -575,7 +575,7 @@ impl FMP4Mux {
} else { } else {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Waiting for keyframe at the beginning of the stream" "Waiting for keyframe at the beginning of the stream"
); );
} }
@ -586,7 +586,7 @@ impl FMP4Mux {
) { ) {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Queued full GOPs duration updated to {}", "Queued full GOPs duration updated to {}",
prev_gop.end_pts.saturating_sub(first_gop.earliest_pts), prev_gop.end_pts.saturating_sub(first_gop.earliest_pts),
); );
@ -594,7 +594,7 @@ impl FMP4Mux {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Queued duration updated to {}", "Queued duration updated to {}",
Option::zip(stream.queued_gops.front(), stream.queued_gops.back()) Option::zip(stream.queued_gops.front(), stream.queued_gops.back())
.map(|(end, start)| end.end_pts.saturating_sub(start.start_pts)) .map(|(end, start)| end.end_pts.saturating_sub(start.start_pts))
@ -669,7 +669,7 @@ impl FMP4Mux {
fragment_end_pts.unwrap_or(fragment_start_pts + settings.fragment_duration); fragment_end_pts.unwrap_or(fragment_start_pts + settings.fragment_duration);
gst::trace!( gst::trace!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Draining up to end PTS {} / duration {}", "Draining up to end PTS {} / duration {}",
dequeue_end_pts, dequeue_end_pts,
dequeue_end_pts - fragment_start_pts dequeue_end_pts - fragment_start_pts
@ -704,7 +704,7 @@ impl FMP4Mux {
fragment_end_pts = Some(last_gop.end_pts); fragment_end_pts = Some(last_gop.end_pts);
gst::info!( gst::info!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Draining up to PTS {} for this fragment", "Draining up to PTS {} for this fragment",
last_gop.end_pts, last_gop.end_pts,
); );
@ -719,7 +719,7 @@ impl FMP4Mux {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Don't have a complete GOP for the first stream on timeout in a live pipeline", "Don't have a complete GOP for the first stream on timeout in a live pipeline",
); );
@ -733,7 +733,7 @@ impl FMP4Mux {
if gops.is_empty() { if gops.is_empty() {
gst::info!( gst::info!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Draining no buffers", "Draining no buffers",
); );
@ -772,7 +772,7 @@ impl FMP4Mux {
gst::info!( gst::info!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Draining {} worth of buffers starting at PTS {} DTS {}, DTS offset {}", "Draining {} worth of buffers starting at PTS {} DTS {}, DTS offset {}",
end_pts.saturating_sub(earliest_pts), end_pts.saturating_sub(earliest_pts),
earliest_pts, earliest_pts,
@ -786,7 +786,7 @@ impl FMP4Mux {
) { ) {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Queued full GOPs duration updated to {}", "Queued full GOPs duration updated to {}",
prev_gop.end_pts.saturating_sub(first_gop.earliest_pts), prev_gop.end_pts.saturating_sub(first_gop.earliest_pts),
); );
@ -794,7 +794,7 @@ impl FMP4Mux {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &stream.sinkpad, obj: stream.sinkpad,
"Queued duration updated to {}", "Queued duration updated to {}",
Option::zip(stream.queued_gops.front(), stream.queued_gops.back()) Option::zip(stream.queued_gops.front(), stream.queued_gops.back())
.map(|(end, start)| end.end_pts.saturating_sub(start.start_pts)) .map(|(end, start)| end.end_pts.saturating_sub(start.start_pts))
@ -847,19 +847,15 @@ impl FMP4Mux {
let dts = buffer.dts.unwrap(); let dts = buffer.dts.unwrap();
if pts > dts { if pts > dts {
Some( Some(i64::try_from((pts - dts).nseconds()).map_err(|_| {
i64::try_from((pts - dts).nseconds()) gst::error!(CAT, obj: stream.sinkpad, "Too big PTS/DTS difference");
.map_err(|_| { gst::FlowError::Error
gst::error!(CAT, obj: &stream.sinkpad, "Too big PTS/DTS difference"); })?)
gst::FlowError::Error
})?,
)
} else { } else {
let diff = i64::try_from((dts - pts).nseconds()) let diff = i64::try_from((dts - pts).nseconds()).map_err(|_| {
.map_err(|_| { gst::error!(CAT, obj: stream.sinkpad, "Too big PTS/DTS difference");
gst::error!(CAT, obj: &stream.sinkpad, "Too big PTS/DTS difference"); gst::FlowError::Error
gst::FlowError::Error })?;
})?;
Some(-diff) Some(-diff)
} }
}; };
@ -933,7 +929,7 @@ impl FMP4Mux {
None => { None => {
gst::error!( gst::error!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"No reference timestamp set on any buffers in the first fragment", "No reference timestamp set on any buffers in the first fragment",
); );
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
@ -984,7 +980,7 @@ impl FMP4Mux {
None => { None => {
gst::error!( gst::error!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"No reference timestamp set on all buffers" "No reference timestamp set on all buffers"
); );
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
@ -1017,7 +1013,7 @@ impl FMP4Mux {
None => { None => {
gst::error!( gst::error!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"No reference timestamp set on all buffers" "No reference timestamp set on all buffers"
); );
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
@ -1045,7 +1041,7 @@ impl FMP4Mux {
if utc_time_dts < state.streams[idx].current_utc_time { if utc_time_dts < state.streams[idx].current_utc_time {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"Decreasing UTC DTS timestamp for buffer {} < {}", "Decreasing UTC DTS timestamp for buffer {} < {}",
utc_time_dts, utc_time_dts,
state.streams[idx].current_utc_time, state.streams[idx].current_utc_time,
@ -1059,7 +1055,7 @@ impl FMP4Mux {
gst::trace!( gst::trace!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"Updating buffer timestamp from {} to relative UTC DTS time {} / absolute DTS time {}, UTC PTS time {}", "Updating buffer timestamp from {} to relative UTC DTS time {} / absolute DTS time {}, UTC PTS time {}",
buffer.timestamp, buffer.timestamp,
timestamp, timestamp,
@ -1090,7 +1086,7 @@ impl FMP4Mux {
gst::trace!( gst::trace!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"Updating buffer with timestamp {} duration from {} to relative UTC duration {}", "Updating buffer with timestamp {} duration from {} to relative UTC duration {}",
buffer.timestamp, buffer.timestamp,
buffer.duration, buffer.duration,
@ -1101,7 +1097,7 @@ impl FMP4Mux {
} else if let Ok(Some(common_duration)) = common_duration { } else if let Ok(Some(common_duration)) = common_duration {
gst::trace!( gst::trace!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"Updating last buffer with timestamp {} duration from {} to common relative UTC duration {}", "Updating last buffer with timestamp {} duration from {} to common relative UTC duration {}",
buffer.timestamp, buffer.timestamp,
buffer.duration, buffer.duration,
@ -1112,7 +1108,7 @@ impl FMP4Mux {
} else { } else {
gst::trace!( gst::trace!(
CAT, CAT,
obj: &state.streams[idx].sinkpad, obj: state.streams[idx].sinkpad,
"Keeping last buffer with timestamp {} duration at {}", "Keeping last buffer with timestamp {} duration at {}",
buffer.timestamp, buffer.timestamp,
buffer.duration, buffer.duration,
@ -1126,7 +1122,7 @@ impl FMP4Mux {
} }
if let Some(start_time) = start_time { if let Some(start_time) = start_time {
gst::debug!(CAT, obj: &state.streams[idx].sinkpad, "Fragment starting at UTC time {}", start_time); gst::debug!(CAT, obj: state.streams[idx].sinkpad, "Fragment starting at UTC time {}", start_time);
timing_info.as_mut().unwrap().start_time = start_time; timing_info.as_mut().unwrap().start_time = start_time;
} else { } else {
assert!(timing_info.is_none()); assert!(timing_info.is_none());
@ -1441,12 +1437,12 @@ impl FMP4Mux {
let caps = match pad.current_caps() { let caps = match pad.current_caps() {
Some(caps) => caps, Some(caps) => caps,
None => { None => {
gst::warning!(CAT, obj: &pad, "Skipping pad without caps"); gst::warning!(CAT, obj: pad, "Skipping pad without caps");
continue; continue;
} }
}; };
gst::info!(CAT, obj: &pad, "Configuring caps {:?}", caps); gst::info!(CAT, obj: pad, "Configuring caps {:?}", caps);
let s = caps.structure(0).unwrap(); let s = caps.structure(0).unwrap();
@ -1454,7 +1450,7 @@ impl FMP4Mux {
match s.name() { match s.name() {
"video/x-h264" | "video/x-h265" => { "video/x-h264" | "video/x-h265" => {
if !s.has_field_with_type("codec_data", gst::Buffer::static_type()) { if !s.has_field_with_type("codec_data", gst::Buffer::static_type()) {
gst::error!(CAT, obj: &pad, "Received caps without codec_data"); gst::error!(CAT, obj: pad, "Received caps without codec_data");
return Err(gst::FlowError::NotNegotiated); return Err(gst::FlowError::NotNegotiated);
} }
} }
@ -1463,7 +1459,7 @@ impl FMP4Mux {
} }
"audio/mpeg" => { "audio/mpeg" => {
if !s.has_field_with_type("codec_data", gst::Buffer::static_type()) { if !s.has_field_with_type("codec_data", gst::Buffer::static_type()) {
gst::error!(CAT, obj: &pad, "Received caps without codec_data"); gst::error!(CAT, obj: pad, "Received caps without codec_data");
return Err(gst::FlowError::NotNegotiated); return Err(gst::FlowError::NotNegotiated);
} }
intra_only = true; intra_only = true;
@ -2004,7 +2000,7 @@ impl AggregatorImpl for FMP4Mux {
{ {
Some(segment) => segment, Some(segment) => segment,
None => { None => {
gst::error!(CAT, obj: &stream.sinkpad, "Got buffer before segment"); gst::error!(CAT, obj: stream.sinkpad, "Got buffer before segment");
return Err(gst::FlowError::Error); return Err(gst::FlowError::Error);
} }
}; };
@ -2024,7 +2020,7 @@ impl AggregatorImpl for FMP4Mux {
if queued_end_pts.saturating_sub(fragment_start_pts) if queued_end_pts.saturating_sub(fragment_start_pts)
>= settings.fragment_duration >= settings.fragment_duration
{ {
gst::debug!(CAT, obj: &stream.sinkpad, "Stream queued enough data for this fragment"); gst::debug!(CAT, obj: stream.sinkpad, "Stream queued enough data for this fragment");
stream.fragment_filled = true; stream.fragment_filled = true;
} }
} }
@ -2085,7 +2081,7 @@ impl AggregatorImpl for FMP4Mux {
if queued_end_pts.saturating_sub(earliest_pts) if queued_end_pts.saturating_sub(earliest_pts)
>= settings.fragment_duration >= settings.fragment_duration
{ {
gst::debug!(CAT, obj: &stream.sinkpad, "Stream queued enough data for this fragment"); gst::debug!(CAT, obj: stream.sinkpad, "Stream queued enough data for this fragment");
stream.fragment_filled = true; stream.fragment_filled = true;
} }
} }

18
net/ndi/src/ndisrc/receiver.rs
View file

@ -760,13 +760,13 @@ impl Receiver {
let flushing = { let flushing = {
let queue = (receiver.0.queue.0).0.lock().unwrap(); let queue = (receiver.0.queue.0).0.lock().unwrap();
if queue.shutdown { if queue.shutdown {
gst::debug!(CAT, obj: &element, "Shutting down"); gst::debug!(CAT, obj: element, "Shutting down");
break; break;
} }
// If an error happened in the meantime, just go out of here // If an error happened in the meantime, just go out of here
if queue.error.is_some() { if queue.error.is_some() {
gst::error!(CAT, obj: &element, "Error while waiting for connection"); gst::error!(CAT, obj: element, "Error while waiting for connection");
return; return;
} }
@ -781,7 +781,7 @@ impl Receiver {
let res = match recv.capture(50) { let res = match recv.capture(50) {
_ if flushing => { _ if flushing => {
gst::debug!(CAT, obj: &element, "Flushing"); gst::debug!(CAT, obj: element, "Flushing");
Err(gst::FlowError::Flushing) Err(gst::FlowError::Flushing)
} }
Err(_) => { Err(_) => {
@ -793,11 +793,11 @@ impl Receiver {
Err(gst::FlowError::Error) Err(gst::FlowError::Error)
} }
Ok(None) if timeout > 0 && timer.elapsed().as_millis() >= timeout as u128 => { Ok(None) if timeout > 0 && timer.elapsed().as_millis() >= timeout as u128 => {
gst::debug!(CAT, obj: &element, "Timed out -- assuming EOS",); gst::debug!(CAT, obj: element, "Timed out -- assuming EOS",);
Err(gst::FlowError::Eos) Err(gst::FlowError::Eos)
} }
Ok(None) => { Ok(None) => {
gst::debug!(CAT, obj: &element, "No frame received yet, retry"); gst::debug!(CAT, obj: element, "No frame received yet, retry");
continue; continue;
} }
Ok(Some(Frame::Video(frame))) => { Ok(Some(Frame::Video(frame))) => {
@ -832,7 +832,7 @@ impl Receiver {
if let Some(metadata) = frame.metadata() { if let Some(metadata) = frame.metadata() {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &element, obj: element,
"Received metadata at timecode {}: {}", "Received metadata at timecode {}: {}",
(frame.timecode() as u64 * 100).nseconds(), (frame.timecode() as u64 * 100).nseconds(),
metadata, metadata,
@ -849,7 +849,7 @@ impl Receiver {
while queue.buffer_queue.len() > receiver.0.max_queue_length { while queue.buffer_queue.len() > receiver.0.max_queue_length {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Dropping old buffer -- queue has {} items", "Dropping old buffer -- queue has {} items",
queue.buffer_queue.len() queue.buffer_queue.len()
); );
@ -860,7 +860,7 @@ impl Receiver {
timer = time::Instant::now(); timer = time::Instant::now();
} }
Err(gst::FlowError::Eos) => { Err(gst::FlowError::Eos) => {
gst::debug!(CAT, obj: &element, "Signalling EOS"); gst::debug!(CAT, obj: element, "Signalling EOS");
let mut queue = (receiver.0.queue.0).0.lock().unwrap(); let mut queue = (receiver.0.queue.0).0.lock().unwrap();
queue.timeout = true; queue.timeout = true;
(receiver.0.queue.0).1.notify_one(); (receiver.0.queue.0).1.notify_one();
@ -874,7 +874,7 @@ impl Receiver {
timer = time::Instant::now(); timer = time::Instant::now();
} }
Err(err) => { Err(err) => {
gst::error!(CAT, obj: &element, "Signalling error"); gst::error!(CAT, obj: element, "Signalling error");
let mut queue = (receiver.0.queue.0).0.lock().unwrap(); let mut queue = (receiver.0.queue.0).0.lock().unwrap();
if queue.error.is_none() { if queue.error.is_none() {
queue.error = Some(err); queue.error = Some(err);

2
net/rtp/src/gcc/imp.rs
View file

@ -1106,7 +1106,7 @@ impl BandwidthEstimator {
if !list.is_empty() { if !list.is_empty() {
if let Err(err) = bwe.imp().push_list(list) { if let Err(err) = bwe.imp().push_list(list) {
gst::error!(CAT, obj: &bwe, "pause task, reason: {err:?}"); gst::error!(CAT, obj: bwe, "pause task, reason: {err:?}");
pause() pause()
} }
} }

26
net/webrtc/src/signaller/imp.rs
View file

@ -85,7 +85,7 @@ impl Signaller {
let send_task_handle = task::spawn(async move { let send_task_handle = task::spawn(async move {
while let Some(msg) = websocket_receiver.next().await { while let Some(msg) = websocket_receiver.next().await {
if let Some(element) = element_clone.upgrade() { if let Some(element) = element_clone.upgrade() {
gst::trace!(CAT, obj: &element, "Sending websocket message {:?}", msg); gst::trace!(CAT, obj: element, "Sending websocket message {:?}", msg);
} }
ws_sink ws_sink
.send(WsMessage::Text(serde_json::to_string(&msg).unwrap())) .send(WsMessage::Text(serde_json::to_string(&msg).unwrap()))
@ -93,7 +93,7 @@ impl Signaller {
} }
if let Some(element) = element_clone.upgrade() { if let Some(element) = element_clone.upgrade() {
gst::info!(CAT, obj: &element, "Done sending"); gst::info!(CAT, obj: element, "Done sending");
} }
ws_sink.send(WsMessage::Close(None)).await?; ws_sink.send(WsMessage::Close(None)).await?;
@ -121,14 +121,14 @@ impl Signaller {
if let Some(element) = element_clone.upgrade() { if let Some(element) = element_clone.upgrade() {
match msg { match msg {
Ok(WsMessage::Text(msg)) => { Ok(WsMessage::Text(msg)) => {
gst::trace!(CAT, obj: &element, "Received message {}", msg); gst::trace!(CAT, obj: element, "Received message {}", msg);
if let Ok(msg) = serde_json::from_str::<p::OutgoingMessage>(&msg) { if let Ok(msg) = serde_json::from_str::<p::OutgoingMessage>(&msg) {
match msg { match msg {
p::OutgoingMessage::Welcome { peer_id } => { p::OutgoingMessage::Welcome { peer_id } => {
gst::info!( gst::info!(
CAT, CAT,
obj: &element, obj: element,
"We are registered with the server, our peer id is {}", "We are registered with the server, our peer id is {}",
peer_id peer_id
); );
@ -140,14 +140,14 @@ impl Signaller {
if let Err(err) = if let Err(err) =
element.start_session(&session_id, &peer_id) element.start_session(&session_id, &peer_id)
{ {
gst::warning!(CAT, obj: &element, "{}", err); gst::warning!(CAT, obj: element, "{}", err);
} }
} }
p::OutgoingMessage::EndSession(session_info) => { p::OutgoingMessage::EndSession(session_info) => {
if let Err(err) = if let Err(err) =
element.end_session(&session_info.session_id) element.end_session(&session_info.session_id)
{ {
gst::warning!(CAT, obj: &element, "{}", err); gst::warning!(CAT, obj: element, "{}", err);
} }
} }
p::OutgoingMessage::Peer(p::PeerMessage { p::OutgoingMessage::Peer(p::PeerMessage {
@ -165,7 +165,7 @@ impl Signaller {
.unwrap(), .unwrap(),
), ),
) { ) {
gst::warning!(CAT, obj: &element, "{}", err); gst::warning!(CAT, obj: element, "{}", err);
} }
} }
p::PeerMessageInner::Sdp(p::SdpMessage::Offer { p::PeerMessageInner::Sdp(p::SdpMessage::Offer {
@ -173,7 +173,7 @@ impl Signaller {
}) => { }) => {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Ignoring offer from peer" "Ignoring offer from peer"
); );
} }
@ -187,14 +187,14 @@ impl Signaller {
None, None,
&candidate, &candidate,
) { ) {
gst::warning!(CAT, obj: &element, "{}", err); gst::warning!(CAT, obj: element, "{}", err);
} }
} }
}, },
_ => { _ => {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Ignoring unsupported message {:?}", "Ignoring unsupported message {:?}",
msg msg
); );
@ -203,7 +203,7 @@ impl Signaller {
} else { } else {
gst::error!( gst::error!(
CAT, CAT,
obj: &element, obj: element,
"Unknown message from server: {}", "Unknown message from server: {}",
msg msg
); );
@ -215,7 +215,7 @@ impl Signaller {
Ok(WsMessage::Close(reason)) => { Ok(WsMessage::Close(reason)) => {
gst::info!( gst::info!(
CAT, CAT,
obj: &element, obj: element,
"websocket connection closed: {:?}", "websocket connection closed: {:?}",
reason reason
); );
@ -235,7 +235,7 @@ impl Signaller {
} }
if let Some(element) = element_clone.upgrade() { if let Some(element) = element_clone.upgrade() {
gst::info!(CAT, obj: &element, "Stopped websocket receiving"); gst::info!(CAT, obj: element, "Stopped websocket receiving");
} }
}); });

20
net/webrtc/src/webrtcsink/imp.rs
View file

@ -1302,7 +1302,7 @@ impl WebRTCSink {
Ok(None) => { Ok(None) => {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Promise returned without a reply for {}", "Promise returned without a reply for {}",
session_id session_id
); );
@ -1312,7 +1312,7 @@ impl WebRTCSink {
Err(err) => { Err(err) => {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Promise returned with an error for {}: {:?}", "Promise returned with an error for {}: {:?}",
session_id, session_id,
err err
@ -1522,7 +1522,7 @@ impl WebRTCSink {
let this = Self::from_instance(&element); let this = Self::from_instance(&element);
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Connection state for in session {} (peer {}) failed", "Connection state for in session {} (peer {}) failed",
session_id_clone, session_id_clone,
peer_id_clone peer_id_clone
@ -1532,7 +1532,7 @@ impl WebRTCSink {
_ => { _ => {
gst::log!( gst::log!(
CAT, CAT,
obj: &element, obj: element,
"Connection state in session {} (peer {}) changed: {:?}", "Connection state in session {} (peer {}) changed: {:?}",
session_id_clone, session_id_clone,
peer_id_clone, peer_id_clone,
@ -1556,7 +1556,7 @@ impl WebRTCSink {
gst_webrtc::WebRTCICEConnectionState::Failed => { gst_webrtc::WebRTCICEConnectionState::Failed => {
gst::warning!( gst::warning!(
CAT, CAT,
obj: &element, obj: element,
"Ice connection state in session {} (peer {}) failed", "Ice connection state in session {} (peer {}) failed",
session_id_clone, session_id_clone,
peer_id_clone, peer_id_clone,
@ -1566,7 +1566,7 @@ impl WebRTCSink {
_ => { _ => {
gst::log!( gst::log!(
CAT, CAT,
obj: &element, obj: element,
"Ice connection state in session {} (peer {}) changed: {:?}", "Ice connection state in session {} (peer {}) changed: {:?}",
session_id_clone, session_id_clone,
peer_id_clone, peer_id_clone,
@ -1603,7 +1603,7 @@ impl WebRTCSink {
if let Some(element) = element_clone.upgrade() { if let Some(element) = element_clone.upgrade() {
gst::log!( gst::log!(
CAT, CAT,
obj: &element, obj: element,
"Ice gathering state in session {} (peer {}) changed: {:?}", "Ice gathering state in session {} (peer {}) changed: {:?}",
session_id_clone, session_id_clone,
peer_id_clone, peer_id_clone,
@ -1708,7 +1708,7 @@ impl WebRTCSink {
} }
gst::MessageView::Latency(..) => { gst::MessageView::Latency(..) => {
if let Some(pipeline) = pipeline_clone.upgrade() { if let Some(pipeline) = pipeline_clone.upgrade() {
gst::info!(CAT, obj: &pipeline, "Recalculating latency"); gst::info!(CAT, obj: pipeline, "Recalculating latency");
let _ = pipeline.recalculate_latency(); let _ = pipeline.recalculate_latency();
} }
} }
@ -2295,7 +2295,7 @@ impl WebRTCSink {
match fut.await { match fut.await {
Ok(Err(err)) => { Ok(Err(err)) => {
gst::error!(CAT, obj: &element, "error: {}", err); gst::error!(CAT, obj: element, "error: {}", err);
gst::element_error!( gst::element_error!(
element, element,
gst::StreamError::CodecNotFound, gst::StreamError::CodecNotFound,
@ -2638,7 +2638,7 @@ impl ObjectImpl for WebRTCSink {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &element, obj: element,
"applying default configuration on encoder {:?}", "applying default configuration on encoder {:?}",
enc enc
); );

4
text/json/src/jsongstparse/imp.rs
View file

@ -542,13 +542,13 @@ impl JsonGstParse {
Ok(buffer) => Some(buffer), Ok(buffer) => Some(buffer),
Err(gst::FlowError::Eos) => None, Err(gst::FlowError::Eos) => None,
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(CAT, obj: &self.sinkpad, "Pausing after pulling buffer, reason: flushing"); gst::debug!(CAT, obj: self.sinkpad, "Pausing after pulling buffer, reason: flushing");
self.sinkpad.pause_task().unwrap(); self.sinkpad.pause_task().unwrap();
return; return;
} }
Err(flow) => { Err(flow) => {
gst::error!(CAT, obj: &self.sinkpad, "Failed to pull, reason: {:?}", flow); gst::error!(CAT, obj: self.sinkpad, "Failed to pull, reason: {:?}", flow);
gst::element_imp_error!( gst::element_imp_error!(
self, self,

4
utils/fallbackswitch/src/fallbacksrc/imp.rs
View file

@ -1914,7 +1914,7 @@ impl FallbackSrc {
Some(gst::PadProbeData::Event(ref ev)) if ev.type_() == gst::EventType::Eos => { Some(gst::PadProbeData::Event(ref ev)) if ev.type_() == gst::EventType::Eos => {
gst::debug!( gst::debug!(
CAT, CAT,
obj: &element, obj: element,
"Received EOS from {}source on pad {}", "Received EOS from {}source on pad {}",
if fallback_source { "fallback " } else { "" }, if fallback_source { "fallback " } else { "" },
pad.name() pad.name()
@ -3083,7 +3083,7 @@ impl FallbackSrc {
Some(element) => element, Some(element) => element,
}; };
gst::debug!(CAT, obj: &element, "Woke up, retrying"); gst::debug!(CAT, obj: element, "Woke up, retrying");
element.call_async(move |element| { element.call_async(move |element| {
let imp = element.imp(); let imp = element.imp();

2
utils/tracers/src/pipeline_snapshot/imp.rs
View file

@ -207,7 +207,7 @@ impl PipelineSnapshot {
for pipeline in pipelines.into_iter() { for pipeline in pipelines.into_iter() {
let pipeline = pipeline.downcast::<gst::Pipeline>().unwrap(); let pipeline = pipeline.downcast::<gst::Pipeline>().unwrap();
gst::debug!(CAT, obj: &tracer, "dump {}", pipeline.name()); gst::debug!(CAT, obj: tracer, "dump {}", pipeline.name());
let dump_name = format!("{}{}", settings.dot_prefix, pipeline.name()); let dump_name = format!("{}{}", settings.dot_prefix, pipeline.name());

6
utils/uriplaylistbin/src/uriplaylistbin/imp.rs
View file

@ -1421,7 +1421,7 @@ impl UriPlaylistBin {
// block pad until next item is ready // block pad until next item is ready
gst::log!( gst::log!(
CAT, CAT,
obj: &element, obj: element,
"blocking pad {}:{} until next item is ready", "blocking pad {}:{} until next item is ready",
parent.name(), parent.name(),
pad.name() pad.name()
@ -1438,7 +1438,7 @@ impl UriPlaylistBin {
gst::log!( gst::log!(
CAT, CAT,
obj: &element, obj: element,
"pad {}:{} has been unblocked", "pad {}:{} has been unblocked",
parent.name(), parent.name(),
pad.name() pad.name()
@ -1454,7 +1454,7 @@ impl UriPlaylistBin {
// all the streams are eos, item is now done // all the streams are eos, item is now done
gst::log!( gst::log!(
CAT, CAT,
obj: &element, obj: element,
"all streams of item #{} are eos", "all streams of item #{} are eos",
item.index() item.index()
); );

4
video/closedcaption/src/mcc_parse/imp.rs
View file

@ -756,13 +756,13 @@ impl MccParse {
Ok(buffer) => Some(buffer), Ok(buffer) => Some(buffer),
Err(gst::FlowError::Eos) => None, Err(gst::FlowError::Eos) => None,
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(CAT, obj: &self.sinkpad, "Pausing after pulling buffer, reason: flushing"); gst::debug!(CAT, obj: self.sinkpad, "Pausing after pulling buffer, reason: flushing");
let _ = self.sinkpad.pause_task(); let _ = self.sinkpad.pause_task();
return; return;
} }
Err(flow) => { Err(flow) => {
gst::error!(CAT, obj: &self.sinkpad, "Failed to pull, reason: {:?}", flow); gst::error!(CAT, obj: self.sinkpad, "Failed to pull, reason: {:?}", flow);
gst::element_imp_error!( gst::element_imp_error!(
self, self,

4
video/closedcaption/src/scc_parse/imp.rs
View file

@ -639,13 +639,13 @@ impl SccParse {
Ok(buffer) => Some(buffer), Ok(buffer) => Some(buffer),
Err(gst::FlowError::Eos) => None, Err(gst::FlowError::Eos) => None,
Err(gst::FlowError::Flushing) => { Err(gst::FlowError::Flushing) => {
gst::debug!(CAT, obj: &self.sinkpad, "Pausing after pulling buffer, reason: flushing"); gst::debug!(CAT, obj: self.sinkpad, "Pausing after pulling buffer, reason: flushing");
let _ = self.sinkpad.pause_task(); let _ = self.sinkpad.pause_task();
return; return;
} }
Err(flow) => { Err(flow) => {
gst::error!(CAT, obj: &self.sinkpad, "Failed to pull, reason: {:?}", flow); gst::error!(CAT, obj: self.sinkpad, "Failed to pull, reason: {:?}", flow);
gst::element_imp_error!( gst::element_imp_error!(
self, self,