Krolyxon
Commit 7f77734

Branches

Tags

Refs →

rmptuisrcconnection.rs

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