Krolyxon
Commit b139f20

Branches

Tags

Refs →

rmptuisrcmpdworker.rs

477 lines · 15.4 KB · Rust

Raw Blame History Download
1//! Background tasks that own the MPD connections.
2//!
3//! * The **idle** task blocks on `idle` and reports changed subsystems.
4//! * The **command** task executes [`Request`]s sent through an [`MpdHandle`]
5//! and reports query results as [`Reply`]s.
6//!
7//! The UI never awaits the server: it sends requests and reacts to events.
8
9use std::{path::PathBuf, time::Duration};
10
11use tokio::{sync::mpsc, time::sleep};
12use tracing::{debug, warn};
13
14use super::{
15 commands,
16 connection::{Address, Connection},
17 error::{MpdError, MpdResult},
18 protocol::{Command, Pair},
19 types::{
20 Album, DirEntry, PlaylistInfo, Song, Status, Subsystem, parse_albums, parse_entries,
21 parse_playlists, parse_songs, parse_subsystems, parse_tag_values,
22 },
23};
24
25/// Data the UI asks for.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub enum Query {
28 /// Status and current song.
29 Status,
30 /// Queue changes since the given queue version (`0` for everything).
31 QueueChanges {
32 since: u32,
33 },
34 Playlists,
35 Directory(String),
36 PlaylistSongs(String),
37 /// All values of a tag, e.g. every `AlbumArtist`.
38 TagValues(String),
39 /// Albums whose `tag` equals `value`, grouped by date.
40 Albums {
41 tag: String,
42 value: String,
43 },
44 /// Songs matching a filter expression.
45 Find(String),
46 /// The server's music directory (only answered on local connections).
47 MusicDirectory,
48}
49
50/// The answer to a [`Query`].
51#[derive(Clone, Debug)]
52pub enum Reply {
53 Status {
54 status: Status,
55 current: Option<Song>,
56 },
57 QueueChanges {
58 status: Status,
59 since: u32,
60 changes: Vec<Song>,
61 },
62 Playlists(Vec<PlaylistInfo>),
63 Directory {
64 path: String,
65 entries: Vec<DirEntry>,
66 },
67 PlaylistSongs {
68 name: String,
69 songs: Vec<Song>,
70 },
71 TagValues {
72 tag: String,
73 values: Vec<String>,
74 },
75 Albums {
76 tag: String,
77 value: String,
78 albums: Vec<Album>,
79 },
80 Find {
81 filter: String,
82 songs: Vec<Song>,
83 },
84 MusicDirectory(Option<PathBuf>),
85}
86
87#[derive(Clone, Debug, PartialEq, Eq)]
88pub enum Request {
89 /// Fire-and-forget commands, executed atomically as one command list.
90 Exec(Vec<Command>),
91 Query(Query),
92 /// Appends songs and starts playing the first one.
93 AddAndPlay(Vec<String>),
94 /// Appends every song matching a filter and plays the first one.
95 FindAddAndPlay(String),
96 /// Appends a stored playlist and plays its song at `offset`.
97 LoadAndPlay {
98 name: String,
99 offset: u32,
100 },
101}
102
103/// Notifications from the background tasks.
104#[derive(Clone, Debug)]
105pub enum MpdEvent {
106 Connected { version: String },
107 Disconnected(String),
108 Changed(Vec<Subsystem>),
109 Reply(Reply),
110 Error(String),
111}
112
113/// Cheap, cloneable sender for requests.
114#[derive(Clone, Debug)]
115pub struct MpdHandle {
116 tx: mpsc::UnboundedSender<Request>,
117}
118
119impl MpdHandle {
120 pub fn request(&self, request: Request) {
121 // The worker lives as long as the application, so a closed channel
122 // only happens during shutdown.
123 let _ = self.tx.send(request);
124 }
125
126 pub fn exec(&self, command: Command) {
127 self.request(Request::Exec(vec![command]));
128 }
129
130 pub fn exec_all(&self, commands: Vec<Command>) {
131 if !commands.is_empty() {
132 self.request(Request::Exec(commands));
133 }
134 }
135
136 pub fn query(&self, query: Query) {
137 self.request(Request::Query(query));
138 }
139
140 /// A handle whose requests go to `tx`, for tests.
141 pub fn from_sender(tx: mpsc::UnboundedSender<Request>) -> Self {
142 Self { tx }
143 }
144}
145
146/// Starts both workers and returns the request handle.
147pub fn spawn<E>(
148 address: Address,
149 password: Option<String>,
150 events: mpsc::UnboundedSender<E>,
151) -> MpdHandle
152where
153 E: From<MpdEvent> + Send + 'static,
154{
155 let (tx, rx) = mpsc::unbounded_channel();
156 let emit = Emitter(events);
157 tokio::spawn(idle_task(address.clone(), password.clone(), emit.clone()));
158 tokio::spawn(command_task(address, password, rx, emit));
159 MpdHandle { tx }
160}
161
162struct Emitter<E>(mpsc::UnboundedSender<E>);
163
164impl<E> Clone for Emitter<E> {
165 fn clone(&self) -> Self {
166 Self(self.0.clone())
167 }
168}
169
170impl<E: From<MpdEvent>> Emitter<E> {
171 /// Returns `false` once the application stopped listening.
172 fn emit(&self, event: MpdEvent) -> bool {
173 self.0.send(event.into()).is_ok()
174 }
175}
176
177const MAX_BACKOFF: Duration = Duration::from_secs(10);
178
179async fn idle_task<E: From<MpdEvent>>(
180 address: Address,
181 password: Option<String>,
182 emit: Emitter<E>,
183) {
184 let mut backoff = Duration::from_millis(500);
185 let mut reported_down = false;
186 loop {
187 match Connection::connect(&address, password.as_deref()).await {
188 Ok(mut conn) => {
189 backoff = Duration::from_millis(500);
190 if !emit.emit(MpdEvent::Connected {
191 version: conn.version().to_owned(),
192 }) {
193 return;
194 }
195 let error = loop {
196 match conn.execute(&Command::new("idle")).await {
197 Ok(pairs) => {
198 if !emit.emit(MpdEvent::Changed(parse_subsystems(&pairs))) {
199 return;
200 }
201 }
202 Err(error) => break error,
203 }
204 };
205 warn!("idle connection lost: {error}");
206 if !emit.emit(MpdEvent::Disconnected(error.to_string())) {
207 return;
208 }
209 reported_down = true;
210 }
211 Err(error) => {
212 debug!("connecting to {address} failed: {error}");
213 if !reported_down {
214 reported_down = true;
215 if !emit.emit(MpdEvent::Disconnected(format!(
216 "cannot connect to {address}: {error}"
217 ))) {
218 return;
219 }
220 }
221 }
222 }
223 sleep(backoff).await;
224 backoff = (backoff * 2).min(MAX_BACKOFF);
225 }
226}
227
228async fn command_task<E: From<MpdEvent>>(
229 address: Address,
230 password: Option<String>,
231 mut requests: mpsc::UnboundedReceiver<Request>,
232 emit: Emitter<E>,
233) {
234 let mut conn: Option<Connection> = None;
235 while let Some(request) = requests.recv().await {
236 // MPD drops connections that stay silent for `connection_timeout`, so a
237 // failure on a reused connection is retried once on a fresh one.
238 let mut attempts = if conn.is_some() { 2 } else { 1 };
239 let result = loop {
240 let c = match conn.as_mut() {
241 Some(c) => c,
242 None => match Connection::connect(&address, password.as_deref()).await {
243 Ok(c) => conn.insert(c),
244 Err(error) => break Err(error),
245 },
246 };
247 attempts -= 1;
248 match execute(c, &request).await {
249 Err(error) if error.is_fatal() => {
250 conn = None;
251 if attempts == 0 {
252 break Err(error);
253 }
254 }
255 other => break other,
256 }
257 };
258 let event = match result {
259 Ok(Some(reply)) => MpdEvent::Reply(reply),
260 Ok(None) => continue,
261 Err(error) => {
262 warn!("request {request:?} failed: {error}");
263 MpdEvent::Error(error.to_string())
264 }
265 };
266 if !emit.emit(event) {
267 return;
268 }
269 }
270}
271
272async fn execute(conn: &mut Connection, request: &Request) -> MpdResult<Option<Reply>> {
273 match request {
274 Request::Exec(commands) => conn.execute_all(commands).await.map(|()| None),
275 Request::Query(query) => run_query(conn, query).await.map(Some),
276 Request::AddAndPlay(uris) => {
277 let Some((first, rest)) = uris.split_first() else {
278 return Ok(None);
279 };
280 let id = conn
281 .execute(&commands::add_id(first))
282 .await?
283 .into_iter()
284 .find(|(key, _)| key == "Id")
285 .and_then(|(_, id)| id.parse().ok())
286 .ok_or_else(|| MpdError::Protocol("addid returned no id".into()))?;
287 let mut batch: Vec<Command> = rest.iter().map(|uri| commands::add(uri)).collect();
288 batch.push(commands::play_id(id));
289 conn.execute_all(&batch).await.map(|()| None)
290 }
291 Request::FindAddAndPlay(filter) => {
292 let start = queue_length(conn).await?;
293 conn.execute_all(&[commands::find_add(filter), commands::play_pos(start)])
294 .await
295 .map(|()| None)
296 }
297 Request::LoadAndPlay { name, offset } => {
298 let start = queue_length(conn).await?;
299 let entries = conn
300 .execute(&Command::new("listplaylist").arg(name))
301 .await?;
302 conn.execute(&commands::load(name)).await?;
303 // `load` silently skips entries the server cannot resolve, so the
304 // playlist index is not necessarily the queue offset.
305 let added = conn
306 .execute(&Command::new("playlistinfo").arg(format!("{start}:")))
307 .await?;
308 let entries: Vec<&str> = files(&entries).collect();
309 let added: Vec<&str> = files(&added).collect();
310 let Some(pos) = queued_offset(&entries, &added, *offset as usize) else {
311 return Ok(None);
312 };
313 conn.execute(&commands::play_pos(start + pos as u32))
314 .await
315 .map(|_| None)
316 }
317 }
318}
319
320fn files(pairs: &[Pair]) -> impl Iterator<Item = &str> {
321 pairs
322 .iter()
323 .filter(|(key, _)| key == "file")
324 .map(|(_, value)| value.as_str())
325}
326
327/// Where playlist entry `index` ended up among the songs `load` appended,
328/// matching entries in order and skipping the ones that were dropped.
329fn queued_offset(entries: &[&str], added: &[&str], index: usize) -> Option<usize> {
330 if entries.len() == added.len() {
331 return (index < added.len()).then_some(index);
332 }
333 let mut pos = 0;
334 for (i, entry) in entries.iter().enumerate() {
335 let matched = added.get(pos).is_some_and(|song| same_song(entry, song));
336 if i == index {
337 // The entry was skipped: play whatever followed it instead.
338 return (pos < added.len()).then_some(pos);
339 }
340 if matched {
341 pos += 1;
342 }
343 }
344 None
345}
346
347/// Playlists may store absolute paths, which `load` turns into paths
348/// relative to the music directory.
349fn same_song(entry: &str, queued: &str) -> bool {
350 entry == queued
351 || entry
352 .strip_suffix(queued)
353 .is_some_and(|prefix| prefix.ends_with('/'))
354}
355
356async fn queue_length(conn: &mut Connection) -> MpdResult<u32> {
357 let status = conn.execute(&Command::new("status")).await?;
358 Ok(Status::from_pairs(&status).playlist_length)
359}
360
361async fn run_query(conn: &mut Connection, query: &Query) -> MpdResult<Reply> {
362 Ok(match query {
363 Query::Status => {
364 let responses = conn
365 .execute_list(&[Command::new("status"), Command::new("currentsong")])
366 .await?;
367 let [status, current] = <[_; 2]>::try_from(responses)
368 .map_err(|_| MpdError::Protocol("short command list response".into()))?;
369 let current = (!current.is_empty()).then(|| Song::from_pairs(&current));
370 Reply::Status {
371 status: Status::from_pairs(&status),
372 current,
373 }
374 }
375 Query::QueueChanges { since } => {
376 let responses = conn
377 .execute_list(&[
378 Command::new("status"),
379 Command::new("plchanges").arg(since.to_string()),
380 ])
381 .await?;
382 let [status, changes] = <[_; 2]>::try_from(responses)
383 .map_err(|_| MpdError::Protocol("short command list response".into()))?;
384 Reply::QueueChanges {
385 status: Status::from_pairs(&status),
386 since: *since,
387 changes: parse_songs(&changes),
388 }
389 }
390 Query::Playlists => Reply::Playlists(parse_playlists(
391 &conn.execute(&Command::new("listplaylists")).await?,
392 )),
393 Query::Directory(path) => {
394 let entries = parse_entries(&conn.execute(&Command::new("lsinfo").arg(path)).await?);
395 Reply::Directory {
396 path: path.clone(),
397 entries,
398 }
399 }
400 Query::PlaylistSongs(name) => {
401 let pairs = conn
402 .execute(&Command::new("listplaylistinfo").arg(name))
403 .await?;
404 Reply::PlaylistSongs {
405 name: name.clone(),
406 songs: parse_songs(&pairs),
407 }
408 }
409 Query::TagValues(tag) => {
410 let pairs = conn.execute(&Command::new("list").arg(tag)).await?;
411 Reply::TagValues {
412 tag: tag.clone(),
413 values: parse_tag_values(&pairs, tag),
414 }
415 }
416 Query::Albums { tag, value } => {
417 let pairs = conn
418 .execute(
419 &Command::new("list")
420 .arg("Album")
421 .arg(super::protocol::filter_eq(tag, value))
422 .arg("group")
423 .arg("Date"),
424 )
425 .await?;
426 Reply::Albums {
427 tag: tag.clone(),
428 value: value.clone(),
429 albums: parse_albums(&pairs),
430 }
431 }
432 Query::Find(filter) => {
433 let mut songs = parse_songs(&conn.execute(&Command::new("find").arg(filter)).await?);
434 songs.sort_by_key(Song::disc_track);
435 Reply::Find {
436 filter: filter.clone(),
437 songs,
438 }
439 }
440 Query::MusicDirectory => {
441 let dir = match conn.execute(&Command::new("config")).await {
442 Ok(pairs) => pairs
443 .into_iter()
444 .find(|(key, _)| key == "music_directory")
445 .map(|(_, dir)| PathBuf::from(dir)),
446 // `config` is only permitted on local (Unix socket) connections.
447 Err(MpdError::Ack(_)) => None,
448 Err(error) => return Err(error),
449 };
450 Reply::MusicDirectory(dir)
451 }
452 })
453}
454
455#[cfg(test)]
456mod tests {
457 use super::queued_offset;
458
459 #[test]
460 fn queued_offset_skips_dropped_entries() {
461 let entries = ["gone.mp3", "a.mp3", "b.mp3", "missing.mp3", "c.mp3"];
462 let added = ["a.mp3", "b.mp3", "c.mp3"];
463 assert_eq!(queued_offset(&entries, &added, 1), Some(0));
464 assert_eq!(queued_offset(&entries, &added, 2), Some(1));
465 assert_eq!(queued_offset(&entries, &added, 4), Some(2));
466 // A dropped entry plays the song after it.
467 assert_eq!(queued_offset(&entries, &added, 0), Some(0));
468 assert_eq!(queued_offset(&entries, &added, 3), Some(2));
469 }
470
471 #[test]
472 fn queued_offset_matches_absolute_paths() {
473 let entries = ["/music/x.mp3", "gone.mp3", "/music/y.mp3"];
474 let added = ["x.mp3", "y.mp3"];
475 assert_eq!(queued_offset(&entries, &added, 2), Some(1));
476 }
477}