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

Fix and harden client/server configuration #48

Merged
sjg merged 13 commits from feat/config-improvements into main 2026-08-06 22:22:40 +02:00
Conversation 0 Commits 13 Files Changed 31
+283 -137
3 changed files with 283 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 88ed3da6cc - Show all commits
src/trx-config/src/server.rs
+134
View File
@@ -21,6 +21,56 @@ pub use trx_decode_log::DecodeLogsConfig;
use trx_core::rig::state::RigMode; use trx_core::rig::state::RigMode;
/// Every decoder the server knows how to run, by config name.
///
/// The same names are used by `[sdr.channels].decoders`, so there is one
/// vocabulary for both.
pub const DECODER_NAMES: &[&str] = &[
"aprs", "aprs_hf", "ais", "cw", "ft2", "ft4", "ft8", "lrpt", "sstv", "vdes", "wefax", "wspr",
];
/// Which decoders run for a rig, and where the ones that write files put them.
///
/// Decoders used to be started unconditionally: every rig ran ten of them
/// whether or not the operator ever looked at the results, which is real CPU on
/// a Pi. `enabled` defaults to all of them so upgrading changes nothing.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct DecodersConfig {
/// Decoder names to run; see `DECODER_NAMES`.
pub enabled: Vec<String>,
/// Base directory for decoders that write images (sstv, wefax, lrpt).
/// Each decoder gets a subdirectory. Defaults to the user cache directory.
pub output_dir: Option<String>,
}
impl Default for DecodersConfig {
fn default() -> Self {
Self {
enabled: DECODER_NAMES.iter().map(|s| s.to_string()).collect(),
output_dir: None,
}
}
}
impl DecodersConfig {
/// Whether `name` should be started for this rig.
pub fn is_enabled(&self, name: &str) -> bool {
self.enabled.iter().any(|n| n == name)
}
/// Where `decoder` should write its files.
pub fn output_dir_for(&self, decoder: &str) -> std::path::PathBuf {
let base = match &self.output_dir {
Some(dir) => std::path::PathBuf::from(dir),
None => dirs::cache_dir()
.unwrap_or_else(|| std::path::PathBuf::from(".cache"))
.join("trx-rs"),
};
base.join(decoder)
}
}
/// Per-rig instance configuration for multi-rig setups. /// Per-rig instance configuration for multi-rig setups.
/// ///
/// Each entry in `[[rigs]]` becomes one of these. The flat top-level /// Each entry in `[[rigs]]` becomes one of these. The flat top-level
@@ -53,6 +103,8 @@ pub struct RigInstanceConfig {
pub aprsfi: AprsFiConfig, pub aprsfi: AprsFiConfig,
/// Decoder file logging for this rig. /// Decoder file logging for this rig.
pub decode_logs: DecodeLogsConfig, pub decode_logs: DecodeLogsConfig,
/// Which decoders to run for this rig.
pub decoders: DecodersConfig,
} }
impl Default for RigInstanceConfig { impl Default for RigInstanceConfig {
@@ -68,6 +120,7 @@ impl Default for RigInstanceConfig {
pskreporter: PskReporterConfig::default(), pskreporter: PskReporterConfig::default(),
aprsfi: AprsFiConfig::default(), aprsfi: AprsFiConfig::default(),
decode_logs: DecodeLogsConfig::default(), decode_logs: DecodeLogsConfig::default(),
decoders: DecodersConfig::default(),
} }
} }
} }
@@ -100,6 +153,8 @@ pub struct ServerConfig {
pub aprsfi: AprsFiConfig, pub aprsfi: AprsFiConfig,
/// Decoder file logging configuration (legacy flat) /// Decoder file logging configuration (legacy flat)
pub decode_logs: DecodeLogsConfig, pub decode_logs: DecodeLogsConfig,
/// Decoder selection (legacy flat)
pub decoders: DecodersConfig,
/// SDR pipeline configuration (legacy flat; used when [rig.access] type = "sdr"). /// SDR pipeline configuration (legacy flat; used when [rig.access] type = "sdr").
pub sdr: SdrConfig, pub sdr: SdrConfig,
/// Timeout and buffer-size tuning knobs. /// Timeout and buffer-size tuning knobs.
@@ -642,6 +697,7 @@ impl ServerConfig {
pskreporter: self.pskreporter.clone(), pskreporter: self.pskreporter.clone(),
aprsfi: self.aprsfi.clone(), aprsfi: self.aprsfi.clone(),
decode_logs: self.decode_logs.clone(), decode_logs: self.decode_logs.clone(),
decoders: self.decoders.clone(),
}] }]
} }
@@ -679,6 +735,7 @@ impl ServerConfig {
pskreporter: PskReporterConfig::default(), pskreporter: PskReporterConfig::default(),
aprsfi: AprsFiConfig::default(), aprsfi: AprsFiConfig::default(),
decode_logs: DecodeLogsConfig::default(), decode_logs: DecodeLogsConfig::default(),
decoders: DecodersConfig::default(),
sdr: SdrConfig::default(), sdr: SdrConfig::default(),
timeouts: TimeoutsConfig::default(), timeouts: TimeoutsConfig::default(),
rigs: Vec::new(), rigs: Vec::new(),
@@ -821,6 +878,16 @@ fn validate_rig_instance(
errors.push(e); errors.push(e);
} }
for name in &rig.decoders.enabled {
if !DECODER_NAMES.contains(&name.as_str()) {
errors.push(format!(
"{prefix}[decoders].enabled contains unknown decoder \"{}\" (valid: {})",
name,
DECODER_NAMES.join(", ")
));
}
}
if rig.decode_logs.enabled { if rig.decode_logs.enabled {
if rig.decode_logs.dir.trim().is_empty() { if rig.decode_logs.dir.trim().is_empty() {
errors.push(format!( errors.push(format!(
@@ -1650,6 +1717,73 @@ port = 4531
); );
} }
// --- Decoder selection ---
#[test]
fn test_decoders_default_enables_everything() {
let cfg = DecodersConfig::default();
for name in DECODER_NAMES {
assert!(cfg.is_enabled(name), "{name} should be on by default");
}
}
#[test]
fn test_decoders_parsed_from_rig_entry() {
let toml_str = r#"
[[rigs]]
id = "hf"
[rigs.rig]
model = "ft817"
[rigs.rig.access]
type = "serial"
port = "/dev/ttyUSB0"
baud = 9600
[rigs.audio]
port = 4531
[rigs.decoders]
enabled = ["ft8", "wspr"]
output_dir = "/var/lib/trx-rs"
"#;
let cfg: ServerConfig = toml::from_str(toml_str).unwrap();
cfg.validate().expect("config should be valid");
let rigs = cfg.resolved_rigs();
assert!(rigs[0].decoders.is_enabled("ft8"));
assert!(!rigs[0].decoders.is_enabled("cw"));
assert_eq!(
rigs[0].decoders.output_dir_for("sstv"),
std::path::PathBuf::from("/var/lib/trx-rs/sstv")
);
}
#[test]
fn test_decoders_flat_layout_reaches_resolved_rig() {
let toml_str = r#"
[decoders]
enabled = ["cw"]
"#;
let cfg: ServerConfig = toml::from_str(toml_str).unwrap();
let rigs = cfg.resolved_rigs();
assert!(rigs[0].decoders.is_enabled("cw"));
assert!(!rigs[0].decoders.is_enabled("ft8"));
}
#[test]
fn test_validate_rejects_unknown_decoder_name() {
let toml_str = r#"
[decoders]
enabled = ["ft8", "morse"]
"#;
let cfg: ServerConfig = toml::from_str(toml_str).unwrap();
let err = cfg.validate().expect_err("expected an unknown decoder error");
assert!(err.contains("morse"), "unexpected error: {err}");
}
#[test]
fn test_decoders_output_dir_defaults_under_cache() {
let cfg = DecodersConfig::default();
assert!(cfg.output_dir_for("sstv").ends_with("trx-rs/sstv"));
}
// --- Per-rig validation: [[rigs]] entries obey the same rules as the // --- Per-rig validation: [[rigs]] entries obey the same rules as the
// legacy flat layout, which they previously escaped entirely. --- // legacy flat layout, which they previously escaped entirely. ---
src/trx-server/src/audio.rs
+1 -4
View File
@@ -2291,15 +2291,12 @@ pub async fn run_wefax_decoder(
mut state_rx: watch::Receiver<RigState>, mut state_rx: watch::Receiver<RigState>,
decode_tx: broadcast::Sender<DecodedMessage>, decode_tx: broadcast::Sender<DecodedMessage>,
histories: Arc<DecoderHistories>, histories: Arc<DecoderHistories>,
wefax_output_dir: std::path::PathBuf,
) { ) {
use trx_wefax::{WefaxConfig, WefaxDecoder, WefaxEvent}; use trx_wefax::{WefaxConfig, WefaxDecoder, WefaxEvent};
info!("WEFAX decoder started ({}Hz, {} ch)", sample_rate, channels); info!("WEFAX decoder started ({}Hz, {} ch)", sample_rate, channels);
let wefax_output_dir = dirs::cache_dir()
.unwrap_or_else(|| std::path::PathBuf::from(".cache"))
.join("trx-rs")
.join("wefax");
let config = WefaxConfig { let config = WefaxConfig {
output_dir: Some(wefax_output_dir.to_string_lossy().into_owned()), output_dir: Some(wefax_output_dir.to_string_lossy().into_owned()),
..WefaxConfig::default() ..WefaxConfig::default()
src/trx-server/src/main.rs
+148 -133
View File
@@ -660,38 +660,44 @@ fn spawn_rig_audio_stack(
} }
// Spawn APRS decoder task // Spawn APRS decoder task
let aprs_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("aprs") {
let aprs_state_rx = state_rx.clone(); let aprs_pcm_rx = pcm_tx.subscribe();
let aprs_decode_tx = decode_tx.clone(); let aprs_state_rx = state_rx.clone();
let aprs_sr = rig_cfg.audio.sample_rate; let aprs_decode_tx = decode_tx.clone();
let aprs_ch = rig_cfg.audio.channels; let aprs_sr = rig_cfg.audio.sample_rate;
let aprs_shutdown_rx = shutdown_rx.clone(); let aprs_ch = rig_cfg.audio.channels;
let aprs_logs = decoder_logs.clone(); let aprs_shutdown_rx = shutdown_rx.clone();
let aprs_histories = histories.clone(); let aprs_logs = decoder_logs.clone();
handles.push(tokio::spawn(async move { let aprs_histories = histories.clone();
tokio::select! { handles.push(tokio::spawn(async move {
_ = audio::run_aprs_decoder(aprs_sr, aprs_ch as u16, aprs_pcm_rx, aprs_state_rx, aprs_decode_tx, aprs_logs, aprs_histories) => {} tokio::select! {
_ = wait_for_shutdown(aprs_shutdown_rx) => {} _ = audio::run_aprs_decoder(aprs_sr, aprs_ch as u16, aprs_pcm_rx, aprs_state_rx, aprs_decode_tx, aprs_logs, aprs_histories) => {}
} _ = wait_for_shutdown(aprs_shutdown_rx) => {}
})); }
}));
}
// Spawn HF APRS decoder task // Spawn HF APRS decoder task
let hf_aprs_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("aprs_hf") {
let hf_aprs_state_rx = state_rx.clone(); let hf_aprs_pcm_rx = pcm_tx.subscribe();
let hf_aprs_decode_tx = decode_tx.clone(); let hf_aprs_state_rx = state_rx.clone();
let hf_aprs_sr = rig_cfg.audio.sample_rate; let hf_aprs_decode_tx = decode_tx.clone();
let hf_aprs_ch = rig_cfg.audio.channels; let hf_aprs_sr = rig_cfg.audio.sample_rate;
let hf_aprs_shutdown_rx = shutdown_rx.clone(); let hf_aprs_ch = rig_cfg.audio.channels;
let hf_aprs_logs = decoder_logs.clone(); let hf_aprs_shutdown_rx = shutdown_rx.clone();
let hf_aprs_histories = histories.clone(); let hf_aprs_logs = decoder_logs.clone();
handles.push(tokio::spawn(async move { let hf_aprs_histories = histories.clone();
tokio::select! { handles.push(tokio::spawn(async move {
_ = audio::run_hf_aprs_decoder(hf_aprs_sr, hf_aprs_ch as u16, hf_aprs_pcm_rx, hf_aprs_state_rx, hf_aprs_decode_tx, hf_aprs_logs, hf_aprs_histories) => {} tokio::select! {
_ = wait_for_shutdown(hf_aprs_shutdown_rx) => {} _ = audio::run_hf_aprs_decoder(hf_aprs_sr, hf_aprs_ch as u16, hf_aprs_pcm_rx, hf_aprs_state_rx, hf_aprs_decode_tx, hf_aprs_logs, hf_aprs_histories) => {}
} _ = wait_for_shutdown(hf_aprs_shutdown_rx) => {}
})); }
}));
}
if let Some((ais_a_pcm_rx, ais_b_pcm_rx)) = sdr_ais_pcm_rx { if let Some((ais_a_pcm_rx, ais_b_pcm_rx)) =
sdr_ais_pcm_rx.filter(|_| rig_cfg.decoders.is_enabled("ais"))
{
let ais_state_rx = state_rx.clone(); let ais_state_rx = state_rx.clone();
let ais_decode_tx = decode_tx.clone(); let ais_decode_tx = decode_tx.clone();
let ais_shutdown_rx = shutdown_rx.clone(); let ais_shutdown_rx = shutdown_rx.clone();
@@ -706,7 +712,7 @@ fn spawn_rig_audio_stack(
})); }));
} }
if let Some(vdes_iq_rx) = sdr_vdes_iq_rx { if let Some(vdes_iq_rx) = sdr_vdes_iq_rx.filter(|_| rig_cfg.decoders.is_enabled("vdes")) {
let vdes_state_rx = state_rx.clone(); let vdes_state_rx = state_rx.clone();
let vdes_decode_tx = decode_tx.clone(); let vdes_decode_tx = decode_tx.clone();
let vdes_shutdown_rx = shutdown_rx.clone(); let vdes_shutdown_rx = shutdown_rx.clone();
@@ -728,55 +734,61 @@ fn spawn_rig_audio_stack(
} }
// Spawn CW decoder task // Spawn CW decoder task
let cw_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("cw") {
let cw_state_rx = state_rx.clone(); let cw_pcm_rx = pcm_tx.subscribe();
let cw_decode_tx = decode_tx.clone(); let cw_state_rx = state_rx.clone();
let cw_sr = rig_cfg.audio.sample_rate; let cw_decode_tx = decode_tx.clone();
let cw_ch = rig_cfg.audio.channels; let cw_sr = rig_cfg.audio.sample_rate;
let cw_shutdown_rx = shutdown_rx.clone(); let cw_ch = rig_cfg.audio.channels;
let cw_logs = decoder_logs.clone(); let cw_shutdown_rx = shutdown_rx.clone();
let cw_histories = histories.clone(); let cw_logs = decoder_logs.clone();
handles.push(tokio::spawn(async move { let cw_histories = histories.clone();
tokio::select! { handles.push(tokio::spawn(async move {
_ = audio::run_cw_decoder(cw_sr, cw_ch as u16, cw_pcm_rx, cw_state_rx, cw_decode_tx, cw_logs, cw_histories) => {} tokio::select! {
_ = wait_for_shutdown(cw_shutdown_rx) => {} _ = audio::run_cw_decoder(cw_sr, cw_ch as u16, cw_pcm_rx, cw_state_rx, cw_decode_tx, cw_logs, cw_histories) => {}
} _ = wait_for_shutdown(cw_shutdown_rx) => {}
})); }
}));
}
// Spawn FT8 decoder task // Spawn FT8 decoder task
let ft8_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("ft8") {
let ft8_state_rx = state_rx.clone(); let ft8_pcm_rx = pcm_tx.subscribe();
let ft8_decode_tx = decode_tx.clone(); let ft8_state_rx = state_rx.clone();
let ft8_sr = rig_cfg.audio.sample_rate; let ft8_decode_tx = decode_tx.clone();
let ft8_ch = rig_cfg.audio.channels; let ft8_sr = rig_cfg.audio.sample_rate;
let ft8_shutdown_rx = shutdown_rx.clone(); let ft8_ch = rig_cfg.audio.channels;
let ft8_logs = decoder_logs.clone(); let ft8_shutdown_rx = shutdown_rx.clone();
let ft8_histories = histories.clone(); let ft8_logs = decoder_logs.clone();
handles.push(tokio::spawn(async move { let ft8_histories = histories.clone();
tokio::select! { handles.push(tokio::spawn(async move {
_ = audio::run_ft8_decoder(ft8_sr, ft8_ch as u16, ft8_pcm_rx, ft8_state_rx, ft8_decode_tx, ft8_logs, ft8_histories) => {} tokio::select! {
_ = wait_for_shutdown(ft8_shutdown_rx) => {} _ = audio::run_ft8_decoder(ft8_sr, ft8_ch as u16, ft8_pcm_rx, ft8_state_rx, ft8_decode_tx, ft8_logs, ft8_histories) => {}
} _ = wait_for_shutdown(ft8_shutdown_rx) => {}
})); }
}));
}
// Spawn FT4 decoder task // Spawn FT4 decoder task
let ft4_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("ft4") {
let ft4_state_rx = state_rx.clone(); let ft4_pcm_rx = pcm_tx.subscribe();
let ft4_decode_tx = decode_tx.clone(); let ft4_state_rx = state_rx.clone();
let ft4_sr = rig_cfg.audio.sample_rate; let ft4_decode_tx = decode_tx.clone();
let ft4_ch = rig_cfg.audio.channels; let ft4_sr = rig_cfg.audio.sample_rate;
let ft4_shutdown_rx = shutdown_rx.clone(); let ft4_ch = rig_cfg.audio.channels;
let ft4_histories = histories.clone(); let ft4_shutdown_rx = shutdown_rx.clone();
handles.push(tokio::spawn(async move { let ft4_histories = histories.clone();
tokio::select! { handles.push(tokio::spawn(async move {
_ = audio::run_ft4_decoder(ft4_sr, ft4_ch as u16, ft4_pcm_rx, ft4_state_rx, ft4_decode_tx, ft4_histories) => {} tokio::select! {
_ = wait_for_shutdown(ft4_shutdown_rx) => {} _ = audio::run_ft4_decoder(ft4_sr, ft4_ch as u16, ft4_pcm_rx, ft4_state_rx, ft4_decode_tx, ft4_histories) => {}
} _ = wait_for_shutdown(ft4_shutdown_rx) => {}
})); }
}));
}
// Spawn FT2 decoder task // Spawn FT2 decoder task
#[cfg(feature = "ft2")] #[cfg(feature = "ft2")]
{ if rig_cfg.decoders.is_enabled("ft2") {
let ft2_pcm_rx = pcm_tx.subscribe(); let ft2_pcm_rx = pcm_tx.subscribe();
let ft2_state_rx = state_rx.clone(); let ft2_state_rx = state_rx.clone();
let ft2_decode_tx = decode_tx.clone(); let ft2_decode_tx = decode_tx.clone();
@@ -793,73 +805,76 @@ fn spawn_rig_audio_stack(
} }
// Spawn WSPR decoder task // Spawn WSPR decoder task
let wspr_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("wspr") {
let wspr_state_rx = state_rx.clone(); let wspr_pcm_rx = pcm_tx.subscribe();
let wspr_decode_tx = decode_tx.clone(); let wspr_state_rx = state_rx.clone();
let wspr_sr = rig_cfg.audio.sample_rate; let wspr_decode_tx = decode_tx.clone();
let wspr_ch = rig_cfg.audio.channels; let wspr_sr = rig_cfg.audio.sample_rate;
let wspr_shutdown_rx = shutdown_rx.clone(); let wspr_ch = rig_cfg.audio.channels;
let wspr_logs = decoder_logs.clone(); let wspr_shutdown_rx = shutdown_rx.clone();
let wspr_histories = histories.clone(); let wspr_logs = decoder_logs.clone();
handles.push(tokio::spawn(async move { let wspr_histories = histories.clone();
tokio::select! { handles.push(tokio::spawn(async move {
_ = audio::run_wspr_decoder(wspr_sr, wspr_ch as u16, wspr_pcm_rx, wspr_state_rx, wspr_decode_tx, wspr_logs, wspr_histories) => {} tokio::select! {
_ = wait_for_shutdown(wspr_shutdown_rx) => {} _ = audio::run_wspr_decoder(wspr_sr, wspr_ch as u16, wspr_pcm_rx, wspr_state_rx, wspr_decode_tx, wspr_logs, wspr_histories) => {}
} _ = wait_for_shutdown(wspr_shutdown_rx) => {}
})); }
}));
}
// Spawn Meteor-M LRPT decoder task // Spawn Meteor-M LRPT decoder task
let lrpt_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("lrpt") {
let lrpt_state_rx = state_rx.clone(); let lrpt_pcm_rx = pcm_tx.subscribe();
let lrpt_decode_tx = decode_tx.clone(); let lrpt_state_rx = state_rx.clone();
let lrpt_sr = rig_cfg.audio.sample_rate; let lrpt_decode_tx = decode_tx.clone();
let lrpt_ch = rig_cfg.audio.channels; let lrpt_sr = rig_cfg.audio.sample_rate;
let lrpt_shutdown_rx = shutdown_rx.clone(); let lrpt_ch = rig_cfg.audio.channels;
let lrpt_histories = histories.clone(); let lrpt_shutdown_rx = shutdown_rx.clone();
let lrpt_output_dir = dirs::cache_dir() let lrpt_histories = histories.clone();
.unwrap_or_else(|| std::path::PathBuf::from(".cache")) let lrpt_output_dir = rig_cfg.decoders.output_dir_for("lrpt");
.join("trx-rs") handles.push(tokio::spawn(async move {
.join("lrpt"); tokio::select! {
handles.push(tokio::spawn(async move { _ = audio::run_lrpt_decoder(lrpt_sr, lrpt_ch as u16, lrpt_pcm_rx, lrpt_state_rx, lrpt_decode_tx, lrpt_histories, lrpt_output_dir) => {}
tokio::select! { _ = wait_for_shutdown(lrpt_shutdown_rx) => {}
_ = audio::run_lrpt_decoder(lrpt_sr, lrpt_ch as u16, lrpt_pcm_rx, lrpt_state_rx, lrpt_decode_tx, lrpt_histories, lrpt_output_dir) => {} }
_ = wait_for_shutdown(lrpt_shutdown_rx) => {} }));
} }
}));
// Spawn WEFAX decoder task // Spawn WEFAX decoder task
let wefax_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("wefax") {
let wefax_state_rx = state_rx.clone(); let wefax_pcm_rx = pcm_tx.subscribe();
let wefax_decode_tx = decode_tx.clone(); let wefax_state_rx = state_rx.clone();
let wefax_sr = rig_cfg.audio.sample_rate; let wefax_decode_tx = decode_tx.clone();
let wefax_ch = rig_cfg.audio.channels; let wefax_sr = rig_cfg.audio.sample_rate;
let wefax_shutdown_rx = shutdown_rx.clone(); let wefax_ch = rig_cfg.audio.channels;
let wefax_histories = histories.clone(); let wefax_shutdown_rx = shutdown_rx.clone();
handles.push(tokio::spawn(async move { let wefax_histories = histories.clone();
tokio::select! { let wefax_output_dir = rig_cfg.decoders.output_dir_for("wefax");
_ = audio::run_wefax_decoder(wefax_sr, wefax_ch as u16, wefax_pcm_rx, wefax_state_rx, wefax_decode_tx, wefax_histories) => {} handles.push(tokio::spawn(async move {
_ = wait_for_shutdown(wefax_shutdown_rx) => {} tokio::select! {
} _ = audio::run_wefax_decoder(wefax_sr, wefax_ch as u16, wefax_pcm_rx, wefax_state_rx, wefax_decode_tx, wefax_histories, wefax_output_dir) => {}
})); _ = wait_for_shutdown(wefax_shutdown_rx) => {}
}
}));
}
// Spawn SSTV decoder task // Spawn SSTV decoder task
let sstv_pcm_rx = pcm_tx.subscribe(); if rig_cfg.decoders.is_enabled("sstv") {
let sstv_state_rx = state_rx.clone(); let sstv_pcm_rx = pcm_tx.subscribe();
let sstv_decode_tx = decode_tx.clone(); let sstv_state_rx = state_rx.clone();
let sstv_sr = rig_cfg.audio.sample_rate; let sstv_decode_tx = decode_tx.clone();
let sstv_ch = rig_cfg.audio.channels; let sstv_sr = rig_cfg.audio.sample_rate;
let sstv_shutdown_rx = shutdown_rx.clone(); let sstv_ch = rig_cfg.audio.channels;
let sstv_histories = histories.clone(); let sstv_shutdown_rx = shutdown_rx.clone();
let sstv_output_dir = dirs::cache_dir() let sstv_histories = histories.clone();
.unwrap_or_else(|| std::path::PathBuf::from(".cache")) let sstv_output_dir = rig_cfg.decoders.output_dir_for("sstv");
.join("trx-rs") handles.push(tokio::spawn(async move {
.join("sstv"); tokio::select! {
handles.push(tokio::spawn(async move { _ = audio::run_sstv_decoder(sstv_sr, sstv_ch as u16, sstv_pcm_rx, sstv_state_rx, sstv_decode_tx, sstv_histories, sstv_output_dir) => {}
tokio::select! { _ = wait_for_shutdown(sstv_shutdown_rx) => {}
_ = audio::run_sstv_decoder(sstv_sr, sstv_ch as u16, sstv_pcm_rx, sstv_state_rx, sstv_decode_tx, sstv_histories, sstv_output_dir) => {} }
_ = wait_for_shutdown(sstv_shutdown_rx) => {} }));
} }
}));
} }
if rig_cfg.audio.tx_enabled { if rig_cfg.audio.tx_enabled {