Fixed "about to finish" message sending improperly

Cargo.toml

@@ -27,3 +27,5 @@ snap = "1.1.0"
 rcue = "0.1.3"
 gstreamer = "0.21.2"
 glib = "0.18.3"
+crossbeam-channel = "0.5.8"
+crossbeam = "0.8.2"

src/music_player.rs

@@ -2,8 +2,9 @@
 //use crate::music_controller::config::Config;
 use crate::music_storage::music_db::URI;
 use std::error::Error;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 use std::sync::mpsc::{self, Receiver, Sender};
+use crossbeam_channel::bounded;
 
 // GStreamer things
 use glib::{FlagsClass, MainContext};
@@ -26,24 +27,32 @@ pub enum PlayerCmd {
 pub struct Player {
     source: Option<URI>,
     //pub message_tx: Sender<PlayerCmd>,
-    pub message_rx: Receiver<PlayerCmd>,
+    pub message_rx: crossbeam::channel::Receiver<PlayerCmd>,
-    playbin: Element,
+    playbin: Arc<RwLock<Element>>,
     paused: bool,
     volume: f64,
-    start: Option<Duration>,
+    start: Arc<RwLock<Option<Duration>>>,
-    end: Option<Duration>,
+    end: Arc<RwLock<Option<Duration>>>,
-    position: Option<Duration>,
+    pub position: Arc<RwLock<Option<Duration>>>,
     gapless: bool,
 }
 
+impl Default for Player {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl Player {
     pub fn new() -> Self {
         // Initialize GStreamer
         gst::init().unwrap();
 
-        let playbin = gst::ElementFactory::make("playbin").build().unwrap();
+        let playbin_arc = Arc::new(RwLock::new(gst::ElementFactory::make("playbin3").build().unwrap()));
 
-        let flags = playbin.property_value("flags");
+        let playbin = playbin_arc.clone();
+
+        let flags = playbin.read().unwrap().property_value("flags");
         let flags_class = FlagsClass::with_type(flags.type_()).unwrap();
 
         // Set up the Playbin flags to only play audio
@@ -56,15 +65,61 @@ impl Player {
             .unset_by_nick("text")
             .build()
             .unwrap();
-        playbin.set_property_from_value("flags", &flags);
 
+        playbin.write().unwrap().set_property_from_value("flags", &flags);
 
-        let (message_tx, message_rx) = std::sync::mpsc::channel();
+        let position = Arc::new(RwLock::new(None));
-        playbin.connect("about-to-finish", false, move |_| {
+        let start = Arc::new(RwLock::new(None));
-                println!("test");
+        let end: Arc<RwLock<Option<Duration>>> = Arc::new(RwLock::new(None));
-                message_tx.send(PlayerCmd::AboutToFinish).unwrap();
+
-                None
+        let position_update = position.clone();
-            });
+        let start_update = start.clone();
+        let end_update = end.clone();
+        let (message_tx, message_rx) = bounded(1);
+        std::thread::spawn(move || {
+            loop {
+                let mut pos_temp = playbin_arc
+                    .read()
+                    .unwrap()
+                    .query_position::<ClockTime>()
+                    .map(|pos| Duration::nanoseconds(pos.nseconds() as i64));
+
+                if pos_temp.is_some()
+                    && start_update.read().unwrap().is_some()
+                    && end_update.read().unwrap().is_some()
+                {
+                    if let Some(time) = *start_update.read().unwrap() { pos_temp = Some(pos_temp.unwrap() - time) }
+
+                    let atf = end_update.read().unwrap().unwrap() - Duration::milliseconds(100);
+                    if pos_temp.unwrap() >= end_update.read().unwrap().unwrap() {
+                        message_tx.try_send(PlayerCmd::Eos).unwrap();
+                        playbin_arc
+                            .write()
+                            .unwrap()
+                            .set_state(gst::State::Ready)
+                            .expect("Unable to set the pipeline state");
+                        *start_update.write().unwrap() = None;
+                        *end_update.write().unwrap() = None;
+                    } else if pos_temp.unwrap() >= atf {
+                        match message_tx.try_send(PlayerCmd::AboutToFinish) {
+                            Ok(_) => println!("Sent ATF"),
+                            Err(err) => println!("{}", err),
+                        }
+                    }
+                }
+
+                *position_update.write().unwrap() = pos_temp;
+
+                std::thread::sleep(std::time::Duration::from_millis(50));
+            }
+        });
+
+        /*
+        playbin.read().unwrap().connect("about-to-finish", false, move |_| {
+            //message_tx.send(PlayerCmd::AboutToFinish).unwrap();
+            None
+        });
+        */
 
         let source = None;
         Self {
@@ -72,11 +127,11 @@ impl Player {
             playbin,
             message_rx,
             paused: false,
-            volume: 0.5,
+            volume: 1.0,
             gapless: false,
-            start: None,
+            start,
-            end: None,
+            end,
-            position: None,
+            position,
         }
     }
 
@@ -84,27 +139,48 @@ impl Player {
         &self.source
     }
 
-    pub fn enqueue_next(&mut self, next_track: URI) {
+    pub fn enqueue_next(&mut self, next_track: &URI) {
-        self.set_state(gst::State::Ready);
+        self.ready();
-
         self.set_source(next_track);
-
         self.play();
     }
 
     /// Set the playback URI
-    pub fn set_source(&mut self, source: URI) {
+    fn set_source(&mut self, source: &URI) {
         self.source = Some(source.clone());
         match source {
-            URI::Cue {start, ..} => {
+            URI::Cue {start, end, ..} => {
-                self.playbin.set_property("uri", source.as_uri());
+                self.playbin.write().unwrap().set_property("uri", source.as_uri());
-                self.play();
+
-                while self.state() != gst::State::Playing {
+                // Set the start and end positions of the CUE file
-                    std::thread::sleep(std::time::Duration::from_millis(10));
+                *self.start.write().unwrap() = Some(Duration::from_std(*start).unwrap());
+                *self.end.write().unwrap() = Some(Duration::from_std(*end).unwrap());
+
+                self.pause();
+
+                // Wait for it to be ready, and then move to the proper position
+                while self.playbin.read().unwrap().query_duration::<ClockTime>().is_none() {
+                    std::thread::sleep(std::time::Duration::from_millis(1));
                 };
-                self.seek_to(Duration::from_std(start).unwrap()).unwrap();
+
-            }
+                self.seek_to(Duration::from_std(*start).unwrap()).unwrap();
-            _ => self.playbin.set_property("uri", source.as_uri()),
+            },
+            _ => {
+                self.playbin.write().unwrap().set_property("uri", source.as_uri());
+
+                self.pause();
+
+                while self.playbin.read().unwrap().query_duration::<ClockTime>().is_none() {
+                    std::thread::sleep(std::time::Duration::from_millis(1));
+                };
+
+                *self.start.write().unwrap() = Some(Duration::seconds(0));
+                *self.end.write().unwrap() = self.playbin
+                    .read()
+                    .unwrap()
+                    .query_duration::<ClockTime>()
+                    .map(|pos| Duration::nanoseconds(pos.nseconds() as i64));
+            },
         }
     }
 
@@ -117,7 +193,7 @@ impl Player {
     /// Set volume of the internal playbin player, can be
     /// used to bypass the main volume control for seeking
     fn set_gstreamer_volume(&mut self, volume: f64) {
-        self.playbin.set_property("volume", volume)
+        self.playbin.write().unwrap().set_property("volume", volume)
     }
 
     /// Returns the current volume level, a float from 0 to 1
@@ -127,10 +203,16 @@ impl Player {
 
     fn set_state(&mut self, state: gst::State) {
         self.playbin
+            .write()
+            .unwrap()
             .set_state(state)
             .expect("Unable to set the pipeline state");
     }
 
+    pub fn ready(&mut self) {
+        self.set_state(gst::State::Ready)
+    }
+
     /// If the player is paused or stopped, starts playback
     pub fn play(&mut self) {
         self.set_state(gst::State::Playing);
@@ -150,30 +232,30 @@ impl Player {
 
     /// Check if playback is paused
     pub fn is_paused(&mut self) -> bool {
-        self.playbin.current_state() == gst::State::Paused
+        self.playbin.read().unwrap().current_state() == gst::State::Paused
     }
 
     /// Get the current playback position of the player
     pub fn position(&mut self) -> Option<Duration> {
-        self.position = self
+        *self.position.read().unwrap()
-            .playbin
-            .query_position::<ClockTime>()
-            .map(|pos| Duration::nanoseconds(pos.nseconds() as i64));
-        self.position
     }
 
     /// Get the duration of the currently playing track
     pub fn duration(&mut self) -> Option<Duration> {
-        if self.end.is_some() && self.start.is_some() {
+        if self.end.read().unwrap().is_some() && self.start.read().unwrap().is_some() {
-            Some(self.end.unwrap() - self.start.unwrap())
+            Some(self.end.read().unwrap().unwrap() - self.start.read().unwrap().unwrap())
         } else {
-            None
+            self.playbin
+                .read()
+                .unwrap()
+                .query_duration::<ClockTime>()
+                .map(|pos| Duration::nanoseconds(pos.nseconds() as i64))
         }
     }
 
     /// Seek relative to the current position
     pub fn seek_by(&mut self, seek_amount: Duration) -> Result<(), Box<dyn Error>> {
-        let time_pos = match self.position() {
+        let time_pos = match *self.position.read().unwrap() {
             Some(pos) => pos,
             None => return Err("No position".into()),
         };
@@ -188,13 +270,15 @@ impl Player {
         let seek_pos_clock = ClockTime::from_useconds(target_pos.num_microseconds().unwrap() as u64);
         self.set_gstreamer_volume(0.0);
         self.playbin
+            .write()
+            .unwrap()
             .seek_simple(gst::SeekFlags::FLUSH, seek_pos_clock)?;
         self.set_gstreamer_volume(self.volume);
         Ok(())
     }
 
     pub fn state(&mut self) -> gst::State {
-        self.playbin.current_state()
+        self.playbin.read().unwrap().current_state()
     }
 }
 
@@ -202,6 +286,8 @@ impl Drop for Player {
     /// Cleans up `GStreamer` pipeline when `Backend` is dropped.
     fn drop(&mut self) {
         self.playbin
+            .write()
+            .unwrap()
             .set_state(gst::State::Null)
             .expect("Unable to set the pipeline to the `Null` state");
     }

src/music_storage/music_db.rs

@@ -76,6 +76,7 @@ impl ToString for Tag {
 }
 
 /// A field within a Song struct
+#[derive(Debug)]
 pub enum Field {
     Location(URI),
     Plays(i32),
@@ -399,7 +400,7 @@ impl MusicLibrary {
         &mut self,
         target_path: &str,
         config: &Config,
-    ) -> Result<usize, Box<dyn std::error::Error>> {
+    ) -> Result<i32, Box<dyn std::error::Error>> {
         let mut total = 0;
         let mut errors = 0;
         for target_file in WalkDir::new(target_path)
@@ -568,7 +569,7 @@ impl MusicLibrary {
         Ok(())
     }
 
-    pub fn add_cuesheet(&mut self, cuesheet: &Path) -> Result<usize, Box<dyn Error>> {
+    pub fn add_cuesheet(&mut self, cuesheet: &Path) -> Result<i32, Box<dyn Error>> {
         let mut tracks_added = 0;
 
         let cue_data = parse_from_file(&cuesheet.to_string_lossy(), false).unwrap();
@@ -602,7 +603,13 @@ impl MusicLibrary {
                     Some(postgap) => postgap,
                     None => Duration::from_secs(0),
                 };
-                let mut start = track.indices[0].1;
+
+                let mut start;
+                if track.indices.len() > 1 {
+                    start = track.indices[1].1;
+                } else {
+                    start = track.indices[0].1;
+                }
                 if !start.is_zero() {
                     start -= pregap;
                 }