feat!: queue system

src/music/player.rs

@@ -1,106 +1,248 @@
 use std::{
+    collections::VecDeque,
     fs,
     io::{BufReader, Cursor},
     path::Path,
+    sync::Arc,
     time::Duration,
 };
 
 use rodio::{Decoder, Sink};
+use tokio::{
+    sync::{
+        mpsc::{self, Receiver},
+        Mutex,
+    },
+    task::JoinHandle,
+};
 
 use crate::{database::tracks::Track, proto::player::PlayerStatus};
 
+use super::queue::QueuedTrack;
+
 pub struct AudioPlayer {
-    sink: Sink,
-    currently_playing: Option<Track>,
+    sink: Arc<Mutex<Sink>>,
+    currently_playing: Arc<Mutex<Option<Track>>>,
+    queue: Arc<Mutex<VecDeque<QueuedTrack>>>,
 }
 
 impl AudioPlayer {
+    const WATCH_SLEEP_TIME: Duration = Duration::from_millis(20);
     pub fn new(sink: Sink) -> Self {
         sink.set_volume(0.5);
 
         Self {
-            sink,
-            currently_playing: None,
+            sink: Arc::new(Mutex::new(sink)),
+            currently_playing: Arc::new(Mutex::new(None)),
+            queue: Arc::new(Mutex::new(VecDeque::new())),
         }
     }
 
-    pub fn play_track<P>(&mut self, track: Track, path: P) -> Result<(), Box<dyn std::error::Error>>
+    pub fn start_watching(&mut self) -> (JoinHandle<()>, Receiver<Option<QueuedTrack>>) {
+        let sink = self.sink.clone();
+        let current_track = self.currently_playing.clone();
+        let queue = self.queue.clone();
+
+        let (tx, rx) = mpsc::channel::<Option<QueuedTrack>>(128);
+
+        (
+            tokio::spawn(async move {
+                loop {
+                    if let Some(current_track) = current_track.lock().await.as_mut() {
+                        if current_track.duration as u128 <= sink.lock().await.get_pos().as_millis()
+                        {
+                            if let Err(_) = tx.send(queue.lock().await.pop_front()).await {
+                                break;
+                            }
+                        }
+                    }
+                    tokio::time::sleep(Self::WATCH_SLEEP_TIME).await;
+                }
+            }),
+            rx,
+        )
+    }
+
+    pub async fn play_track<P>(
+        &mut self,
+        track: Track,
+        path: P,
+        clear_queue: bool,
+    ) -> Result<(), Box<dyn std::error::Error>>
     where
         P: AsRef<Path>,
     {
-        self.sink.clear();
+        let sink = self.sink.lock().await;
+
+        sink.clear();
+        if clear_queue {
+            self.queue.lock().await.clear();
+        }
 
         let file = BufReader::new(Cursor::new(fs::read(path)?));
 
         let source = Decoder::new(file)?;
 
-        self.currently_playing = Some(track);
-        self.sink.append(source);
+        *self.currently_playing.lock().await = Some(track);
+        sink.append(source);
 
-        self.sink.play();
+        sink.play();
 
         Ok(())
     }
 
-    pub fn resume(&self) {
-        self.sink.play();
+    pub async fn play_track_next(
+        &mut self,
+        track: Track,
+        path: &Path,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let sink = self.sink.lock().await;
+
+        let mut queue = self.queue.lock().await;
+
+        if sink.empty() && queue.is_empty() {
+            drop(queue);
+            drop(sink);
+            return self.play_track(track, path, false).await;
+        }
+
+        let queued_track = QueuedTrack::new(track, path.to_path_buf());
+
+        queue.push_front(queued_track);
+
+        Ok(())
     }
 
-    pub fn pause(&self) {
-        self.sink.pause();
+    pub async fn add_to_queue(
+        &mut self,
+        track: Track,
+        path: &Path,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let sink = self.sink.lock().await;
+        let mut queue = self.queue.lock().await;
+
+        if sink.empty() && queue.is_empty() {
+            drop(queue);
+            drop(sink);
+            return self.play_track(track, path, false).await;
+        }
+
+        let queued_track = QueuedTrack::new(track, path.to_path_buf());
+
+        queue.push_back(queued_track);
+
+        Ok(())
+    }
+
+    pub async fn skip_track(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+        let mut queue = self.queue.lock().await;
+
+        let Some(queued_track) = queue.pop_front() else {
+            drop(queue);
+            self.clear().await;
+
+            return Ok(());
+        };
+
+        drop(queue);
+
+        self.play_track(queued_track.track, queued_track.path, false)
+            .await
+    }
+
+    pub async fn skip_to_queue_index(
+        &mut self,
+        index: usize,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let mut queue = self.queue.lock().await;
+
+        let Some(queued_track) = queue.remove(index) else {
+            drop(queue);
+            self.clear().await;
+
+            return Ok(());
+        };
+
+        drop(queue);
+        self.play_track(queued_track.track, queued_track.path, false)
+            .await
+    }
+
+    pub async fn resume(&self) {
+        self.sink.lock().await.play();
+    }
+
+    pub async fn pause(&self) {
+        self.sink.lock().await.pause();
     }
 
     /// Toggles the player's pause state and returns the new value
-    pub fn toggle_pause(&self) -> bool {
-        if self.is_paused() {
-            self.resume();
+    pub async fn toggle_pause(&self) -> bool {
+        if self.is_paused().await {
+            self.resume().await;
             false
         } else {
-            self.pause();
+            self.pause().await;
             true
         }
     }
 
-    pub fn volume(&self) -> f32 {
-        self.sink.volume()
+    pub async fn volume(&self) -> f32 {
+        self.sink.lock().await.volume()
     }
 
-    pub fn set_volume(&self, value: f32) {
-        self.sink.set_volume(value);
+    pub async fn set_volume(&self, value: f32) {
+        self.sink.lock().await.set_volume(value);
     }
 
-    pub fn position(&self) -> u128 {
-        self.sink.get_pos().as_millis()
+    pub async fn position(&self) -> u128 {
+        self.sink.lock().await.get_pos().as_millis()
     }
 
-    pub fn set_position(&self, position: u64) {
-        let _ = self.sink.try_seek(Duration::from_millis(position));
+    pub async fn set_position(&self, position: u64) {
+        let _ = self
+            .sink
+            .lock()
+            .await
+            .try_seek(Duration::from_millis(position));
     }
 
-    pub fn is_paused(&self) -> bool {
-        self.sink.is_paused()
+    pub async fn is_paused(&self) -> bool {
+        self.sink.lock().await.is_paused()
     }
 
-    pub fn currently_playing(&self) -> Option<Track> {
-        self.currently_playing.clone()
+    pub async fn currently_playing(&self) -> Option<Track> {
+        self.currently_playing.lock().await.clone()
     }
 
-    pub fn get_snapshot(&self) -> StatusSnapshot {
+    pub async fn queue(&self) -> VecDeque<QueuedTrack> {
+        self.queue.lock().await.clone()
+    }
+
+    pub async fn clear(&mut self) {
+        self.queue.lock().await.clear();
+        self.sink.lock().await.clear();
+        *self.currently_playing.lock().await = None;
+    }
+
+    pub async fn get_snapshot(&self) -> StatusSnapshot {
         StatusSnapshot {
-            volume: self.volume(),
-            position: self.position(),
-            is_paused: self.is_paused(),
-            currently_playing: self.currently_playing(),
+            volume: self.volume().await,
+            position: self.position().await,
+            is_paused: self.is_paused().await,
+            currently_playing: self.currently_playing().await,
+            queue: self.queue().await,
         }
     }
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone)]
 pub struct StatusSnapshot {
     pub volume: f32,
     pub position: u128,
     pub is_paused: bool,
     pub currently_playing: Option<Track>,
+    pub queue: VecDeque<QueuedTrack>,
 }
 
 impl Into<PlayerStatus> for StatusSnapshot {
@@ -110,6 +252,7 @@ impl Into<PlayerStatus> for StatusSnapshot {
             is_paused: self.is_paused,
             progress: self.position as u64,
             currently_playing: self.currently_playing.clone().map(|t| t.into()),
+            queue: Vec::from_iter(self.queue.into_iter().map(|t| t.track.into())),
         }
     }
 }