task: remove forwarders

src/app/cast.rs

@@ -1,13 +1,13 @@
 use std::net::IpAddr;
 use std::sync::Arc;
-use std::sync::mpsc::{Receiver, SyncSender};
+use std::sync::mpsc::{Receiver, SyncSender, channel};
 use std::thread;
 use std::time::Duration;
 
+use super::AppEvent;
 use super::keys::Action;
 use super::state::{AppState, Modal};
-use super::{AppEvent, forward};
-use crate::backend::PlayerCmd;
+use crate::backend::{Events, PlayerCmd};
 use crate::cast::Device;
 use crate::plex::PlexClient;
 
@@ -149,20 +149,24 @@ impl AppState {
         let id = self.cast.request;
         let (events, plex) = (self.cast.events.clone(), self.cast.plex.clone());
         let label = name.clone();
-        let spawned = thread::Builder::new()
-            .name("cast-connect".into())
-            .spawn(move || {
-                let (result, rx) = match crate::cast::spawn(plex, ip) {
-                    Ok(((player, rx), done)) => (Ok((player, done)), Some(rx)),
+
+        let spawned = thread::Builder::new().name("cast".into()).spawn(move || {
+            let (done_tx, done) = channel();
+
+            let (result, session) =
+                match crate::cast::connect(plex, ip, Events::new(events.clone(), id)) {
+                    Ok((player, session)) => (Ok((player, done)), Some(session)),
                     Err(e) => (Err(format!("{e:#}")), None),
                 };
-                // the app drops a session's events until it has switched to it,
-                // so they must not overtake Connected
-                let _ = events.send(AppEvent::Cast(CastEvent::Connected { id, name, result }));
-                if let Some(rx) = rx {
-                    forward(rx, events, id);
-                }
-            });
+
+            // drop all session events until switch is done
+            let _ = events.send(AppEvent::Cast(CastEvent::Connected { id, name, result }));
+            if let Some(session) = session {
+                let _done = done_tx;
+                session();
+            }
+        });
+
         match spawned {
             Ok(_) => self.info(format!("connecting to {label}…")),
             Err(e) => self.error(format!("can't cast to {label}: {e}")),

src/app/mod.rs

@@ -19,7 +19,7 @@ use ratatui::crossterm::event::{self, Event, KeyEventKind};
 use tracing::warn;
 
 use crate::audio::player;
-use crate::backend::PlayerEvent;
+use crate::backend::{Events, PlayerEvent};
 #[cfg(feature = "cast")]
 use crate::cast::Device;
 use crate::config::{self, SavedState};
@@ -118,19 +118,22 @@ pub fn run(
         .clone()
         .context("no music section selected")?;
 
-    let (tx, rx) = sync_channel(64);
+    let (tx, rx) = sync_channel(256);
     let (api_tx, api_rx) = sync_channel(16);
 
     let cache = config::dirs()?.cache_dir().join("stream");
-    let (player_tx, player_rx) = player::spawn(client.clone(), cache, saved.volume)?;
+    let player_tx = player::spawn(
+        client.clone(),
+        cache,
+        saved.volume,
+        Events::new(tx.clone(), 0),
+    )?;
 
     spawn_api_workers(client.clone(), section, saved, api_rx, tx.clone())?;
 
     let (report_tx, report_rx) = channel();
     let reports_done = spawn_reporter(client.clone(), report_rx)?;
 
-    spawn_forwarder(player_rx, tx.clone(), 0)?;
-
     #[cfg(unix)]
     spawn_signals(tx.clone())?;
 
@@ -353,21 +356,6 @@ fn spawn_reporter(client: Arc<PlexClient>, rx: Receiver<Vec<Report>>) -> Result<
     Ok(done_rx)
 }
 
-fn spawn_forwarder(rx: Receiver<PlayerEvent>, tx: SyncSender<AppEvent>, id: u64) -> Result<()> {
-    thread::Builder::new()
-        .name("player-events".into())
-        .spawn(move || forward(rx, tx, id))?;
-    Ok(())
-}
-
-fn forward(rx: Receiver<PlayerEvent>, tx: SyncSender<AppEvent>, id: u64) {
-    for ev in rx {
-        if tx.send(AppEvent::Player(id, ev)).is_err() {
-            break;
-        }
-    }
-}
-
 fn spawn_input(tx: SyncSender<AppEvent>) -> Result<()> {
     thread::Builder::new().name("input".into()).spawn(move || {
         while let Ok(ev) = event::read() {

src/audio/player.rs

@@ -15,7 +15,7 @@ use super::decoder::Decoder;
 use super::output::Output;
 use super::resample::Resampler;
 use super::source::CachedHttpSource;
-use crate::backend::{Channels, PlayerCmd, PlayerEvent};
+use crate::backend::{Events, PlayerCmd, PlayerEvent};
 use crate::plex::PlexClient;
 use crate::plex::models::Track;
 
@@ -24,9 +24,13 @@ const IDLE_POLL: Duration = Duration::from_millis(20);
 const FILL_BUDGET: Duration = Duration::from_millis(50);
 const DECODE_AHEAD_PACKETS: usize = 32;
 
-pub fn spawn(client: Arc<PlexClient>, cache_dir: PathBuf, volume: f32) -> Result<Channels> {
+pub fn spawn(
+    client: Arc<PlexClient>,
+    cache_dir: PathBuf,
+    volume: f32,
+    events: Events,
+) -> Result<SyncSender<PlayerCmd>> {
     let (cmd_tx, cmd_rx) = sync_channel(32);
-    let (ev_tx, ev_rx) = sync_channel(256);
     let (ready_tx, ready_rx) = channel();
     let _ = std::fs::remove_dir_all(&cache_dir);
     thread::Builder::new()
@@ -44,7 +48,7 @@ pub fn spawn(client: Arc<PlexClient>, cache_dir: PathBuf, volume: f32) -> Result
                 out,
                 client,
                 cache_dir,
-                events: ev_tx,
+                events,
                 decoding: None,
                 next: None,
                 starved: false,
@@ -63,7 +67,7 @@ pub fn spawn(client: Arc<PlexClient>, cache_dir: PathBuf, volume: f32) -> Result
             .run(cmd_rx);
         })?;
     ready_rx.recv().context("player thread died")??;
-    Ok((cmd_tx, ev_rx))
+    Ok(cmd_tx)
 }
 
 enum Chunk {
@@ -90,7 +94,7 @@ impl DecodeThread {
         client: Arc<PlexClient>,
         cache_dir: PathBuf,
         track: Arc<Track>,
-        events: SyncSender<PlayerEvent>,
+        events: Events,
     ) -> Self {
         let (chunk_tx, chunks) = sync_channel(DECODE_AHEAD_PACKETS);
         let (seeks, seek_rx) = channel();
@@ -116,7 +120,7 @@ fn decode(
     client: &Arc<PlexClient>,
     cache_dir: &std::path::Path,
     track: Arc<Track>,
-    events: &SyncSender<PlayerEvent>,
+    events: &Events,
     seeks: &Receiver<(u64, u64)>,
     out: &SyncSender<Chunk>,
 ) {
@@ -166,7 +170,7 @@ struct Player {
     out: Output,
     client: Arc<PlexClient>,
     cache_dir: PathBuf,
-    events: SyncSender<PlayerEvent>,
+    events: Events,
     decoding: Option<DecodeThread>,
     next: Option<DecodeThread>,
     starved: bool,
@@ -566,7 +570,7 @@ fn open(
     client: &Arc<PlexClient>,
     cache_dir: &std::path::Path,
     track: &Track,
-    events: &SyncSender<PlayerEvent>,
+    events: &Events,
 ) -> Result<Decoder> {
     static SEQ: AtomicU64 = AtomicU64::new(0);
     if track.part_key.is_empty() {

src/backend.rs

@@ -2,11 +2,32 @@
 //! device. A backend runs on its own thread, takes `PlayerCmd`s and answers with `PlayerEvent`s
 
 use std::sync::Arc;
-use std::sync::mpsc::{Receiver, SyncSender};
+use std::sync::mpsc::{SendError, SyncSender, TrySendError};
 
+use crate::app::AppEvent;
 use crate::plex::models::Track;
 
-pub type Channels = (SyncSender<PlayerCmd>, Receiver<PlayerEvent>);
+/// sends a backends events into the apps channel, tagged with the backends id so the app
+/// can ignore any backends it has switched away from
+#[derive(Clone)]
+pub struct Events {
+    tx: SyncSender<AppEvent>,
+    id: u64,
+}
+
+impl Events {
+    pub fn new(tx: SyncSender<AppEvent>, id: u64) -> Self {
+        Self { tx, id }
+    }
+
+    pub fn send(&self, ev: PlayerEvent) -> Result<(), SendError<AppEvent>> {
+        self.tx.send(AppEvent::Player(self.id, ev))
+    }
+
+    pub fn try_send(&self, ev: PlayerEvent) -> Result<(), TrySendError<AppEvent>> {
+        self.tx.try_send(AppEvent::Player(self.id, ev))
+    }
+}
 
 #[derive(Debug)]
 pub enum PlayerCmd {

src/cast/caster/mod.rs

@@ -5,14 +5,13 @@ mod messages;
 mod status;
 
 use std::sync::Arc;
-use std::sync::mpsc::SyncSender;
 use std::time::Instant;
 
 use cast_sender::Payload;
 use cast_sender::namespace::media::Media;
 use tracing::debug;
 
-use crate::backend::{DeviceVolume, PlayerEvent};
+use crate::backend::{DeviceVolume, Events, PlayerEvent};
 use crate::plex::PlexClient;
 use crate::plex::models::Track;
 
@@ -66,7 +65,7 @@ impl Position {
 pub struct Caster {
     plex: Arc<PlexClient>,
     token: String,
-    events: SyncSender<PlayerEvent>,
+    events: Events,
     pub outbox: Vec<(String, Payload, Option<u32>)>,
     pub conn: Conn,
     last_req: u32,
@@ -87,7 +86,7 @@ pub struct Caster {
 }
 
 impl Caster {
-    pub fn new(plex: Arc<PlexClient>, token: String, events: SyncSender<PlayerEvent>) -> Self {
+    pub fn new(plex: Arc<PlexClient>, token: String, events: Events) -> Self {
         Self {
             plex,
             token,

src/cast/mod.rs

@@ -9,18 +9,24 @@ pub use discover::{Device, discover};
 
 use std::net::IpAddr;
 use std::sync::Arc;
-use std::sync::mpsc::{Receiver, channel as std_channel, sync_channel};
+use std::sync::mpsc::{SyncSender, sync_channel};
 use std::thread;
 
 use anyhow::{Context, Result, bail};
 use smol::block_on;
 use smol::channel;
 
-use crate::backend::Channels;
+use crate::backend::{Events, PlayerCmd};
 use crate::plex::PlexClient;
 use caster::Caster;
 
-pub fn spawn(plex: Arc<PlexClient>, ip: IpAddr) -> Result<(Channels, Receiver<()>)> {
+/// connects to the device and returns its command sender and the session, which runs on
+/// the calling thread until the device disconnects or the sender is dropped
+pub fn connect(
+    plex: Arc<PlexClient>,
+    ip: IpAddr,
+    events: Events,
+) -> Result<(SyncSender<PlayerCmd>, impl FnOnce())> {
     let server = plex.base();
     if loopback(&server) {
         bail!("rexamp reaches plex at {server}, which the device can't");
@@ -30,7 +36,6 @@ pub fn spawn(plex: Arc<PlexClient>, ip: IpAddr) -> Result<(Channels, Receiver<()
         .context("getting a cast token from plex")?;
     let first = block_on(io::connect(ip)).with_context(|| format!("casting to {ip}"))?;
     let (cmd_tx, cmd_rx) = sync_channel(32);
-    let (ev_tx, ev_rx) = sync_channel(256);
     let (fwd_tx, fwd_rx) = channel::bounded(32);
 
     thread::Builder::new()
@@ -43,12 +48,9 @@ pub fn spawn(plex: Arc<PlexClient>, ip: IpAddr) -> Result<(Channels, Receiver<()
                 }
             }
         })?;
-    let (done_tx, done_rx) = std_channel::<()>();
-    thread::Builder::new().name("cast".into()).spawn(move || {
-        let _done = done_tx;
-        block_on(io::run(Caster::new(plex, token, ev_tx), ip, first, fwd_rx))
-    })?;
-    Ok(((cmd_tx, ev_rx), done_rx))
+
+    let session = move || block_on(io::run(Caster::new(plex, token, events), ip, first, fwd_rx));
+    Ok((cmd_tx, session))
 }
 
 fn loopback(url: &str) -> bool {