koan_core/upnp/
mod.rs

1//! Playing to UPnP AV MediaRenderers: network amplifiers and streamers that
2//! fetch a URL they are given and decode it themselves.
3//!
4//! A renderer is this koan's output, the way a USB DAC is. The queue, cursor,
5//! history and everything else stay in the `Player`; only the audio goes
6//! elsewhere, as the original file, so playback stays bit-perfect up to the
7//! renderer's own DAC. Nothing here involves the server.
8//!
9//! Hand-rolled on std threads and the blocking reqwest koan already uses:
10//! eight AVTransport actions, two RenderingControl ones, SSDP and GENA.
11
12pub mod description;
13pub mod didl;
14pub mod discovery;
15pub mod serve;
16pub mod session;
17pub mod soap;
18pub mod stream;
19mod xml;
20
21#[cfg(test)]
22pub(crate) mod fake;
23
24pub use description::Renderer;
25
26/// The renderer this koan is playing to, as the front ends show it.
27#[derive(Debug, Clone, PartialEq)]
28pub struct Output {
29    pub udn: String,
30    pub name: String,
31    /// 0–100, when the renderer has a volume control.
32    pub volume: Option<u8>,
33    /// Why the current track is not playing there, when koan skipped it.
34    pub problem: Option<String>,
35}
36
37/// An open session with a renderer, for the player to play to.
38///
39/// What the renderer is heard to do reaches the player as
40/// `PlayerCommand::Renderer`, tagged with the number of the player session it
41/// was heard in: `tag`, which the player keeps up to date. An event from a
42/// session already over is recognised and dropped.
43#[derive(Debug)]
44pub struct Connection {
45    pub session: session::Session,
46    pub tag: std::sync::Arc<std::sync::atomic::AtomicU64>,
47}
48
49/// Open a session with `renderer` whose events go to `player`.
50pub fn open(
51    renderer: Renderer,
52    player: &crossbeam_channel::Sender<crate::player::commands::PlayerCommand>,
53) -> Result<Connection, String> {
54    let tag = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
55    let tx = player.clone();
56    let tagged = tag.clone();
57    let session = session::Session::open(renderer, move |event| {
58        let cmd = crate::player::commands::PlayerCommand::Renderer {
59            session: tagged.load(std::sync::atomic::Ordering::Acquire),
60            event,
61        };
62        match &cmd {
63            // Only a wake-up: the player reads the loss from the session. It
64            // can be raised on the player's own thread, or under the
65            // discovery lock, where waiting for room in the channel could
66            // never end.
67            crate::player::commands::PlayerCommand::Renderer {
68                event: session::Event::Gone,
69                ..
70            } => {
71                let _ = tx.try_send(cmd);
72            }
73            _ => {
74                let _ = tx.send(cmd);
75            }
76        }
77    })?;
78    Ok(Connection { session, tag })
79}
80
81/// Each choice of output, in the order they were made. Opening a session
82/// takes a few round trips; one that finishes after a later choice is
83/// dropped, so the last one picked is where the music goes.
84static CHOICE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
85
86/// Record a choice of output, made now, and return its place. Called when the
87/// choice is made, before any work towards it: every output change (a
88/// renderer, a local device, the system default) takes one.
89pub fn choose() -> u64 {
90    CHOICE.fetch_add(1, std::sync::atomic::Ordering::AcqRel) + 1
91}
92
93/// Make the renderer `udn` this koan's output, carrying on from where the
94/// music is, unless a later choice was made while the session opened.
95/// `choice` is what `choose` returned when this one was picked. Blocks for
96/// the few round trips a session takes to open, so call it from a thread
97/// nothing is waiting on.
98pub fn connect(
99    udn: &str,
100    choice: u64,
101    player: &crossbeam_channel::Sender<crate::player::commands::PlayerCommand>,
102) -> Result<(), String> {
103    let renderer = discovery::find(udn)
104        .ok_or_else(|| "That renderer is no longer on the network.".to_string())?;
105    let connection = open(renderer, player)?;
106    if CHOICE.load(std::sync::atomic::Ordering::Acquire) != choice {
107        log::info!(
108            "upnp: {} opened after a later choice of output; not used",
109            connection.session.renderer().name
110        );
111        return Ok(());
112    }
113    player
114        .send(crate::player::commands::PlayerCommand::UseRenderer(Some(
115            Box::new(connection),
116        )))
117        .map_err(|_| "The player has stopped.".to_string())
118}
119
120/// How long after launch the renderer used last time is looked for.
121const RESUME_WINDOW: std::time::Duration = std::time::Duration::from_secs(6);
122
123/// Go back to the renderer `udn`, used last time, on a thread of its own:
124/// look for it for `RESUME_WINDOW`, waking on each change discovery
125/// announces, and hand the player an open session as
126/// `PlayerCommand::ResumeRenderer` if it turns up idle. The player takes it
127/// only if nobody has played anything or picked an output meanwhile.
128pub fn resume(
129    udn: String,
130    player: &crossbeam_channel::Sender<crate::player::commands::PlayerCommand>,
131) {
132    let choice = CHOICE.load(std::sync::atomic::Ordering::Acquire);
133    let player = player.clone();
134    let _ = std::thread::Builder::new()
135        .name("koan-upnp-resume".into())
136        .spawn(move || {
137            let _ = player.send(match resume_onto(&udn, RESUME_WINDOW, choice, &player) {
138                Some(connection) => {
139                    crate::player::commands::PlayerCommand::ResumeRenderer(Box::new(connection))
140                }
141                None => crate::player::commands::PlayerCommand::ResumeRendererMissed,
142            });
143        });
144}
145
146/// The session `resume` hands over, if the renderer is found within `window`,
147/// nothing else has been picked since `choice`, and it is not playing or
148/// paused for something else: whatever is driving it, a phone on the same
149/// amplifier say, is not taken over by a launch nobody asked for. Still on a
150/// track a kōan on this machine sent it, a run that ended without stopping
151/// it, it is ours to take back.
152fn resume_onto(
153    udn: &str,
154    window: std::time::Duration,
155    choice: u64,
156    player: &crossbeam_channel::Sender<crate::player::commands::PlayerCommand>,
157) -> Option<Connection> {
158    let Some(renderer) = await_renderer(udn, window) else {
159        log::info!("upnp: {udn}, used last time, is not on the network; playing here");
160        return None;
161    };
162    if CHOICE.load(std::sync::atomic::Ordering::Acquire) != choice {
163        return None;
164    }
165    match discovery::in_use(&renderer) {
166        Some(false) => {}
167        Some(true)
168            if discovery::playing_uri(&renderer)
169                .is_some_and(|uri| served_from_here(&renderer, &uri) && !still_served(&uri)) =>
170        {
171            log::info!(
172                "upnp: {} is still on what a kōan here sent it; taking it back",
173                renderer.name
174            );
175        }
176        _ => {
177            log::info!(
178                "upnp: {}, used last time, is busy or not answering; playing here",
179                renderer.name
180            );
181            return None;
182        }
183    }
184    open(renderer, player)
185        .inspect_err(|e| log::info!("upnp: could not go back to {udn}: {e}"))
186        .ok()
187}
188
189/// Whether `uri` is one a kōan on this machine served `renderer`: a track's
190/// tokenised path (`serve`'s `/t/<token>`) on this machine's address towards
191/// it. The port is not compared: every run listens on a new one.
192fn served_from_here(renderer: &Renderer, uri: &str) -> bool {
193    let Ok(url) = url::Url::parse(uri) else {
194        return false;
195    };
196    let host = |u: &url::Url| {
197        u.host_str()
198            .and_then(|h| h.parse::<std::net::IpAddr>().ok())
199    };
200    let Some(here) = host(&renderer.location).and_then(serve::local_ip_towards) else {
201        return false;
202    };
203    let token = url
204        .path()
205        .strip_prefix("/t/")
206        .and_then(|t| t.split('.').next())
207        .unwrap_or_default();
208    host(&url) == Some(here) && token.len() == 32 && token.bytes().all(|b| b.is_ascii_hexdigit())
209}
210
211/// Whether a kōan is still serving `uri`: another one on this machine, a
212/// TUI beside the app say, is playing to the renderer, and it is not ours to
213/// take. A run that has ended refuses the connection; a live one that has let
214/// the track go answers 404.
215fn still_served(uri: &str) -> bool {
216    use std::io::{Read, Write};
217    let Ok(url) = url::Url::parse(uri) else {
218        return false;
219    };
220    let Some(addr) = url
221        .socket_addrs(|| Some(80))
222        .ok()
223        .and_then(|a| a.into_iter().next())
224    else {
225        return false;
226    };
227    let timeout = std::time::Duration::from_millis(500);
228    let Ok(mut stream) = std::net::TcpStream::connect_timeout(&addr, timeout) else {
229        return false;
230    };
231    let _ = stream.set_read_timeout(Some(timeout));
232    if write!(stream, "HEAD {} HTTP/1.1\r\nHost: x\r\n\r\n", url.path()).is_err() {
233        return false;
234    }
235    let mut head = [0u8; 12];
236    stream.read_exact(&mut head).is_ok() && head.starts_with(b"HTTP/1.1 2")
237}
238
239/// The renderer `udn` once discovery has found it, waiting at most `window`.
240fn await_renderer(udn: &str, window: std::time::Duration) -> Option<Renderer> {
241    let signal = crate::signal::engine_changed();
242    let mut seen = signal.generation();
243    let deadline = std::time::Instant::now() + window;
244    discovery::search();
245    loop {
246        if let Some(renderer) = discovery::find(udn) {
247            return Some(renderer);
248        }
249        let left = deadline.saturating_duration_since(std::time::Instant::now());
250        if left.is_zero() {
251            return None;
252        }
253        seen = signal.wait_until(seen, left);
254    }
255}
256
257/// Bring the music back to this device's own output.
258pub fn disconnect(player: &crossbeam_channel::Sender<crate::player::commands::PlayerCommand>) {
259    choose();
260    let _ = player.send(crate::player::commands::PlayerCommand::UseRenderer(None));
261}
262
263#[cfg(test)]
264mod tests {
265    use super::*;
266
267    /// A renderer discovery knows is found at once; one it never hears of
268    /// is given up on when the window closes, and the music stays here.
269    #[test]
270    fn a_remembered_renderer_is_found_or_given_up_on() {
271        let fake = fake::FakeRenderer::start("http-get:*:audio/wav:*", true, false);
272        let renderer = fake.renderer();
273        discovery::remember(renderer.clone(), std::time::Duration::from_secs(60));
274        assert_eq!(
275            await_renderer(&renderer.udn, std::time::Duration::from_secs(1)).map(|r| r.udn),
276            Some(renderer.udn)
277        );
278
279        let start = std::time::Instant::now();
280        let window = std::time::Duration::from_millis(300);
281        assert!(await_renderer("uuid:nowhere", window).is_none());
282        assert!(start.elapsed() >= window);
283    }
284
285    /// A renderer playing for something else when the app launches is left
286    /// to it; an idle one is gone back to.
287    #[test]
288    fn a_busy_renderer_is_not_taken_at_launch() {
289        let fake = fake::FakeRenderer::start("http-get:*:audio/wav:*", true, false);
290        let renderer = fake.renderer();
291        discovery::remember(renderer.clone(), std::time::Duration::from_secs(60));
292        let (tx, _rx) = crossbeam_channel::unbounded();
293        let window = std::time::Duration::from_secs(1);
294        let now = CHOICE.load(std::sync::atomic::Ordering::Acquire);
295
296        fake.play_foreign("http://phone/track.flac");
297        assert!(resume_onto(&renderer.udn, window, now, &tx).is_none());
298
299        // Still on a track a kōan here sent it, from a run that ended
300        // without stopping it: ours, taken back.
301        let gone = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
302        let gone_port = gone.local_addr().unwrap().port();
303        drop(gone);
304        fake.play_foreign(&format!(
305            "http://127.0.0.1:{gone_port}/t/{}.flac",
306            "0a".repeat(16)
307        ));
308        assert!(resume_onto(&renderer.udn, window, now, &tx).is_some());
309
310        // On a track a kōan here is still serving, a TUI beside the app say:
311        // that one's, left alone.
312        let dir = tempfile::tempdir().unwrap();
313        let file = dir.path().join("t.flac");
314        std::fs::write(&file, b"fLaC").unwrap();
315        let live = serve::Listener::start(Box::new(|_, _| {})).unwrap();
316        let token = live.add(serve::Served::File {
317            path: file,
318            mime: "audio/flac".into(),
319        });
320        fake.play_foreign(&format!("http://127.0.0.1:{}/t/{token}.flac", live.port()));
321        assert!(resume_onto(&renderer.udn, window, now, &tx).is_none());
322
323        fake.press_stop(0);
324        assert!(resume_onto(&renderer.udn, window, now, &tx).is_some());
325    }
326}
koan_core/upnp/mod.rs

koan_core/upnp/
mod.rs
