sjg/trx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[fix](trx-frontend-http): fall back to global stream info for per-rig audio

Browse Source 
Per-rig info_rx watch channel is transiently None when the rig's audio
TCP connection is between reconnect cycles.  This caused the WebSocket
handler to hang indefinitely waiting for stream info that might never
arrive, regressed in 7d76606.

Prefer the per-rig info_rx when it holds a value, otherwise fall back to
the global info channel (the pre-regression behaviour).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Stan Grams <sjg@haxx.space>
This commit is contained in:
Stan Grams
2026-03-24 07:55:09 +01:00
parent cfc8407c3a
commit 6e4c5e3c72
1 changed files with 15 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-client/trx-frontend/trx-frontend-http/src/audio.rs
+15 -20
View File
@@ -19,7 +19,7 @@ use actix_ws::Message;
use bytes::Bytes;
use serde::Deserialize;
use tokio::sync::broadcast;
use tracing::{info, warn};
use tracing::warn;
use uuid::Uuid;
use trx_core::decode::{
@@ -505,6 +505,7 @@ pub async fn audio_ws(
let info_rx = if let Some(ref remote) = query.remote {
context
.rig_audio_info_rx(remote)
.filter(|rx| rx.borrow().is_some())
.or_else(|| context.audio_info.as_ref().cloned())
} else {
context.audio_info.as_ref().cloned()
@@ -533,28 +534,28 @@ pub async fn audio_ws(
// Do NOT fall back to global — that would silently deliver the wrong
// rig's audio. Wait briefly for the per-rig channel to appear (it is
// lazily created by the audio relay sync task every 500ms).
info!("Audio WS: per-rig request for remote={}", remote);
let deadline = Instant::now() + Duration::from_secs(3);
let (rx_sub, info_rx) = loop {
let has_audio = context.rig_audio_subscribe(remote).is_some();
let has_info = context.rig_audio_info_rx(remote).is_some();
info!("Audio WS: remote={} has_audio={} has_info={}", remote, has_audio, has_info);
if let (Some(rx), Some(info_rx)) = (
context.rig_audio_subscribe(remote),
context.rig_audio_info_rx(remote),
) {
let current = info_rx.borrow().clone();
info!("Audio WS: remote={} info_rx current value is_some={}", remote, current.is_some());
break (rx, info_rx);
let rx_sub = loop {
if let Some(rx) = context.rig_audio_subscribe(remote) {
break rx;
}
if Instant::now() >= deadline {
info!("Audio WS: timeout waiting for rig {}", remote);
return Ok(
HttpResponse::NotFound().body(format!("audio not available for rig {remote}"))
);
}
tokio::time::sleep(Duration::from_millis(100)).await;
};
// Prefer per-rig stream info when available; fall back to the global
// info channel so the WebSocket does not stall when the per-rig audio
// TCP connection is between reconnect cycles (value transiently None).
let info_rx = context
.rig_audio_info_rx(remote)
.filter(|rx| rx.borrow().is_some())
.or_else(|| context.audio_info.as_ref().cloned());
let Some(info_rx) = info_rx else {
return Ok(HttpResponse::NotFound().body("audio not enabled"));
};
(rx_sub, info_rx)
} else {
let Some(info_rx) = context.audio_info.as_ref().cloned() else {
@@ -570,15 +571,11 @@ pub async fn audio_ws(
let (response, mut session, mut msg_stream) = actix_ws::handle(&req, body)?;
actix_web::rt::spawn(async move {
info!("Audio WS: session started, waiting for stream info");
let mut current_info = loop {
if let Some(info) = info_rx.borrow().clone() {
info!("Audio WS: got stream info: {}Hz {}ch", info.sample_rate, info.channels);
break info;
}
info!("Audio WS: info_rx is None, waiting for changed()");
if info_rx.changed().await.is_err() {
info!("Audio WS: info_rx sender dropped, closing");
let _ = session.close(None).await;
return;
}
@@ -591,9 +588,7 @@ pub async fn audio_ws(
return;
}
};
info!("Audio WS: sending stream info JSON to browser");
if session.text(info_json).await.is_err() {
info!("Audio WS: failed to send stream info, closing");
return;
}