Skip to content

fix: Various improvements to WHEP input, AES67 output, and logging #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
srperens merged 1 commit into from
Nov 24, 2025
Merged

fix: Various improvements to WHEP input, AES67 output, and logging #36

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

15 changes: 9 additions & 6 deletions backend/src/api/flows.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use strom_types::{
Flow, FlowId,
};
use tempfile::NamedTempFile;
use tracing::{error, info};
use tracing::{error, info, trace};
use utoipa;

use crate::layout;
Expand Down Expand Up @@ -829,7 +829,7 @@ pub async fn get_flow_latency(
State(state): State<AppState>,
Path(id): Path<FlowId>,
) -> Result<Json<LatencyResponse>, (StatusCode, Json<ErrorResponse>)> {
info!("Getting latency for flow {}", id);
trace!("Getting latency for flow {}", id);

let latency = state.get_flow_latency(&id).await.ok_or_else(|| {
(
Expand All @@ -841,9 +841,12 @@ pub async fn get_flow_latency(
})?;

let (min_ns, max_ns, live) = latency;
info!(
trace!(
"Flow {} latency: min={}ns, max={}ns, live={}",
id, min_ns, max_ns, live
id,
min_ns,
max_ns,
live
);

Ok(Json(LatencyResponse::new(min_ns, max_ns, live)))
Expand All @@ -870,7 +873,7 @@ pub async fn get_flow_stats(
State(state): State<AppState>,
Path(id): Path<FlowId>,
) -> Result<Json<FlowStatsResponse>, (StatusCode, Json<ErrorResponse>)> {
info!("Getting statistics for flow {}", id);
trace!("Getting statistics for flow {}", id);

let stats = state.get_flow_stats(&id).await.ok_or_else(|| {
(
Expand All @@ -881,7 +884,7 @@ pub async fn get_flow_stats(
)
})?;

info!(
trace!(
"Flow {} stats: {} blocks with statistics",
id,
stats.block_stats.len()
Expand Down
9 changes: 8 additions & 1 deletion backend/src/blocks/builtin/aes67.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -480,12 +480,19 @@ impl BlockBuilder for AES67OutputBuilder {
.build()
.map_err(|e| BlockBuildError::ElementCreation(format!("{}: {}", payloader_type, e)))?;

// Set processing-deadline to match ptime for proper timing
let processing_deadline_ns = ptime_ns as u64;

let udpsink = gst::ElementFactory::make("udpsink")
.name(&udpsink_id)
.property("host", &host)
.property("port", port)
.property("async", false)
.property("sync", false)
.property("sync", true)
.property(
"processing-deadline",
gst::ClockTime::from_nseconds(processing_deadline_ns),
)
.build()
.map_err(|e| BlockBuildError::ElementCreation(format!("udpsink: {}", e)))?;

Expand Down
47 changes: 45 additions & 2 deletions backend/src/blocks/builtin/whep.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,10 +70,23 @@ impl BlockBuilder for WHEPInputBuilder {
})
.unwrap_or_else(|| "stun://stun.l.google.com:19302".to_string());

// Get mixer latency (default 30ms - lower than default 200ms for lower latency)
let mixer_latency_ms = properties
.get("mixer_latency_ms")
.and_then(|v| {
if let PropertyValue::Int(i) = v {
Some(*i as u64)
} else {
None
}
})
.unwrap_or(30);

// Create namespaced element IDs
let instance_id_owned = instance_id.to_string();
let whepclientsrc_id = format!("{}:whepclientsrc", instance_id);
let liveadder_id = format!("{}:liveadder", instance_id);
let capsfilter_id = format!("{}:capsfilter", instance_id);
let output_audioconvert_id = format!("{}:output_audioconvert", instance_id);
let output_audioresample_id = format!("{}:output_audioresample", instance_id);

Expand All @@ -95,8 +108,10 @@ impl BlockBuilder for WHEPInputBuilder {
}

// Create liveadder - this is our always-present mixer for dynamic audio streams
// latency property is in milliseconds as a guint (u32)
let liveadder = gst::ElementFactory::make("liveadder")
.name(&liveadder_id)
.property("latency", mixer_latency_ms as u32)
.build()
.map_err(|e| BlockBuildError::ElementCreation(format!("liveadder: {}", e)))?;

Expand All @@ -112,7 +127,18 @@ impl BlockBuilder for WHEPInputBuilder {
BlockBuildError::ElementCreation(format!("audiotestsrc (silence): {}", e))
})?;

// Create output audio processing chain (after liveadder)
// Create capsfilter to enforce 48kHz stereo audio after liveadder
let caps = gst::Caps::builder("audio/x-raw")
.field("rate", 48000i32)
.field("channels", 2i32)
.build();
let capsfilter = gst::ElementFactory::make("capsfilter")
.name(&capsfilter_id)
.property("caps", &caps)
.build()
.map_err(|e| BlockBuildError::ElementCreation(format!("capsfilter: {}", e)))?;

// Create output audio processing chain (after liveadder -> capsfilter)
let output_audioconvert = gst::ElementFactory::make("audioconvert")
.name(&output_audioconvert_id)
.build()
Expand Down Expand Up @@ -302,7 +328,7 @@ impl BlockBuilder for WHEPInputBuilder {
whep_endpoint, stun_server
);

// Internal links: silence -> liveadder -> audioconvert -> audioresample
// Internal links: silence -> liveadder -> capsfilter -> audioconvert -> audioresample
// The whepclientsrc pads are linked dynamically via pad-added callback
let internal_links = vec![
(
Expand All @@ -311,6 +337,10 @@ impl BlockBuilder for WHEPInputBuilder {
),
(
format!("{}:src", liveadder_id),
format!("{}:sink", capsfilter_id),
),
(
format!("{}:src", capsfilter_id),
format!("{}:sink", output_audioconvert_id),
),
(
Expand All @@ -324,6 +354,7 @@ impl BlockBuilder for WHEPInputBuilder {
(whepclientsrc_id, whepclientsrc),
(silence_id, silence),
(liveadder_id, liveadder),
(capsfilter_id, capsfilter),
(output_audioconvert_id, output_audioconvert),
(output_audioresample_id, output_audioresample),
],
Expand Down Expand Up @@ -866,6 +897,18 @@ fn whep_input_definition() -> BlockDefinition {
transform: None,
},
},
ExposedProperty {
name: "mixer_latency_ms".to_string(),
label: "Mixer Latency (ms)".to_string(),
description: "Latency of the audio mixer in milliseconds (default 30ms, lower = less delay but may cause glitches)".to_string(),
property_type: PropertyType::Int,
default_value: Some(PropertyValue::Int(30)),
mapping: PropertyMapping {
element_id: "_block".to_string(),
property_name: "mixer_latency_ms".to_string(),
transform: None,
},
},
],
external_pads: ExternalPads {
inputs: vec![],
Expand Down
57 changes: 31 additions & 26 deletions backend/src/gst/pipeline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ use std::collections::HashMap;
use strom_types::flow::ThreadPriorityStatus;
use strom_types::{Element, Flow, FlowId, Link, PipelineState, PropertyValue, StromEvent};
use thiserror::Error;
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info, trace, warn};

/// Result of processing links with automatic tee insertion.
struct ProcessedLinks {
Expand Down Expand Up @@ -2015,9 +2015,12 @@ impl PipelineManager {
let (live, min, max) = query.result();
let min_ns = min.nseconds();
let max_ns = max.map_or(u64::MAX, |t| t.nseconds());
info!(
trace!(
"Pipeline '{}' latency query: live={}, min={}ns, max={}ns",
self.flow_name, live, min_ns, max_ns
self.flow_name,
live,
min_ns,
max_ns
);

// If pipeline is live and has meaningful latency, use it
Expand All @@ -2030,9 +2033,11 @@ impl PipelineManager {
let sink_latency = self.query_sink_latency();
if let Some((sink_min, sink_max)) = sink_latency {
if sink_min > 0 {
info!(
trace!(
"Pipeline '{}' using sink latency: min={}ns, max={}ns",
self.flow_name, sink_min, sink_max
self.flow_name,
sink_min,
sink_max
);
return Some((sink_min, sink_max, live));
}
Expand All @@ -2041,7 +2046,7 @@ impl PipelineManager {
// Return pipeline values even if 0 (user sees it's not live)
Some((min_ns, max_ns, live))
} else {
info!(
trace!(
"Pipeline '{}' latency query failed (may not be in playing state)",
self.flow_name
);
Expand Down Expand Up @@ -2174,20 +2179,20 @@ impl PipelineManager {

// Find all webrtcbin elements in the pipeline
let webrtcbins = self.find_webrtcbin_elements();
info!(
trace!(
"get_webrtc_stats: Found {} webrtcbin element(s)",
webrtcbins.len()
);

for (name, webrtcbin) in webrtcbins {
info!("get_webrtc_stats: Getting stats from webrtcbin: {}", name);
trace!("get_webrtc_stats: Getting stats from webrtcbin: {}", name);

let mut conn_stats = WebRtcConnectionStats::default();

// First check if ICE connection is established - skip if not ready
// This avoids blocking on promise.wait() for webrtcbins that aren't connected
let ice_state = self.get_ice_connection_state(&webrtcbin);
info!("get_webrtc_stats: ICE state for {}: {:?}", name, ice_state);
trace!("get_webrtc_stats: ICE state for {}: {:?}", name, ice_state);

// Only get detailed stats if we have a reasonable ICE state
let should_get_stats = match ice_state.as_deref() {
Expand Down Expand Up @@ -2218,7 +2223,7 @@ impl PipelineManager {
// Returns void, stats come via the promise
let pad_none: Option<&gst::Pad> = None;
let promise = gst::Promise::new();
info!("get_webrtc_stats: Emitting get-stats signal...");
trace!("get_webrtc_stats: Emitting get-stats signal...");
webrtcbin.emit_by_name::<()>("get-stats", &[&pad_none, &promise]);

// Wait for the promise with a timeout using interrupt from another thread
Expand All @@ -2228,55 +2233,55 @@ impl PipelineManager {
promise_clone.interrupt();
});

info!("get_webrtc_stats: Waiting for promise (500ms timeout)...");
trace!("get_webrtc_stats: Waiting for promise (500ms timeout)...");
let promise_result = promise.wait();

// Clean up timeout thread (it will either have interrupted or not)
let _ = timeout_thread.join();

info!("get_webrtc_stats: Promise result: {:?}", promise_result);
trace!("get_webrtc_stats: Promise result: {:?}", promise_result);

match promise_result {
gst::PromiseResult::Replied => {
if let Some(reply) = promise.get_reply() {
// The reply is a GstStructure containing the stats
info!(
trace!(
"get_webrtc_stats: Got reply structure with {} fields: {}",
reply.n_fields(),
reply.name()
);
// Log all field names
for i in 0..reply.n_fields() {
if let Some(field_name) = reply.nth_field_name(i) {
info!("get_webrtc_stats: Field [{}]: {}", i, field_name);
trace!("get_webrtc_stats: Field [{}]: {}", i, field_name);
}
}
// Convert StructureRef to owned Structure for parsing
conn_stats = self.parse_webrtc_stats_structure(&reply.to_owned());
info!(
trace!(
"get_webrtc_stats: Parsed stats - ICE: {:?}, inbound_rtp: {}, outbound_rtp: {}",
conn_stats.ice_candidates.is_some(),
conn_stats.inbound_rtp.len(),
conn_stats.outbound_rtp.len()
);
} else {
info!(
trace!(
"get_webrtc_stats: No stats in promise reply from webrtcbin: {}",
name
);
}
}
gst::PromiseResult::Interrupted => {
warn!(
debug!(
"get_webrtc_stats: Promise timed out (interrupted) for webrtcbin: {}",
name
);
}
gst::PromiseResult::Expired => {
info!("get_webrtc_stats: Promise expired for webrtcbin: {}", name);
trace!("get_webrtc_stats: Promise expired for webrtcbin: {}", name);
}
gst::PromiseResult::Pending => {
info!(
trace!(
"get_webrtc_stats: Promise still pending for webrtcbin: {}",
name
);
Expand All @@ -2303,7 +2308,7 @@ impl PipelineManager {
stats.connections.insert(name, conn_stats);
}

info!(
trace!(
"get_webrtc_stats: Returning stats with {} connection(s)",
stats.connections.len()
);
Expand Down Expand Up @@ -2404,21 +2409,21 @@ impl PipelineManager {

let mut conn_stats = WebRtcConnectionStats::default();

info!(
trace!(
"parse_webrtc_stats_structure: Parsing structure '{}' with {} fields",
structure.name(),
structure.n_fields()
);

// Log ALL field names and their types for debugging
info!("=== RAW WEBRTC STATS STRUCTURE ===");
trace!("=== RAW WEBRTC STATS STRUCTURE ===");
for (field_name, value) in structure.iter() {
let type_name = value.type_().name();
info!(" Field: '{}' (type: {})", field_name, type_name);
trace!(" Field: '{}' (type: {})", field_name, type_name);

// If it's a nested structure, log its contents too
if let Ok(nested) = value.get::<gst::Structure>() {
info!(" Nested structure '{}' fields:", nested.name());
trace!(" Nested structure '{}' fields:", nested.name());
for (nested_field, nested_value) in nested.iter() {
let nested_type = nested_value.type_().name();
// Try to get the actual value for common types
Expand All @@ -2441,11 +2446,11 @@ impl PipelineManager {
} else {
format!("<{}>", nested_type)
};
info!(" {}: {} = {}", nested_field, nested_type, value_str);
trace!(" {}: {} = {}", nested_field, nested_type, value_str);
}
}
}
info!("=== END RAW STATS ===");
trace!("=== END RAW STATS ===");

// WebRTC stats structure contains nested structures for each stat type
// The field NAME indicates the type (e.g., "rtp-inbound-stream-stats_1234")
Expand Down
4 changes: 2 additions & 2 deletions backend/src/stats/collector.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
use strom_types::block::BlockInstance;
use strom_types::stats::{BlockStats, FlowStats, Statistic};
use strom_types::Flow;
use tracing::{debug, info, warn};
use tracing::{debug, trace, warn};

/// Collector for pipeline statistics.
pub struct StatsCollector;
Expand Down Expand Up @@ -90,7 +90,7 @@ impl StatsCollector {
if let Ok(bin) = sdpdemux.dynamic_cast::<gst::Bin>() {
let jb_stats = collect_all_jitterbuffer_stats(&bin);
let jb_count = jb_stats.len();
info!("Found {} jitterbuffer(s) in {}", jb_count, sdpdemux_name);
trace!("Found {} jitterbuffer(s) in {}", jb_count, sdpdemux_name);

for (jb_name, stats) in jb_stats {
debug!("Jitterbuffer '{}' stats: {:?}", jb_name, stats);
Expand Down
Loading