Krolyxon
Commit 61cb83e

Branches

Tags

Refs →

rmptuisrcmpdworker.rs

403 lines · 12.7 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,
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 conn.execute_all(&[commands::load(name), commands::play_pos(start + offset)])
300 .await
301 .map(|()| None)
302 }
303 }
304}
305
306async fn queue_length(conn: &mut Connection) -> MpdResult<u32> {
307 let status = conn.execute(&Command::new("status")).await?;
308 Ok(Status::from_pairs(&status).playlist_length)
309}
310
311async fn run_query(conn: &mut Connection, query: &Query) -> MpdResult<Reply> {
312 Ok(match query {
313 Query::Status => {
314 let responses = conn
315 .execute_list(&[Command::new("status"), Command::new("currentsong")])
316 .await?;
317 let [status, current] = <[_; 2]>::try_from(responses)
318 .map_err(|_| MpdError::Protocol("short command list response".into()))?;
319 let current = (!current.is_empty()).then(|| Song::from_pairs(&current));
320 Reply::Status {
321 status: Status::from_pairs(&status),
322 current,
323 }
324 }
325 Query::QueueChanges { since } => {
326 let responses = conn
327 .execute_list(&[
328 Command::new("status"),
329 Command::new("plchanges").arg(since.to_string()),
330 ])
331 .await?;
332 let [status, changes] = <[_; 2]>::try_from(responses)
333 .map_err(|_| MpdError::Protocol("short command list response".into()))?;
334 Reply::QueueChanges {
335 status: Status::from_pairs(&status),
336 since: *since,
337 changes: parse_songs(&changes),
338 }
339 }
340 Query::Playlists => Reply::Playlists(parse_playlists(
341 &conn.execute(&Command::new("listplaylists")).await?,
342 )),
343 Query::Directory(path) => {
344 let entries = parse_entries(&conn.execute(&Command::new("lsinfo").arg(path)).await?);
345 Reply::Directory {
346 path: path.clone(),
347 entries,
348 }
349 }
350 Query::PlaylistSongs(name) => {
351 let pairs = conn
352 .execute(&Command::new("listplaylistinfo").arg(name))
353 .await?;
354 Reply::PlaylistSongs {
355 name: name.clone(),
356 songs: parse_songs(&pairs),
357 }
358 }
359 Query::TagValues(tag) => {
360 let pairs = conn.execute(&Command::new("list").arg(tag)).await?;
361 Reply::TagValues {
362 tag: tag.clone(),
363 values: parse_tag_values(&pairs, tag),
364 }
365 }
366 Query::Albums { tag, value } => {
367 let pairs = conn
368 .execute(
369 &Command::new("list")
370 .arg("Album")
371 .arg(super::protocol::filter_eq(tag, value))
372 .arg("group")
373 .arg("Date"),
374 )
375 .await?;
376 Reply::Albums {
377 tag: tag.clone(),
378 value: value.clone(),
379 albums: parse_albums(&pairs),
380 }
381 }
382 Query::Find(filter) => {
383 let mut songs = parse_songs(&conn.execute(&Command::new("find").arg(filter)).await?);
384 songs.sort_by_key(Song::disc_track);
385 Reply::Find {
386 filter: filter.clone(),
387 songs,
388 }
389 }
390 Query::MusicDirectory => {
391 let dir = match conn.execute(&Command::new("config")).await {
392 Ok(pairs) => pairs
393 .into_iter()
394 .find(|(key, _)| key == "music_directory")
395 .map(|(_, dir)| PathBuf::from(dir)),
396 // `config` is only permitted on local (Unix socket) connections.
397 Err(MpdError::Ack(_)) => None,
398 Err(error) => return Err(error),
399 };
400 Reply::MusicDirectory(dir)
401 }
402 })
403}