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

[feat](trx-frontend-http): per-session SSE rig selection

Browse Source 
Add SessionRigManager to track per-SSE-session rig_id so different
browser tabs can independently select rigs without interfering.
The /events SSE stream filters state updates by session rig (falling
back to the global active rig), and /select_rig accepts an optional
session_id to update the per-session mapping.

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-21 23:38:24 +01:00
parent dffaed6216
commit ea362c9cdd
3 changed files with 76 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/trx-client/trx-frontend/trx-frontend-http/assets/web/app.js
+8 -1
View File
@@ -599,6 +599,7 @@ let sdrSquelchSupported = false;
let lastRigIds = [];
let lastRigDisplayNames = {};
let lastActiveRigId = null;
let sseSessionId = null;
const originalTitle = document.title;
const savedTheme = loadSetting("theme", null);
@@ -3176,6 +3177,7 @@ async function pollFreshSnapshot() {
function connect() {
if (es) {
es.close();
sseSessionId = null;
}
if (esHeartbeat) {
clearInterval(esHeartbeat);
@@ -3210,6 +3212,10 @@ function connect() {
lastEventAt = Date.now();
});
es.addEventListener("session", evt => {
try {
const d = JSON.parse(evt.data);
sseSessionId = d.session_id || null;
} catch (_) {}
if (typeof vchanHandleSession === "function") vchanHandleSession(evt.data);
});
es.addEventListener("channels", evt => {
@@ -3361,7 +3367,8 @@ async function switchRigFromSelect(selectEl) {
// follow. Commands already carry rig_id per-tab, but SSE is still
// global until per-session streams are implemented.
try {
await postPath(`/select_rig?rig_id=${encodeURIComponent(selectEl.value)}`);
const sidParam = sseSessionId ? `&session_id=${encodeURIComponent(sseSessionId)}` : "";
await postPath(`/select_rig?rig_id=${encodeURIComponent(selectEl.value)}${sidParam}`);
} catch (err) {
console.error("select_rig failed:", err);
}
src/trx-client/trx-frontend/trx-frontend-http/src/api.rs
+62 -1
View File
@@ -2,6 +2,7 @@
//
// SPDX-License-Identifier: BSD-2-Clause
use std::collections::HashMap;
use std::io::Write;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -97,6 +98,43 @@ struct FrontendMeta {
server_connected: bool,
}
/// Tracks per-SSE-session rig selection so different browser tabs can
/// independently view different rigs without interfering.
#[derive(Default)]
pub struct SessionRigManager {
/// Maps SSE session UUID → selected rig_id.
sessions: std::sync::RwLock<HashMap<Uuid, String>>,
}
impl SessionRigManager {
pub fn register(&self, session_id: Uuid, rig_id: String) {
if let Ok(mut sessions) = self.sessions.write() {
sessions.insert(session_id, rig_id);
}
}
pub fn unregister(&self, session_id: Uuid) {
if let Ok(mut sessions) = self.sessions.write() {
sessions.remove(&session_id);
}
}
pub fn get_rig(&self, session_id: Uuid) -> Option<String> {
self.sessions
.read()
.ok()
.and_then(|sessions| sessions.get(&session_id).cloned())
}
pub fn set_rig(&self, session_id: Uuid, rig_id: String) {
if let Ok(mut sessions) = self.sessions.write() {
sessions.insert(session_id, rig_id);
}
}
}
pub type SharedSessionRigManager = Arc<SessionRigManager>;
#[get("/status")]
pub async fn status_api(
state: web::Data<watch::Receiver<RigState>>,
@@ -286,6 +324,7 @@ fn decode_history_retention_min_from_context(context: &FrontendRuntimeContext) -
}
#[get("/events")]
#[allow(clippy::too_many_arguments)]
pub async fn events(
state: web::Data<watch::Receiver<RigState>>,
clients: web::Data<Arc<AtomicUsize>>,
@@ -294,6 +333,7 @@ pub async fn events(
bookmark_store: web::Data<Arc<crate::server::bookmarks::BookmarkStore>>,
scheduler_status: web::Data<crate::server::scheduler::SchedulerStatusMap>,
scheduler_control: web::Data<crate::server::scheduler::SharedSchedulerControlManager>,
session_rig_mgr: web::Data<Arc<SessionRigManager>>,
) -> Result<HttpResponse, Error> {
let rx = state.get_ref().clone();
let initial = wait_for_view(rx.clone()).await?;
@@ -313,6 +353,7 @@ pub async fn events(
.ok()
.and_then(|g| g.clone());
if let Some(ref rid) = active_rig_id {
session_rig_mgr.register(session_id, rid.clone());
vchan_mgr.init_rig(
rid,
initial.status.freq.hz,
@@ -357,6 +398,7 @@ pub async fn events(
let bookmark_store_updates = bookmark_store.get_ref().clone();
let scheduler_status_updates = scheduler_status.get_ref().clone();
let scheduler_control_updates = scheduler_control.get_ref().clone();
let session_rig_mgr_updates = session_rig_mgr.get_ref().clone();
let updates = WatchStream::new(rx).filter_map(move |state| {
let counter = counter_updates.clone();
let context = context_updates.clone();
@@ -364,9 +406,17 @@ pub async fn events(
let bookmark_store = bookmark_store_updates.clone();
let scheduler_status = scheduler_status_updates.clone();
let scheduler_control = scheduler_control_updates.clone();
let session_rig_mgr = session_rig_mgr_updates.clone();
async move {
state.snapshot().and_then(|v| {
if let Ok(Some(rig_id)) = context.remote_active_rig_id.lock().map(|g| g.clone()) {
let rig_id_opt = session_rig_mgr.get_rig(session_id).or_else(|| {
context
.remote_active_rig_id
.lock()
.ok()
.and_then(|g| g.clone())
});
if let Some(rig_id) = rig_id_opt {
vchan.update_primary(
&rig_id,
v.status.freq.hz,
@@ -426,12 +476,14 @@ pub async fn events(
let vchan_drop = vchan_mgr.get_ref().clone();
let counter_drop = counter.clone();
let scheduler_control_drop = scheduler_control.get_ref().clone();
let session_rig_mgr_drop = session_rig_mgr.get_ref().clone();
let live = select(select(pings, updates), chan_updates);
let stream = prefix_stream.chain(live);
let stream = DropStream::new(Box::pin(stream), move || {
counter_drop.fetch_sub(1, Ordering::Relaxed);
vchan_drop.release_session(session_id);
scheduler_control_drop.unregister_session(session_id);
session_rig_mgr_drop.unregister(session_id);
});
Ok(HttpResponse::Ok()
@@ -1520,6 +1572,7 @@ pub async fn list_rigs(
#[derive(serde::Deserialize)]
pub struct SelectRigQuery {
pub rig_id: String,
pub session_id: Option<String>,
}
#[post("/select_rig")]
@@ -1527,6 +1580,7 @@ pub async fn select_rig(
query: web::Query<SelectRigQuery>,
context: web::Data<Arc<FrontendRuntimeContext>>,
vchan_mgr: web::Data<Arc<ClientChannelManager>>,
session_rig_mgr: web::Data<Arc<SessionRigManager>>,
) -> Result<HttpResponse, Error> {
let rig_id = query.rig_id.trim();
if rig_id.is_empty() {
@@ -1551,6 +1605,13 @@ pub async fn select_rig(
*active = Some(rig_id.to_string());
}
// Update per-session rig selection if session_id is provided.
if let Some(ref sid) = query.session_id {
if let Ok(uuid) = Uuid::parse_str(sid) {
session_rig_mgr.set_rig(uuid, rig_id.to_string());
}
}
// Broadcast the channel list for the newly selected rig so all SSE
// clients receive the correct virtual channels immediately.
let chans = vchan_mgr.channels(rig_id);
src/trx-client/trx-frontend/trx-frontend-http/src/server.rs
+6 -1
View File
@@ -3,7 +3,7 @@
// SPDX-License-Identifier: BSD-2-Clause
#[path = "api.rs"]
mod api;
pub mod api;
#[path = "audio.rs"]
pub mod audio;
#[path = "auth.rs"]
@@ -90,6 +90,7 @@ 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 session_rig_mgr = Arc::new(api::SessionRigManager::default());
let background_decode_mgr = BackgroundDecodeManager::new(
background_decode_store,
bookmark_store.clone(),
@@ -137,6 +138,7 @@ async fn serve(
scheduler_status,
scheduler_control,
vchan_mgr,
session_rig_mgr,
background_decode_mgr,
)?;
let handle = server.handle();
@@ -162,6 +164,7 @@ fn build_server(
scheduler_status: SchedulerStatusMap,
scheduler_control: Arc<SchedulerControlManager>,
vchan_mgr: Arc<ClientChannelManager>,
session_rig_mgr: Arc<api::SessionRigManager>,
background_decode_mgr: Arc<BackgroundDecodeManager>,
) -> Result<Server, actix_web::Error> {
let state_data = web::Data::new(state_rx);
@@ -176,6 +179,7 @@ fn build_server(
let scheduler_status = web::Data::new(scheduler_status);
let scheduler_control = web::Data::new(scheduler_control);
let vchan_mgr = web::Data::new(vchan_mgr);
let session_rig_mgr = web::Data::new(session_rig_mgr);
let background_decode_mgr = web::Data::new(background_decode_mgr);
// Extract auth config values before moving context
@@ -221,6 +225,7 @@ fn build_server(
.app_data(scheduler_status.clone())
.app_data(scheduler_control.clone())
.app_data(vchan_mgr.clone())
.app_data(session_rig_mgr.clone())
.app_data(background_decode_mgr.clone())
.wrap(Compress::default())
.wrap(