diff --git a/src/app/cast.rs b/src/app/cast.rs index 4d00bbc..b693791 100644 --- a/src/app/cast.rs +++ b/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}")), diff --git a/src/app/mod.rs b/src/app/mod.rs index a2f8300..0244f9a 100644 --- a/src/app/mod.rs +++ b/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, rx: Receiver>) -> Result< Ok(done_rx) } -fn spawn_forwarder(rx: Receiver, tx: SyncSender, id: u64) -> Result<()> { - thread::Builder::new() - .name("player-events".into()) - .spawn(move || forward(rx, tx, id))?; - Ok(()) -} - -fn forward(rx: Receiver, tx: SyncSender, id: u64) { - for ev in rx { - if tx.send(AppEvent::Player(id, ev)).is_err() { - break; - } - } -} - fn spawn_input(tx: SyncSender) -> Result<()> { thread::Builder::new().name("input".into()).spawn(move || { while let Ok(ev) = event::read() { diff --git a/src/audio/player.rs b/src/audio/player.rs index 9d374b9..f5cb902 100644 --- a/src/audio/player.rs +++ b/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, cache_dir: PathBuf, volume: f32) -> Result { +pub fn spawn( + client: Arc, + cache_dir: PathBuf, + volume: f32, + events: Events, +) -> Result> { 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, 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, 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, cache_dir: PathBuf, track: Arc, - events: SyncSender, + 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, cache_dir: &std::path::Path, track: Arc, - events: &SyncSender, + events: &Events, seeks: &Receiver<(u64, u64)>, out: &SyncSender, ) { @@ -166,7 +170,7 @@ struct Player { out: Output, client: Arc, cache_dir: PathBuf, - events: SyncSender, + events: Events, decoding: Option, next: Option, starved: bool, @@ -566,7 +570,7 @@ fn open( client: &Arc, cache_dir: &std::path::Path, track: &Track, - events: &SyncSender, + events: &Events, ) -> Result { static SEQ: AtomicU64 = AtomicU64::new(0); if track.part_key.is_empty() { diff --git a/src/backend.rs b/src/backend.rs index 399f0ab..71f2793 100644 --- a/src/backend.rs +++ b/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, Receiver); +/// 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, + id: u64, +} + +impl Events { + pub fn new(tx: SyncSender, id: u64) -> Self { + Self { tx, id } + } + + pub fn send(&self, ev: PlayerEvent) -> Result<(), SendError> { + self.tx.send(AppEvent::Player(self.id, ev)) + } + + pub fn try_send(&self, ev: PlayerEvent) -> Result<(), TrySendError> { + self.tx.try_send(AppEvent::Player(self.id, ev)) + } +} #[derive(Debug)] pub enum PlayerCmd { diff --git a/src/cast/caster/mod.rs b/src/cast/caster/mod.rs index 186a9dd..030f413 100644 --- a/src/cast/caster/mod.rs +++ b/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, token: String, - events: SyncSender, + events: Events, pub outbox: Vec<(String, Payload, Option)>, pub conn: Conn, last_req: u32, @@ -87,7 +86,7 @@ pub struct Caster { } impl Caster { - pub fn new(plex: Arc, token: String, events: SyncSender) -> Self { + pub fn new(plex: Arc, token: String, events: Events) -> Self { Self { plex, token, diff --git a/src/cast/mod.rs b/src/cast/mod.rs index a315f97..a32bdc1 100644 --- a/src/cast/mod.rs +++ b/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, 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, + ip: IpAddr, + events: Events, +) -> Result<(SyncSender, 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, 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, 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 {