Krolyxon
Commit ad32349

Branches

Tags

Refs →

rmptuisrcconnection.rs

741 lines · 22.5 KB · Rust

Raw Blame History Download
1use std::{
2 collections::HashMap,
3 error::Error,
4 fmt,
5 time::{Duration, Instant},
6};
7
8use tokio::{
9 io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
10 net::TcpStream,
11};
12
13use crate::AppResult;
14
15#[derive(Clone, Debug, Default, PartialEq, Eq)]
16pub struct Song {
17 pub file: String,
18 pub title: Option<String>,
19 pub artist: Option<String>,
20 pub tags: Vec<(String, String)>,
21 pub duration: Option<Duration>,
22 pub place: Option<Place>,
23}
24
25#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
26pub struct Place {
27 pub pos: u32,
28}
29
30#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
31pub enum PlaybackState {
32 Playing,
33 Paused,
34 #[default]
35 Stopped,
36}
37
38#[derive(Clone, Debug, Default)]
39pub struct Status {
40 pub state: PlaybackState,
41 pub volume: i8,
42 pub repeat: bool,
43 pub random: bool,
44 pub song: Option<Place>,
45 pub elapsed: Duration,
46 pub total: Duration,
47 pub updating_db: Option<u32>,
48}
49
50#[derive(Clone, Debug, Default)]
51pub struct Stats {
52 pub songs: u32,
53}
54
55#[derive(Clone, Copy, Debug)]
56pub enum VolumeStatus {
57 Muted(i8),
58 Unmuted,
59}
60
61#[derive(Clone, Debug)]
62pub struct MpdSnapshot {
63 pub status: Status,
64 pub stats: Stats,
65 pub current_song: Song,
66 pub queue: Vec<Song>,
67}
68
69#[derive(Debug)]
70struct MpdError(String);
71
72impl fmt::Display for MpdError {
73 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
74 f.write_str(&self.0)
75 }
76}
77
78impl Error for MpdError {}
79
80#[derive(Debug)]
81pub struct Connection {
82 stream: BufReader<TcpStream>,
83 pub current_song: Song,
84 pub stats: Stats,
85 pub status: Status,
86 pub volume_status: VolumeStatus,
87 last_status_update: Instant,
88}
89
90impl Connection {
91 async fn open(address: &str, password: Option<&str>) -> AppResult<Self> {
92 let stream = TcpStream::connect(address).await?;
93 let mut client = Self {
94 stream: BufReader::new(stream),
95 current_song: Song::default(),
96 stats: Stats::default(),
97 status: Status::default(),
98 volume_status: VolumeStatus::Unmuted,
99 last_status_update: Instant::now(),
100 };
101 let mut greeting = String::new();
102 client.stream.read_line(&mut greeting).await?;
103 if !greeting.starts_with("OK MPD ") {
104 return Err(Box::new(MpdError(format!(
105 "unexpected MPD greeting: {}",
106 greeting.trim_end()
107 ))));
108 }
109 if let Some(password) = password {
110 client.command("password", &[password]).await?;
111 }
112 Ok(client)
113 }
114
115 async fn command(&mut self, name: &str, args: &[&str]) -> AppResult<Vec<String>> {
116 let mut request = name.to_string();
117 for arg in args {
118 request.push(' ');
119 request.push_str(&quote(arg));
120 }
121 request.push('\n');
122 self.stream.get_mut().write_all(request.as_bytes()).await?;
123 self.stream.get_mut().flush().await?;
124
125 let mut lines = Vec::new();
126 loop {
127 let mut line = String::new();
128 let read = self.stream.read_line(&mut line).await?;
129 if read == 0 {
130 return Err(Box::new(MpdError(
131 "MPD closed the connection while replying to a command".to_string(),
132 )));
133 }
134 let line = line.trim_end_matches(['\r', '\n']);
135 if line == "OK" {
136 return Ok(lines);
137 }
138 if line.starts_with("ACK ") {
139 return Err(Box::new(MpdError(format!("MPD rejected `{name}`: {line}"))));
140 }
141 lines.push(line.to_string());
142 }
143 }
144
145 async fn read_status(&mut self) -> AppResult<Status> {
146 let fields = fields(self.command("status", &[]).await?);
147 let time = fields.get("time").and_then(|s| {
148 let (elapsed, total) = s.split_once(':')?;
149 Some((parse_duration(elapsed), parse_duration(total)))
150 });
151 let elapsed = fields
152 .get("elapsed")
153 .map(|s| parse_duration(s))
154 .or_else(|| time.map(|(elapsed, _)| elapsed))
155 .unwrap_or_default();
156 let total = fields
157 .get("duration")
158 .map(|s| parse_duration(s))
159 .or_else(|| time.map(|(_, total)| total))
160 .unwrap_or_default();
161
162 Ok(Status {
163 state: match fields.get("state").map(String::as_str) {
164 Some("play") => PlaybackState::Playing,
165 Some("pause") => PlaybackState::Paused,
166 _ => PlaybackState::Stopped,
167 },
168 volume: fields
169 .get("volume")
170 .and_then(|s| s.parse().ok())
171 .unwrap_or(-1),
172 repeat: bool_field(&fields, "repeat"),
173 random: bool_field(&fields, "random"),
174 song: fields
175 .get("song")
176 .and_then(|s| s.parse().ok())
177 .map(|pos| Place { pos }),
178 elapsed,
179 total,
180 updating_db: fields.get("updating_db").and_then(|s| s.parse().ok()),
181 })
182 }
183
184 async fn read_stats(&mut self) -> AppResult<Stats> {
185 let fields = fields(self.command("stats", &[]).await?);
186 Ok(Stats {
187 songs: fields
188 .get("songs")
189 .and_then(|s| s.parse().ok())
190 .unwrap_or_default(),
191 })
192 }
193
194 async fn read_current_song(&mut self) -> AppResult<Song> {
195 Ok(parse_songs(self.command("currentsong", &[]).await?)
196 .into_iter()
197 .next()
198 .unwrap_or_default())
199 }
200
201 pub async fn queue(&mut self) -> AppResult<Vec<Song>> {
202 Ok(parse_songs(self.command("playlistinfo", &[]).await?))
203 }
204
205 pub async fn snapshot(&mut self) -> AppResult<MpdSnapshot> {
206 let status = self.read_status().await?;
207 let stats = self.read_stats().await?;
208 let current_song = self.read_current_song().await?;
209 let queue = self.queue().await?;
210 Ok(MpdSnapshot {
211 status,
212 stats,
213 current_song,
214 queue,
215 })
216 }
217
218 pub async fn list_files(&mut self, path: &str) -> AppResult<Vec<(String, String)>> {
219 let lines = if path.is_empty() || path == "." {
220 self.command("listfiles", &[]).await?
221 } else {
222 self.command("listfiles", &[path]).await?
223 };
224 Ok(lines
225 .into_iter()
226 .filter_map(|line| {
227 let (kind, value) = parse_field(&line)?;
228 matches!(kind.as_str(), "file" | "directory").then_some((kind, value))
229 })
230 .collect())
231 }
232
233 pub async fn ls_info(&mut self, path: &str) -> AppResult<(Vec<String>, Vec<Song>)> {
234 let lines = if path.is_empty() || path == "." {
235 self.command("lsinfo", &[]).await?
236 } else {
237 self.command("lsinfo", &[path]).await?
238 };
239 let mut directories = Vec::new();
240 let mut songs = Vec::new();
241 let mut current: Option<Song> = None;
242 for line in lines {
243 let Some((key, value)) = parse_field(&line) else {
244 continue;
245 };
246 match key.as_str() {
247 "directory" => directories.push(value),
248 "file" => {
249 if let Some(song) = current.take() {
250 songs.push(song);
251 }
252 current = Some(Song {
253 file: value,
254 ..Song::default()
255 });
256 }
257 "playlist" => {}
258 _ => {
259 if let Some(song) = current.as_mut() {
260 add_metadata(song, key, value);
261 }
262 }
263 }
264 }
265 if let Some(song) = current {
266 songs.push(song);
267 }
268 Ok((directories, songs))
269 }
270
271 pub async fn playlists(&mut self) -> AppResult<Vec<String>> {
272 Ok(self
273 .command("listplaylists", &[])
274 .await?
275 .into_iter()
276 .filter_map(|line| parse_field(&line))
277 .filter_map(|(key, value)| (key == "playlist").then_some(value))
278 .collect())
279 }
280
281 pub async fn playlist(&mut self, name: &str) -> AppResult<Vec<Song>> {
282 Ok(parse_songs(
283 self.command("listplaylistinfo", &[name]).await?,
284 ))
285 }
286
287 pub async fn add(&mut self, file: &str) -> AppResult<()> {
288 self.command("add", &[file]).await?;
289 Ok(())
290 }
291
292 pub async fn play(&mut self, position: Option<u32>) -> AppResult<()> {
293 match position {
294 Some(position) => self.command("play", &[&position.to_string()]).await?,
295 None => self.command("play", &[]).await?,
296 };
297 Ok(())
298 }
299
300 pub async fn set_repeat(&mut self, enabled: bool) -> AppResult<()> {
301 self.command("repeat", &[if enabled { "1" } else { "0" }])
302 .await?;
303 Ok(())
304 }
305
306 pub async fn set_random(&mut self, enabled: bool) -> AppResult<()> {
307 self.command("random", &[if enabled { "1" } else { "0" }])
308 .await?;
309 Ok(())
310 }
311
312 pub async fn set_volume(&mut self, volume: i8) -> AppResult<()> {
313 self.command("setvol", &[&volume.clamp(0, 100).to_string()])
314 .await?;
315 Ok(())
316 }
317
318 pub async fn seek(&mut self, position: u32, elapsed: Duration) -> AppResult<()> {
319 self.command(
320 "seek",
321 &[&position.to_string(), &elapsed.as_secs().to_string()],
322 )
323 .await?;
324 Ok(())
325 }
326
327 pub async fn next(&mut self) -> AppResult<()> {
328 self.command("next", &[]).await?;
329 Ok(())
330 }
331
332 pub async fn previous(&mut self) -> AppResult<()> {
333 self.command("previous", &[]).await?;
334 Ok(())
335 }
336
337 pub async fn clear(&mut self) -> AppResult<()> {
338 self.command("clear", &[]).await?;
339 Ok(())
340 }
341
342 pub async fn delete(&mut self, position: u32) -> AppResult<()> {
343 self.command("delete", &[&position.to_string()]).await?;
344 Ok(())
345 }
346
347 pub async fn swap(&mut self, first: u32, second: u32) -> AppResult<()> {
348 self.command("swap", &[&first.to_string(), &second.to_string()])
349 .await?;
350 Ok(())
351 }
352
353 pub async fn append_playlist(&mut self, name: &str) -> AppResult<()> {
354 self.command("load", &[name]).await?;
355 Ok(())
356 }
357
358 pub async fn delete_playlist_song(&mut self, playlist: &str, position: usize) -> AppResult<()> {
359 self.command("playlistdelete", &[playlist, &position.to_string()])
360 .await?;
361 Ok(())
362 }
363
364 pub async fn move_playlist_song(
365 &mut self,
366 playlist: &str,
367 from: usize,
368 to: usize,
369 ) -> AppResult<()> {
370 self.command(
371 "playlistmove",
372 &[playlist, &from.to_string(), &to.to_string()],
373 )
374 .await?;
375 Ok(())
376 }
377
378 pub async fn delete_playlist(&mut self, playlist: &str) -> AppResult<()> {
379 self.command("rm", &[playlist]).await?;
380 Ok(())
381 }
382
383 pub async fn rename_playlist(&mut self, from: &str, to: &str) -> AppResult<()> {
384 self.command("rename", &[from, to]).await?;
385 Ok(())
386 }
387
388 pub async fn update_database(&mut self) -> AppResult<()> {
389 self.command("rescan", &[]).await?;
390 Ok(())
391 }
392
393 pub async fn idle(&mut self) -> AppResult<()> {
394 self.command(
395 "idle",
396 &[
397 "database",
398 "stored_playlist",
399 "player",
400 "playlist",
401 "mixer",
402 "options",
403 ],
404 )
405 .await?;
406 Ok(())
407 }
408}
409
410impl Connection {
411 pub async fn connect(address: &str, password: Option<&str>) -> AppResult<Self> {
412 let mut connection = Self::open(address, password).await?;
413 connection.update_status().await?;
414 Ok(connection)
415 }
416
417 pub fn apply_snapshot(&mut self, status: Status, stats: Stats, current_song: Song) {
418 self.status = status;
419 self.stats = stats;
420 self.current_song = current_song;
421 self.last_status_update = Instant::now();
422 }
423
424 pub async fn update_status(&mut self) -> AppResult<()> {
425 self.status = self.read_status().await?;
426 self.stats = self.read_stats().await?;
427 self.current_song = self.read_current_song().await?;
428 self.last_status_update = Instant::now();
429 Ok(())
430 }
431
432 pub fn get_progress_ratio(&self) -> f64 {
433 let total = self.status.total.as_secs_f64();
434 if total <= 0.0 {
435 0.0
436 } else {
437 (self.current_elapsed().as_secs_f64() / total).clamp(0.0, 1.0)
438 }
439 }
440
441 pub fn current_elapsed(&self) -> Duration {
442 let elapsed = if self.status.state == PlaybackState::Playing {
443 self.status
444 .elapsed
445 .saturating_add(self.last_status_update.elapsed())
446 } else {
447 self.status.elapsed
448 };
449 if self.status.total.is_zero() {
450 elapsed
451 } else {
452 elapsed.min(self.status.total)
453 }
454 }
455
456 pub async fn push(&mut self, song: &Song) -> AppResult<()> {
457 self.add(&song.file).await?;
458 let queue = self.queue().await?;
459 if !queue.is_empty() {
460 self.play(Some(queue.len() as u32 - 1)).await?;
461 }
462 self.update_status().await
463 }
464
465 pub async fn load_playlist(&mut self, playlist: &str) -> AppResult<()> {
466 self.append_playlist(playlist).await?;
467 self.play(None).await?;
468 self.update_status().await
469 }
470
471 pub async fn add_to_playlist(&mut self, playlist: &str, song: &Song) -> AppResult<()> {
472 self.command("playlistadd", &[playlist, &song.file]).await?;
473 Ok(())
474 }
475
476 pub fn song_from_file(filename: &str) -> Song {
477 Song {
478 file: filename.to_string(),
479 ..Song::default()
480 }
481 }
482
483 pub fn now_playing(&self) -> String {
484 match (&self.current_song.title, &self.current_song.artist) {
485 (Some(title), Some(artist)) => format!("\"{title}\" By {artist}"),
486 (Some(title), None) => title.clone(),
487 _ if self.current_song.file.is_empty() => "No Song playing or in Queue".to_string(),
488 _ => self.current_song.file.clone(),
489 }
490 }
491
492 pub async fn toggle_pause(&mut self) -> AppResult<()> {
493 self.command("pause", &[]).await?;
494 self.update_status().await
495 }
496
497 pub async fn pause(&mut self) -> AppResult<()> {
498 self.command("pause", &["1"]).await?;
499 self.update_status().await
500 }
501
502 pub async fn toggle_repeat(&mut self) -> AppResult<()> {
503 self.set_repeat(!self.status.repeat).await?;
504 self.update_status().await
505 }
506
507 pub async fn toggle_random(&mut self) -> AppResult<()> {
508 self.set_random(!self.status.random).await?;
509 self.update_status().await
510 }
511
512 pub async fn inc_volume(&mut self, amount: i8) -> AppResult<()> {
513 let volume = self.status.volume.saturating_add(amount).min(100);
514 self.volume_status = VolumeStatus::Unmuted;
515 self.set_volume(volume).await?;
516 self.update_status().await
517 }
518
519 pub async fn dec_volume(&mut self, amount: i8) -> AppResult<()> {
520 let volume = self.status.volume.saturating_sub(amount).max(0);
521 self.volume_status = VolumeStatus::Unmuted;
522 self.set_volume(volume).await?;
523 self.update_status().await
524 }
525
526 pub async fn refresh_song_list(&mut self) -> AppResult<()> {
527 self.stats = self.read_stats().await?;
528 Ok(())
529 }
530}
531
532fn fields(lines: Vec<String>) -> HashMap<String, String> {
533 lines
534 .into_iter()
535 .filter_map(|line| parse_field(&line))
536 .collect()
537}
538
539fn parse_songs(lines: Vec<String>) -> Vec<Song> {
540 let mut songs = Vec::new();
541 let mut current: Option<Song> = None;
542 for line in lines {
543 let Some((key, value)) = parse_field(&line) else {
544 continue;
545 };
546 if key == "file" {
547 if let Some(song) = current.take() {
548 songs.push(song);
549 }
550 current = Some(Song {
551 file: value,
552 ..Song::default()
553 });
554 } else if let Some(song) = current.as_mut() {
555 if key.eq_ignore_ascii_case("pos") {
556 song.place = value.parse().ok().map(|pos| Place { pos });
557 } else if key == "duration" || key == "Time" {
558 song.duration = Some(parse_duration(&value));
559 } else {
560 add_metadata(song, key, value);
561 }
562 }
563 }
564 if let Some(song) = current {
565 songs.push(song);
566 }
567 songs
568}
569
570fn add_metadata(song: &mut Song, key: String, value: String) {
571 match key.to_ascii_lowercase().as_str() {
572 "title" => song.title = Some(value),
573 "artist" => song.artist = Some(value),
574 "duration" | "time" => song.duration = Some(parse_duration(&value)),
575 _ => song.tags.push((key, value)),
576 }
577}
578
579fn parse_field(line: &str) -> Option<(String, String)> {
580 let (key, value) = line.split_once(": ")?;
581 Some((key.to_string(), unquote(value)))
582}
583
584fn unquote(value: &str) -> String {
585 let value = value
586 .strip_prefix('"')
587 .and_then(|v| v.strip_suffix('"'))
588 .unwrap_or(value);
589 let mut result = String::with_capacity(value.len());
590 let mut escaped = false;
591 for character in value.chars() {
592 if escaped {
593 result.push(character);
594 escaped = false;
595 } else if character == '\\' {
596 escaped = true;
597 } else {
598 result.push(character);
599 }
600 }
601 if escaped {
602 result.push('\\');
603 }
604 result
605}
606
607fn quote(value: &str) -> String {
608 format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\""))
609}
610
611fn bool_field(fields: &HashMap<String, String>, key: &str) -> bool {
612 fields.get(key).is_some_and(|value| value == "1")
613}
614
615fn parse_duration(value: &str) -> Duration {
616 value
617 .parse::<f64>()
618 .ok()
619 .filter(|seconds| seconds.is_finite() && *seconds >= 0.0)
620 .and_then(|seconds| Duration::try_from_secs_f64(seconds).ok())
621 .unwrap_or_default()
622}
623
624#[cfg(test)]
625mod tests {
626 use super::*;
627 use tokio::{
628 io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
629 net::TcpListener,
630 };
631
632 #[test]
633 fn parses_quoted_mpd_values_and_escapes() {
634 assert_eq!(
635 parse_field(r#"Title: "A \"quoted\" song\\file""#),
636 Some(("Title".to_string(), "A \"quoted\" song\\file".to_string()))
637 );
638 assert_eq!(quote(r#"A "song"\path"#), r#""A \"song\"\\path""#);
639 }
640
641 #[test]
642 fn parses_songs_with_tags_and_queue_positions() {
643 let songs = parse_songs(vec![
644 "file: music/example.flac".to_string(),
645 "Title: Example".to_string(),
646 "Artist: Someone".to_string(),
647 "duration: 124.5".to_string(),
648 "Pos: 3".to_string(),
649 "file: music/another.ogg".to_string(),
650 ]);
651 assert_eq!(songs.len(), 2);
652 assert_eq!(songs[0].title.as_deref(), Some("Example"));
653 assert_eq!(songs[0].artist.as_deref(), Some("Someone"));
654 assert_eq!(songs[0].duration, Some(Duration::from_secs_f64(124.5)));
655 assert_eq!(songs[0].place, Some(Place { pos: 3 }));
656 assert_eq!(songs[1].file, "music/another.ogg");
657 }
658
659 #[tokio::test]
660 async fn sends_quoted_commands_over_the_mpd_tcp_protocol() {
661 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
662 let address = listener.local_addr().unwrap();
663 let server = tokio::spawn(async move {
664 let (stream, _) = listener.accept().await.unwrap();
665 let mut stream = BufReader::new(stream);
666 stream
667 .get_mut()
668 .write_all(b"OK MPD 0.23.0\n")
669 .await
670 .unwrap();
671 let mut command = String::new();
672 stream.read_line(&mut command).await.unwrap();
673 assert_eq!(command, "play \"0\"\n");
674 stream.get_mut().write_all(b"OK\n").await.unwrap();
675 });
676
677 let mut client = Connection::open(&address.to_string(), None).await.unwrap();
678 client.command("play", &["0"]).await.unwrap();
679 server.await.unwrap();
680 }
681
682 #[tokio::test]
683 async fn omits_empty_root_paths_from_mpd_commands() {
684 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
685 let address = listener.local_addr().unwrap();
686 let server = tokio::spawn(async move {
687 let (stream, _) = listener.accept().await.unwrap();
688 let mut stream = BufReader::new(stream);
689 stream
690 .get_mut()
691 .write_all(b"OK MPD 0.23.0\n")
692 .await
693 .unwrap();
694 for expected in ["lsinfo\n", "listfiles\n"] {
695 let mut command = String::new();
696 stream.read_line(&mut command).await.unwrap();
697 assert_eq!(command, expected);
698 stream.get_mut().write_all(b"OK\n").await.unwrap();
699 }
700 });
701
702 let mut client = Connection::open(&address.to_string(), None).await.unwrap();
703 client.ls_info("").await.unwrap();
704 client.list_files("").await.unwrap();
705 server.await.unwrap();
706 }
707
708 #[tokio::test]
709 async fn sends_stored_playlist_edit_commands() {
710 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
711 let address = listener.local_addr().unwrap();
712 let server = tokio::spawn(async move {
713 let (stream, _) = listener.accept().await.unwrap();
714 let mut stream = BufReader::new(stream);
715 stream
716 .get_mut()
717 .write_all(b"OK MPD 0.23.0\n")
718 .await
719 .unwrap();
720 for expected in [
721 "playlistdelete \"mix \\\"one\\\"\" \"2\"\n",
722 "playlistmove \"mix \\\"one\\\"\" \"1\" \"2\"\n",
723 "rm \"mix \\\"one\\\"\"\n",
724 ] {
725 let mut command = String::new();
726 stream.read_line(&mut command).await.unwrap();
727 assert_eq!(command, expected);
728 stream.get_mut().write_all(b"OK\n").await.unwrap();
729 }
730 });
731
732 let mut client = Connection::open(&address.to_string(), None).await.unwrap();
733 client.delete_playlist_song("mix \"one\"", 2).await.unwrap();
734 client
735 .move_playlist_song("mix \"one\"", 1, 2)
736 .await
737 .unwrap();
738 client.delete_playlist("mix \"one\"").await.unwrap();
739 server.await.unwrap();
740 }
741}