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

[fix](trx-client): preserve per-rig stream info across audio reconnects

Browse Source 
The audio client was clearing per_rig_info_tx to None every time the
TCP connection dropped, even during transient reconnect cycles.  This
caused WebSocket clients subscribing to per-rig audio to stall
indefinitely waiting for stream info that would only arrive after the
next successful reconnect.

Move the None send to the permanent shutdown path only.  The last-known
stream info remains valid for the same rig across reconnects.

Also revert the global info_rx fallback from 6e4c5e3 since the root
cause is now fixed.

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 08:00:13 +01:00
parent 6e4c5e3c72
commit 723e3ad7cb
2 changed files with 13 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-client/src/audio_client.rs
+7 -4
View File
@@ -315,10 +315,9 @@ async fn run_single_rig_audio_client(
}
}
let _ = per_rig_info_tx.send(None);
if is_selected(&selected_rig_id, &rig_id) {
let _ = global_info_tx.send(None);
}
// Do NOT clear per_rig_info_tx here — the last-known stream info
// remains valid for this rig and clearing it would stall WebSocket
// clients that subscribe while the TCP connection is reconnecting.
tokio::select! {
_ = time::sleep(reconnect_delay) => {}
@@ -326,6 +325,10 @@ async fn run_single_rig_audio_client(
match changed {
Ok(()) if *shutdown_rx.borrow() => {
info!("Audio client [{}]: shutting down", rig_id);
let _ = per_rig_info_tx.send(None);
if is_selected(&selected_rig_id, &rig_id) {
let _ = global_info_tx.send(None);
}
return;
}
Ok(()) => {}
src/trx-client/trx-frontend/trx-frontend-http/src/audio.rs
+6 -14
View File
@@ -505,7 +505,6 @@ 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()
@@ -535,9 +534,12 @@ pub async fn audio_ws(
// rig's audio. Wait briefly for the per-rig channel to appear (it is
// lazily created by the audio relay sync task every 500ms).
let deadline = Instant::now() + Duration::from_secs(3);
let rx_sub = loop {
if let Some(rx) = context.rig_audio_subscribe(remote) {
break rx;
let (rx_sub, info_rx) = loop {
if let (Some(rx), Some(info_rx)) = (
context.rig_audio_subscribe(remote),
context.rig_audio_info_rx(remote),
) {
break (rx, info_rx);
}
if Instant::now() >= deadline {
return Ok(
@@ -546,16 +548,6 @@ pub async fn audio_ws(
}
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 {