[feat](trx-client): add per-rig spectrum and audio streams

Each browser tab can now subscribe to a specific rig's spectrum and
audio independently via ?rig_id= query params on /spectrum and /audio.
The remote client polls spectrum for all rigs with active subscribers
and routes responses to per-rig watch channels. Virtual channel
commands are routed through per-rig senders with global fallback.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Stan Grams <sjg@haxx.space>

src/trx-client/trx-frontend/src/lib.rs

@@ -266,6 +266,17 @@ pub struct FrontendRuntimeContext {
     pub ais_vessel_url_base: Option<String>,
     /// Spectrum sender; SSE clients subscribe via `spectrum.subscribe()`.
     pub spectrum: Arc<watch::Sender<SharedSpectrum>>,
+    /// Per-rig spectrum watch channels, keyed by rig_id.
+    /// Populated by the remote client spectrum polling task so each SSE
+    /// session can subscribe to a specific rig's spectrum independently.
+    pub rig_spectrums: Arc<RwLock<HashMap<String, watch::Sender<SharedSpectrum>>>>,
+    /// Per-rig RX audio broadcast senders, keyed by rig_id.
+    /// Each rig's audio client task publishes Opus frames here.
+    pub rig_audio_rx: Arc<RwLock<HashMap<String, broadcast::Sender<Bytes>>>>,
+    /// Per-rig audio stream info watch channels, keyed by rig_id.
+    pub rig_audio_info: Arc<RwLock<HashMap<String, watch::Sender<Option<AudioStreamInfo>>>>>,
+    /// Per-rig virtual-channel command senders, keyed by rig_id.
+    pub rig_vchan_audio_cmd: Arc<RwLock<HashMap<String, mpsc::UnboundedSender<VChanAudioCmd>>>>,
     /// Per-virtual-channel Opus audio senders.
     /// Key: server-side virtual channel UUID.
     /// Value: `broadcast::Sender<Bytes>` that receives per-channel Opus packets
@@ -293,6 +304,44 @@ impl FrontendRuntimeContext {
             .and_then(|map| map.get(rig_id).map(|tx| tx.subscribe()))
     }
 
+    /// Get a watch receiver for a specific rig's spectrum.
+    /// Lazily inserts a new channel if the rig_id is not yet present.
+    pub fn rig_spectrum_rx(&self, rig_id: &str) -> watch::Receiver<SharedSpectrum> {
+        if let Ok(map) = self.rig_spectrums.read() {
+            if let Some(tx) = map.get(rig_id) {
+                return tx.subscribe();
+            }
+        }
+        // Insert on miss.
+        if let Ok(mut map) = self.rig_spectrums.write() {
+            map.entry(rig_id.to_string())
+                .or_insert_with(|| watch::channel(SharedSpectrum::default()).0)
+                .subscribe()
+        } else {
+            // Poisoned lock fallback: return a dummy receiver.
+            watch::channel(SharedSpectrum::default()).1
+        }
+    }
+
+    /// Subscribe to a specific rig's RX audio broadcast.
+    pub fn rig_audio_subscribe(&self, rig_id: &str) -> Option<broadcast::Receiver<Bytes>> {
+        self.rig_audio_rx
+            .read()
+            .ok()
+            .and_then(|map| map.get(rig_id).map(|tx| tx.subscribe()))
+    }
+
+    /// Get a watch receiver for a specific rig's audio stream info.
+    pub fn rig_audio_info_rx(
+        &self,
+        rig_id: &str,
+    ) -> Option<watch::Receiver<Option<AudioStreamInfo>>> {
+        self.rig_audio_info
+            .read()
+            .ok()
+            .and_then(|map| map.get(rig_id).map(|tx| tx.subscribe()))
+    }
+
     /// Create a new empty runtime context.
     pub fn new() -> Self {
         Self {
@@ -338,6 +387,10 @@ impl FrontendRuntimeContext {
                 let (tx, _rx) = watch::channel(SharedSpectrum::default());
                 Arc::new(tx)
             },
+            rig_spectrums: Arc::new(RwLock::new(HashMap::new())),
+            rig_audio_rx: Arc::new(RwLock::new(HashMap::new())),
+            rig_audio_info: Arc::new(RwLock::new(HashMap::new())),
+            rig_vchan_audio_cmd: Arc::new(RwLock::new(HashMap::new())),
             vchan_audio: Arc::new(RwLock::new(HashMap::new())),
             vchan_audio_cmd: Arc::new(Mutex::new(None)),
             vchan_destroyed: None,

src/trx-client/trx-frontend/trx-frontend-http/assets/web/app.js

@@ -3378,6 +3378,14 @@ async function switchRigFromSelect(selectEl) {
   } catch (err) {
     console.error("select_rig failed:", err);
   }
+  // Reconnect spectrum SSE to the new rig's spectrum channel.
+  stopSpectrumStreaming();
+  startSpectrumStreaming();
+  // Reconnect audio to the new rig if audio is active.
+  if (rxActive) {
+    stopRxAudio();
+    startRxAudio();
+  }
   showHint(`Rig: ${lastActiveRigId}`, 1500);
 }
 
@@ -7497,9 +7505,14 @@ function startRxAudio() {
     return;
   }
   const proto = location.protocol === "https:" ? "wss:" : "ws:";
-  const audioPath = _audioChannelOverride
-    ? `/audio?channel_id=${encodeURIComponent(_audioChannelOverride)}`
-    : "/audio";
+  let audioPath;
+  if (_audioChannelOverride) {
+    audioPath = `/audio?channel_id=${encodeURIComponent(_audioChannelOverride)}`;
+  } else if (lastActiveRigId) {
+    audioPath = `/audio?rig_id=${encodeURIComponent(lastActiveRigId)}`;
+  } else {
+    audioPath = "/audio";
+  }
   audioWs = new WebSocket(`${proto}//${location.host}${audioPath}`);
   audioWs.binaryType = "arraybuffer";
   audioStatus.textContent = "Connecting…";
@@ -8533,7 +8546,10 @@ function scheduleSpectrumReconnect() {
 
 function startSpectrumStreaming() {
   if (spectrumSource !== null) return;
-  spectrumSource = new EventSource("/spectrum");
+  const spectrumUrl = lastActiveRigId
+    ? `/spectrum?rig_id=${encodeURIComponent(lastActiveRigId)}`
+    : "/spectrum";
+  spectrumSource = new EventSource(spectrumUrl);
   // Unnamed event = reset signal.
   spectrumSource.onmessage = (evt) => {
     if (evt.data === "null") {

src/trx-client/trx-frontend/trx-frontend-http/src/api.rs

@@ -365,17 +365,13 @@ pub async fn events(
     // Use the client-requested rig_id if provided, otherwise fall back to
     // the global default.  This allows each tab to reconnect SSE for the
     // rig it has selected without mutating global state.
-    let active_rig_id = query
-        .rig_id
-        .clone()
-        .filter(|s| !s.is_empty())
-        .or_else(|| {
-            context
-                .remote_active_rig_id
-                .lock()
-                .ok()
-                .and_then(|g| g.clone())
-        });
+    let active_rig_id = query.rig_id.clone().filter(|s| !s.is_empty()).or_else(|| {
+        context
+            .remote_active_rig_id
+            .lock()
+            .ok()
+            .and_then(|g| g.clone())
+    });
 
     // Subscribe to the per-rig watch channel for this session's rig,
     // falling back to the global state watch when unavailable.
@@ -804,11 +800,16 @@ impl<I> futures_util::Stream for DropStream<I> {
 /// Emits an unnamed `data: null` event when spectrum data becomes unavailable.
 #[get("/spectrum")]
 pub async fn spectrum(
+    query: web::Query<RigIdQuery>,
     context: web::Data<Arc<FrontendRuntimeContext>>,
 ) -> Result<HttpResponse, Error> {
-    // Subscribe to the watch channel: each client gets its own receiver and is
-    // woken exactly when new spectrum data is pushed (no 40 ms polling needed).
-    let rx = context.spectrum.subscribe();
+    // Subscribe to a per-rig spectrum channel when rig_id is specified,
+    // otherwise fall back to the global channel for backward compat.
+    let rx = if let Some(ref rig_id) = query.rig_id {
+        context.rig_spectrum_rx(rig_id)
+    } else {
+        context.spectrum.subscribe()
+    };
     let mut last_rds_json: Option<String> = None;
     let mut last_vchan_rds_json: Option<String> = None;
     let mut last_had_frame = false;

src/trx-client/trx-frontend/trx-frontend-http/src/audio.rs

@@ -475,6 +475,7 @@ pub fn start_decode_history_collector(context: Arc<FrontendRuntimeContext>) {
 #[derive(Deserialize)]
 pub struct AudioQuery {
     pub channel_id: Option<Uuid>,
+    pub rig_id: Option<String>,
 }
 
 #[get("/audio")]
@@ -516,6 +517,18 @@ pub async fn audio_ws(
             }
             tokio::time::sleep(Duration::from_millis(50)).await;
         }
+    } else if let Some(ref rig_id) = query.rig_id {
+        // Per-rig audio: subscribe to the specific rig's broadcast.
+        match context.rig_audio_subscribe(rig_id) {
+            Some(rx) => rx,
+            None => {
+                // Rig not yet connected; fall back to global.
+                let Some(rx) = context.audio_rx.as_ref() else {
+                    return Ok(HttpResponse::NotFound().body("audio not enabled"));
+                };
+                rx.subscribe()
+            }
+        }
     } else {
         let Some(rx) = context.audio_rx.as_ref() else {
             return Ok(HttpResponse::NotFound().body("audio not enabled"));
@@ -524,6 +537,13 @@ pub async fn audio_ws(
     };
     let mut rx_sub = rx_sub;
 
+    // Use per-rig audio info if available and rig_id was specified.
+    if let Some(ref rig_id) = query.rig_id {
+        if let Some(rig_info_rx) = context.rig_audio_info_rx(rig_id) {
+            info_rx = rig_info_rx;
+        }
+    }
+
     let (response, mut session, mut msg_stream) = actix_ws::handle(&req, body)?;
 
     actix_web::rt::spawn(async move {

src/trx-client/trx-frontend/trx-frontend-http/src/background_decode.rs

@@ -250,6 +250,16 @@ impl BackgroundDecodeManager {
     }
 
     fn send_audio_cmd(&self, cmd: VChanAudioCmd) {
+        // Route through per-rig sender when available.
+        if let Some(rig_id) = self.active_rig_id() {
+            if let Ok(map) = self.context.rig_vchan_audio_cmd.read() {
+                if let Some(tx) = map.get(&rig_id) {
+                    let _ = tx.send(cmd);
+                    return;
+                }
+            }
+        }
+        // Fall back to global sender.
         if let Ok(guard) = self.context.vchan_audio_cmd.lock() {
             if let Some(tx) = guard.as_ref() {
                 let _ = tx.send(cmd);

src/trx-client/trx-frontend/trx-frontend-http/src/server.rs

@@ -89,7 +89,10 @@ async fn serve(
 
     let background_decode_path = BackgroundDecodeStore::default_path();
     let background_decode_store = Arc::new(BackgroundDecodeStore::open(&background_decode_path));
-    let vchan_mgr = Arc::new(ClientChannelManager::new(4));
+    let vchan_mgr = Arc::new(ClientChannelManager::new(
+        4,
+        context.rig_vchan_audio_cmd.clone(),
+    ));
     let session_rig_mgr = Arc::new(api::SessionRigManager::default());
     let background_decode_mgr = BackgroundDecodeManager::new(
         background_decode_store,

src/trx-client/trx-frontend/trx-frontend-http/src/vchan.rs

@@ -16,10 +16,10 @@
 //! tunes a channel.
 
 use std::collections::HashMap;
-use std::sync::RwLock;
+use std::sync::{Arc, RwLock};
 
 use serde::{Deserialize, Serialize};
-use tokio::sync::broadcast;
+use tokio::sync::{broadcast, mpsc};
 use uuid::Uuid;
 
 use trx_frontend::VChanAudioCmd;
@@ -107,12 +107,18 @@ pub struct ClientChannelManager {
     /// `"<rig_id>:"` so subscribers can filter by rig.
     pub change_tx: broadcast::Sender<String>,
     pub max_channels: usize,
-    /// Optional sender to the audio-client task for virtual-channel audio commands.
-    pub audio_cmd: std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedSender<VChanAudioCmd>>>,
+    /// Global fallback sender to the audio-client task for virtual-channel audio commands.
+    pub audio_cmd: std::sync::Mutex<Option<mpsc::UnboundedSender<VChanAudioCmd>>>,
+    /// Per-rig vchan command senders. Commands are routed to the per-rig sender
+    /// when available, falling back to the global `audio_cmd`.
+    pub rig_vchan_audio_cmd: Arc<RwLock<HashMap<String, mpsc::UnboundedSender<VChanAudioCmd>>>>,
 }
 
 impl ClientChannelManager {
-    pub fn new(max_channels: usize) -> Self {
+    pub fn new(
+        max_channels: usize,
+        rig_vchan_audio_cmd: Arc<RwLock<HashMap<String, mpsc::UnboundedSender<VChanAudioCmd>>>>,
+    ) -> Self {
         let (change_tx, _) = broadcast::channel(64);
         Self {
             rigs: RwLock::new(HashMap::new()),
@@ -120,17 +126,26 @@ impl ClientChannelManager {
             change_tx,
             max_channels: max_channels.max(1),
             audio_cmd: std::sync::Mutex::new(None),
+            rig_vchan_audio_cmd,
         }
     }
 
-    /// Wire the audio-command sender so the manager can dispatch
-    /// `VChanAudioCmd` messages when channels are allocated/deleted/changed.
-    pub fn set_audio_cmd(&self, tx: tokio::sync::mpsc::UnboundedSender<VChanAudioCmd>) {
+    /// Wire the global audio-command sender as fallback.
+    pub fn set_audio_cmd(&self, tx: mpsc::UnboundedSender<VChanAudioCmd>) {
         *self.audio_cmd.lock().unwrap() = Some(tx);
     }
 
-    /// Fire-and-forget: send a `VChanAudioCmd` to the audio-client task.
-    fn send_audio_cmd(&self, cmd: VChanAudioCmd) {
+    /// Fire-and-forget: send a `VChanAudioCmd`, routing to the per-rig sender
+    /// when available or falling back to the global sender.
+    fn send_audio_cmd_for_rig(&self, rig_id: &str, cmd: VChanAudioCmd) {
+        // Try per-rig sender first.
+        if let Ok(map) = self.rig_vchan_audio_cmd.read() {
+            if let Some(tx) = map.get(rig_id) {
+                let _ = tx.send(cmd);
+                return;
+            }
+        }
+        // Fall back to global sender.
         if let Some(tx) = self.audio_cmd.lock().unwrap().as_ref() {
             let _ = tx.send(cmd);
         }
@@ -265,13 +280,16 @@ impl ClientChannelManager {
             .insert(session_id, (rig_id.to_string(), id));
 
         // Request server-side DSP channel + audio subscription.
-        self.send_audio_cmd(VChanAudioCmd::Subscribe {
-            uuid: id,
-            freq_hz,
-            mode: mode.to_string(),
-            bandwidth_hz: 0,
-            decoder_kinds: Vec::new(),
-        });
+        self.send_audio_cmd_for_rig(
+            rig_id,
+            VChanAudioCmd::Subscribe {
+                uuid: id,
+                freq_hz,
+                mode: mode.to_string(),
+                bandwidth_hz: 0,
+                decoder_kinds: Vec::new(),
+            },
+        );
 
         Ok(snapshot)
     }
@@ -362,7 +380,7 @@ impl ClientChannelManager {
         drop(rigs);
 
         for channel_id in removed_channel_ids {
-            self.send_audio_cmd(VChanAudioCmd::Remove(channel_id));
+            self.send_audio_cmd_for_rig(rig_id, VChanAudioCmd::Remove(channel_id));
         }
     }
 
@@ -389,7 +407,7 @@ impl ClientChannelManager {
         }
 
         // Remove server-side DSP channel and stop audio encoding.
-        self.send_audio_cmd(VChanAudioCmd::Remove(channel_id));
+        self.send_audio_cmd_for_rig(rig_id, VChanAudioCmd::Remove(channel_id));
 
         Ok(())
     }
@@ -446,10 +464,13 @@ impl ClientChannelManager {
         ch.freq_hz = freq_hz;
         self.broadcast_change(rig_id, channels);
         drop(rigs);
-        self.send_audio_cmd(VChanAudioCmd::SetFreq {
-            uuid: channel_id,
-            freq_hz,
-        });
+        self.send_audio_cmd_for_rig(
+            rig_id,
+            VChanAudioCmd::SetFreq {
+                uuid: channel_id,
+                freq_hz,
+            },
+        );
         Ok(())
     }
 
@@ -468,10 +489,13 @@ impl ClientChannelManager {
         ch.mode = mode.to_string();
         self.broadcast_change(rig_id, channels);
         drop(rigs);
-        self.send_audio_cmd(VChanAudioCmd::SetMode {
-            uuid: channel_id,
-            mode: mode.to_string(),
-        });
+        self.send_audio_cmd_for_rig(
+            rig_id,
+            VChanAudioCmd::SetMode {
+                uuid: channel_id,
+                mode: mode.to_string(),
+            },
+        );
         Ok(())
     }
 
@@ -490,10 +514,13 @@ impl ClientChannelManager {
         ch.bandwidth_hz = bandwidth_hz;
         self.broadcast_change(rig_id, channels);
         drop(rigs);
-        self.send_audio_cmd(VChanAudioCmd::SetBandwidth {
-            uuid: channel_id,
-            bandwidth_hz,
-        });
+        self.send_audio_cmd_for_rig(
+            rig_id,
+            VChanAudioCmd::SetBandwidth {
+                uuid: channel_id,
+                bandwidth_hz,
+            },
+        );
         Ok(())
     }
 
@@ -557,7 +584,7 @@ impl ClientChannelManager {
             if remove {
                 let channel_id = channels[idx].id;
                 channels.remove(idx);
-                self.send_audio_cmd(VChanAudioCmd::Remove(channel_id));
+                self.send_audio_cmd_for_rig(rig_id, VChanAudioCmd::Remove(channel_id));
                 changed = true;
                 continue;
             }
@@ -574,37 +601,49 @@ impl ClientChannelManager {
             };
             if channel.freq_hz != *freq_hz {
                 channel.freq_hz = *freq_hz;
-                self.send_audio_cmd(VChanAudioCmd::SetFreq {
-                    uuid: channel.id,
-                    freq_hz: *freq_hz,
-                });
+                self.send_audio_cmd_for_rig(
+                    rig_id,
+                    VChanAudioCmd::SetFreq {
+                        uuid: channel.id,
+                        freq_hz: *freq_hz,
+                    },
+                );
                 changed = true;
             }
             if channel.mode != *mode {
                 channel.mode = mode.clone();
-                self.send_audio_cmd(VChanAudioCmd::SetMode {
-                    uuid: channel.id,
-                    mode: mode.clone(),
-                });
+                self.send_audio_cmd_for_rig(
+                    rig_id,
+                    VChanAudioCmd::SetMode {
+                        uuid: channel.id,
+                        mode: mode.clone(),
+                    },
+                );
                 changed = true;
             }
             if channel.bandwidth_hz != *bandwidth_hz {
                 channel.bandwidth_hz = *bandwidth_hz;
-                self.send_audio_cmd(VChanAudioCmd::SetBandwidth {
-                    uuid: channel.id,
-                    bandwidth_hz: *bandwidth_hz,
-                });
+                self.send_audio_cmd_for_rig(
+                    rig_id,
+                    VChanAudioCmd::SetBandwidth {
+                        uuid: channel.id,
+                        bandwidth_hz: *bandwidth_hz,
+                    },
+                );
                 changed = true;
             }
             if channel.decoder_kinds != *decoder_kinds {
                 channel.decoder_kinds = decoder_kinds.clone();
-                self.send_audio_cmd(VChanAudioCmd::Subscribe {
-                    uuid: channel.id,
-                    freq_hz: channel.freq_hz,
-                    mode: channel.mode.clone(),
-                    bandwidth_hz: channel.bandwidth_hz,
-                    decoder_kinds: channel.decoder_kinds.clone(),
-                });
+                self.send_audio_cmd_for_rig(
+                    rig_id,
+                    VChanAudioCmd::Subscribe {
+                        uuid: channel.id,
+                        freq_hz: channel.freq_hz,
+                        mode: channel.mode.clone(),
+                        bandwidth_hz: channel.bandwidth_hz,
+                        decoder_kinds: channel.decoder_kinds.clone(),
+                    },
+                );
                 changed = true;
             }
         }
@@ -630,13 +669,16 @@ impl ClientChannelManager {
                 scheduler_bookmark_id: Some(bookmark_id.clone()),
                 session_ids: Vec::new(),
             });
-            self.send_audio_cmd(VChanAudioCmd::Subscribe {
-                uuid: channel_id,
-                freq_hz: *freq_hz,
-                mode: mode.clone(),
-                bandwidth_hz: *bandwidth_hz,
-                decoder_kinds: decoder_kinds.clone(),
-            });
+            self.send_audio_cmd_for_rig(
+                rig_id,
+                VChanAudioCmd::Subscribe {
+                    uuid: channel_id,
+                    freq_hz: *freq_hz,
+                    mode: mode.clone(),
+                    bandwidth_hz: *bandwidth_hz,
+                    decoder_kinds: decoder_kinds.clone(),
+                },
+            );
             changed = true;
         }
 
@@ -652,7 +694,7 @@ mod tests {
 
     #[test]
     fn release_session_removes_last_non_permanent_channel() {
-        let mgr = ClientChannelManager::new(4);
+        let mgr = ClientChannelManager::new(4, Arc::new(RwLock::new(HashMap::new())));
         let rig_id = "rig-a";
         let session_id = Uuid::new_v4();
 
@@ -673,7 +715,7 @@ mod tests {
 
     #[test]
     fn sync_scheduler_channels_materializes_visible_scheduler_channels() {
-        let mgr = ClientChannelManager::new(4);
+        let mgr = ClientChannelManager::new(4, Arc::new(RwLock::new(HashMap::new())));
         let rig_id = "rig-a";
 
         mgr.init_rig(rig_id, 14_074_000, "USB");
@@ -699,7 +741,7 @@ mod tests {
 
     #[test]
     fn release_session_keeps_scheduler_managed_channels() {
-        let mgr = ClientChannelManager::new(4);
+        let mgr = ClientChannelManager::new(4, Arc::new(RwLock::new(HashMap::new())));
         let rig_id = "rig-a";
         let session_id = Uuid::new_v4();
 
@@ -728,7 +770,7 @@ mod tests {
 
     #[test]
     fn subscribed_scheduler_channel_survives_scheduler_clear_until_released() {
-        let mgr = ClientChannelManager::new(4);
+        let mgr = ClientChannelManager::new(4, Arc::new(RwLock::new(HashMap::new())));
         let rig_id = "rig-a";
         let session_id = Uuid::new_v4();