Krolyxon
Commit 222ab2f

Branches

Tags

Refs →

rmptuisrcmpdworker.rs

475 lines · 15.3 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.execute(&Command::new("listplaylist").arg(name)).await?;
300 conn.execute(&commands::load(name)).await?;
301 // `load` silently skips entries the server cannot resolve, so the
302 // playlist index is not necessarily the queue offset.
303 let added = conn
304 .execute(&Command::new("playlistinfo").arg(format!("{start}:")))
305 .await?;
306 let entries: Vec<&str> = files(&entries).collect();
307 let added: Vec<&str> = files(&added).collect();
308 let Some(pos) = queued_offset(&entries, &added, *offset as usize) else {
309 return Ok(None);
310 };
311 conn.execute(&commands::play_pos(start + pos as u32))
312 .await
313 .map(|_| None)
314 }
315 }
316}
317
318fn files(pairs: &[Pair]) -> impl Iterator<Item = &str> {
319 pairs
320 .iter()
321 .filter(|(key, _)| key == "file")
322 .map(|(_, value)| value.as_str())
323}
324
325/// Where playlist entry `index` ended up among the songs `load` appended,
326/// matching entries in order and skipping the ones that were dropped.
327fn queued_offset(entries: &[&str], added: &[&str], index: usize) -> Option<usize> {
328 if entries.len() == added.len() {
329 return (index < added.len()).then_some(index);
330 }
331 let mut pos = 0;
332 for (i, entry) in entries.iter().enumerate() {
333 let matched = added.get(pos).is_some_and(|song| same_song(entry, song));
334 if i == index {
335 // The entry was skipped: play whatever followed it instead.
336 return (pos < added.len()).then_some(pos);
337 }
338 if matched {
339 pos += 1;
340 }
341 }
342 None
343}
344
345/// Playlists may store absolute paths, which `load` turns into paths
346/// relative to the music directory.
347fn same_song(entry: &str, queued: &str) -> bool {
348 entry == queued
349 || entry
350 .strip_suffix(queued)
351 .is_some_and(|prefix| prefix.ends_with('/'))
352}
353
354async fn queue_length(conn: &mut Connection) -> MpdResult<u32> {
355 let status = conn.execute(&Command::new("status")).await?;
356 Ok(Status::from_pairs(&status).playlist_length)
357}
358
359async fn run_query(conn: &mut Connection, query: &Query) -> MpdResult<Reply> {
360 Ok(match query {
361 Query::Status => {
362 let responses = conn
363 .execute_list(&[Command::new("status"), Command::new("currentsong")])
364 .await?;
365 let [status, current] = <[_; 2]>::try_from(responses)
366 .map_err(|_| MpdError::Protocol("short command list response".into()))?;
367 let current = (!current.is_empty()).then(|| Song::from_pairs(&current));
368 Reply::Status {
369 status: Status::from_pairs(&status),
370 current,
371 }
372 }
373 Query::QueueChanges { since } => {
374 let responses = conn
375 .execute_list(&[
376 Command::new("status"),
377 Command::new("plchanges").arg(since.to_string()),
378 ])
379 .await?;
380 let [status, changes] = <[_; 2]>::try_from(responses)
381 .map_err(|_| MpdError::Protocol("short command list response".into()))?;
382 Reply::QueueChanges {
383 status: Status::from_pairs(&status),
384 since: *since,
385 changes: parse_songs(&changes),
386 }
387 }
388 Query::Playlists => Reply::Playlists(parse_playlists(
389 &conn.execute(&Command::new("listplaylists")).await?,
390 )),
391 Query::Directory(path) => {
392 let entries = parse_entries(&conn.execute(&Command::new("lsinfo").arg(path)).await?);
393 Reply::Directory {
394 path: path.clone(),
395 entries,
396 }
397 }
398 Query::PlaylistSongs(name) => {
399 let pairs = conn
400 .execute(&Command::new("listplaylistinfo").arg(name))
401 .await?;
402 Reply::PlaylistSongs {
403 name: name.clone(),
404 songs: parse_songs(&pairs),
405 }
406 }
407 Query::TagValues(tag) => {
408 let pairs = conn.execute(&Command::new("list").arg(tag)).await?;
409 Reply::TagValues {
410 tag: tag.clone(),
411 values: parse_tag_values(&pairs, tag),
412 }
413 }
414 Query::Albums { tag, value } => {
415 let pairs = conn
416 .execute(
417 &Command::new("list")
418 .arg("Album")
419 .arg(super::protocol::filter_eq(tag, value))
420 .arg("group")
421 .arg("Date"),
422 )
423 .await?;
424 Reply::Albums {
425 tag: tag.clone(),
426 value: value.clone(),
427 albums: parse_albums(&pairs),
428 }
429 }
430 Query::Find(filter) => {
431 let mut songs = parse_songs(&conn.execute(&Command::new("find").arg(filter)).await?);
432 songs.sort_by_key(Song::disc_track);
433 Reply::Find {
434 filter: filter.clone(),
435 songs,
436 }
437 }
438 Query::MusicDirectory => {
439 let dir = match conn.execute(&Command::new("config")).await {
440 Ok(pairs) => pairs
441 .into_iter()
442 .find(|(key, _)| key == "music_directory")
443 .map(|(_, dir)| PathBuf::from(dir)),
444 // `config` is only permitted on local (Unix socket) connections.
445 Err(MpdError::Ack(_)) => None,
446 Err(error) => return Err(error),
447 };
448 Reply::MusicDirectory(dir)
449 }
450 })
451}
452
453#[cfg(test)]
454mod tests {
455 use super::queued_offset;
456
457 #[test]
458 fn queued_offset_skips_dropped_entries() {
459 let entries = ["gone.mp3", "a.mp3", "b.mp3", "missing.mp3", "c.mp3"];
460 let added = ["a.mp3", "b.mp3", "c.mp3"];
461 assert_eq!(queued_offset(&entries, &added, 1), Some(0));
462 assert_eq!(queued_offset(&entries, &added, 2), Some(1));
463 assert_eq!(queued_offset(&entries, &added, 4), Some(2));
464 // A dropped entry plays the song after it.
465 assert_eq!(queued_offset(&entries, &added, 0), Some(0));
466 assert_eq!(queued_offset(&entries, &added, 3), Some(2));
467 }
468
469 #[test]
470 fn queued_offset_matches_absolute_paths() {
471 let entries = ["/music/x.mp3", "gone.mp3", "/music/y.mp3"];
472 let added = ["x.mp3", "y.mp3"];
473 assert_eq!(queued_offset(&entries, &added, 2), Some(1));
474 }
475}