Get barebones playback working.

music-player/server/Cargo.lock

@@ -19,6 +19,12 @@ dependencies = [
  "version_check",
 ]
 
+[[package]]
+name = "anyhow"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "224afbd727c3d6e4b90103ece64b8d1b67fbb1973b1046c2281eed3f3803f800"
+
 [[package]]
 name = "autocfg"
 version = "1.1.0"
@@ -68,6 +74,15 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dfb24e866b15a1af2a1b663f10c6b6b8f397a84aadb828f12e5b289ec23a3a3c"
 
+[[package]]
+name = "cfg-expr"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b0357a6402b295ca3a86bc148e84df46c02e41f41fef186bda662557ef6328aa"
+dependencies = [
+ "smallvec",
+]
+
 [[package]]
 name = "cfg-if"
 version = "1.0.0"
@@ -178,6 +193,28 @@ version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04909a7a7e4633ae6c4a9ab280aeb86da1236243a77b694a49eacd659a4bd3ac"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7acc85df6714c176ab5edf386123fafe217be88c0840ec11f199441134a074e2"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-macro"
+version = "0.3.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bdfb8ce053d86b91919aad980c220b1fb8401a9394410e1c289ed7e66b61835d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.25"
@@ -197,6 +234,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "197676987abd2f9cadff84926f410af1c183608d36641465df73ae8211dc65d6"
 dependencies = [
  "futures-core",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "pin-project-lite",
@@ -225,6 +263,113 @@ dependencies = [
  "wasi",
 ]
 
+[[package]]
+name = "gio-sys"
+version = "0.16.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e9b693b8e39d042a95547fc258a7b07349b1f0b48f4b2fa3108ba3c51c0b5229"
+dependencies = [
+ "glib-sys",
+ "gobject-sys",
+ "libc",
+ "system-deps",
+ "winapi",
+]
+
+[[package]]
+name = "glib"
+version = "0.16.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ddd4df61a866ed7259d6189b8bcb1464989a77f1d85d25d002279bbe9dd38b2f"
+dependencies = [
+ "bitflags",
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-task",
+ "futures-util",
+ "gio-sys",
+ "glib-macros",
+ "glib-sys",
+ "gobject-sys",
+ "libc",
+ "once_cell",
+ "smallvec",
+ "thiserror",
+]
+
+[[package]]
+name = "glib-macros"
+version = "0.16.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e084807350b01348b6d9dbabb724d1a0bb987f47a2c85de200e98e12e30733bf"
+dependencies = [
+ "anyhow",
+ "heck",
+ "proc-macro-crate",
+ "proc-macro-error",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "glib-sys"
+version = "0.16.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c61a4f46316d06bfa33a7ac22df6f0524c8be58e3db2d9ca99ccb1f357b62a65"
+dependencies = [
+ "libc",
+ "system-deps",
+]
+
+[[package]]
+name = "gobject-sys"
+version = "0.16.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3520bb9c07ae2a12c7f2fbb24d4efc11231c8146a86956413fb1a79bb760a0f1"
+dependencies = [
+ "glib-sys",
+ "libc",
+ "system-deps",
+]
+
+[[package]]
+name = "gstreamer"
+version = "0.19.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85fc926d081923c840403ec5ec3b2157a7cd236a2587c3031a4f0206f13ed500"
+dependencies = [
+ "bitflags",
+ "cfg-if",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "glib",
+ "gstreamer-sys",
+ "libc",
+ "muldiv",
+ "num-integer",
+ "num-rational",
+ "once_cell",
+ "option-operations",
+ "paste",
+ "pretty-hex",
+ "thiserror",
+]
+
+[[package]]
+name = "gstreamer-sys"
+version = "0.19.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "545f52ad8a480732cc4290fd65dfe42952c8ae374fe581831ba15981fedf18a4"
+dependencies = [
+ "glib-sys",
+ "gobject-sys",
+ "libc",
+ "system-deps",
+]
+
 [[package]]
 name = "h2"
 version = "0.3.15"
@@ -287,6 +432,12 @@ dependencies = [
  "http",
 ]
 
+[[package]]
+name = "heck"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
+
 [[package]]
 name = "hermit-abi"
 version = "0.2.6"
@@ -478,6 +629,12 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "muldiv"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "956787520e75e9bd233246045d19f42fb73242759cc57fba9611d940ae96d4b0"
+
 [[package]]
 name = "multipart"
 version = "0.18.0"
@@ -501,6 +658,7 @@ name = "music-player"
 version = "0.1.0"
 dependencies = [
  "flow",
+ "gstreamer",
  "id3",
  "mime",
  "mime_guess",
@@ -509,10 +667,41 @@ dependencies = [
  "thiserror",
  "tokio",
  "url",
+ "urlencoding",
  "uuid",
  "warp",
 ]
 
+[[package]]
+name = "num-integer"
+version = "0.1.45"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "225d3389fb3509a24c93f5c29eb6bde2586b98d9f016636dff58d7c6f7569cd9"
+dependencies = [
+ "autocfg",
+ "num-traits",
+]
+
+[[package]]
+name = "num-rational"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0638a1c9d0a3c0914158145bc76cff373a75a627e6ecbfb71cbe6f453a5a19b0"
+dependencies = [
+ "autocfg",
+ "num-integer",
+ "num-traits",
+]
+
+[[package]]
+name = "num-traits"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "578ede34cf02f8924ab9447f50c28075b4d3e5b269972345e7e0372b38c6cdcd"
+dependencies = [
+ "autocfg",
+]
+
 [[package]]
 name = "num_cpus"
 version = "1.15.0"
@@ -529,6 +718,15 @@ version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6f61fba1741ea2b3d6a1e3178721804bb716a68a6aeba1149b5d52e3d464ea66"
 
+[[package]]
+name = "option-operations"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7c26d27bb1aeab65138e4bf7666045169d1717febcc9ff870166be8348b223d0"
+dependencies = [
+ "paste",
+]
+
 [[package]]
 name = "parking_lot"
 version = "0.12.1"
@@ -552,6 +750,12 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "paste"
+version = "1.0.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f746c4065a8fa3fe23974dd82f15431cc8d40779821001404d10d2e79ca7d79"
+
 [[package]]
 name = "percent-encoding"
 version = "2.2.0"
@@ -602,6 +806,46 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
 
+[[package]]
+name = "pretty-hex"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6fa0831dd7cc608c38a5e323422a0077678fa5744aa2be4ad91c4ece8eec8d5"
+
+[[package]]
+name = "proc-macro-crate"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f4c021e1093a56626774e81216a4ce732a735e5bad4868a03f3ed65ca0c3919"
+dependencies = [
+ "once_cell",
+ "toml_edit",
+]
+
+[[package]]
+name = "proc-macro-error"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da25490ff9892aab3fcf7c36f08cfb902dd3e71ca0f9f9517bea02a73a5ce38c"
+dependencies = [
+ "proc-macro-error-attr",
+ "proc-macro2",
+ "quote",
+ "syn",
+ "version_check",
+]
+
+[[package]]
+name = "proc-macro-error-attr"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1be40180e52ecc98ad80b184934baf3d0d29f979574e439af5a55274b35f869"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "version_check",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.50"
@@ -831,6 +1075,19 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "system-deps"
+version = "6.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2955b1fe31e1fa2fbd1976b71cc69a606d7d4da16f6de3333d0c92d51419aeff"
+dependencies = [
+ "cfg-expr",
+ "heck",
+ "pkg-config",
+ "toml",
+ "version-compare",
+]
+
 [[package]]
 name = "tempfile"
 version = "3.3.0"
@@ -948,6 +1205,32 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "toml"
+version = "0.5.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f4f7f0dd8d50a853a531c426359045b1998f04219d88799810762cd4ad314234"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "toml_datetime"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ab8ed2edee10b50132aed5f331333428b011c99402b5a534154ed15746f9622"
+
+[[package]]
+name = "toml_edit"
+version = "0.19.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a1eb0622d28f4b9c90adc4ea4b2b46b47663fde9ac5fafcb14a1369d5508825"
+dependencies = [
+ "indexmap",
+ "toml_datetime",
+ "winnow",
+]
+
 [[package]]
 name = "tower-service"
 version = "0.3.2"
@@ -1056,6 +1339,12 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "urlencoding"
+version = "2.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8db7427f936968176eaa7cdf81b7f98b980b18495ec28f1b5791ac3bfe3eea9"
+
 [[package]]
 name = "utf-8"
 version = "0.7.6"
@@ -1077,6 +1366,12 @@ version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
 
+[[package]]
+name = "version-compare"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "579a42fc0b8e0c63b76519a339be31bed574929511fa53c1a3acae26eb258f29"
+
 [[package]]
 name = "version_check"
 version = "0.9.4"
@@ -1208,3 +1503,12 @@ name = "windows_x86_64_msvc"
 version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "447660ad36a13288b1db4d4248e857b510e8c3a225c822ba4fb748c0aafecffd"
+
+[[package]]
+name = "winnow"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee7b2c67f962bf5042bfd8b6a916178df33a26eec343ae064cb8e069f638fa6f"
+dependencies = [
+ "memchr",
+]

music-player/server/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 
 [dependencies]
 flow        = { path = "../../flow" }
+gstreamer   = { version = "0.19" }
 id3         = { version = "1.6" }
 mime_guess  = { version = "2.0" }
 mime        = { version = "0.3" }
@@ -17,5 +18,6 @@ tokio       = { version = "1.24", features = ["full"] }
 url         = { version = "2.3" }
 uuid        = { version = "1", features = ["v4"] }
 warp        = { version = "0.3" }
+urlencoding = { version = "2.1" }
 
 [lib]

music-player/server/src/bin/server.rs

@@ -1,17 +1,23 @@
 use flow::Flow;
+use music_player::{
+    core::{ControlMsg, Core},
+    database::{MemoryIndex, MusicIndex},
+    media::{TrackId, TrackInfo},
+    scanner::FileScanner,
+};
+use serde::Deserialize;
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::PathBuf,
     sync::Arc,
+    thread,
 };
 use warp::Filter;
 
-use music_player::{
+#[derive(Clone, Debug, Deserialize)]
-    core::Core,
+struct TrackRequest {
-    database::{MemoryIndex, MusicIndex},
+    id: String,
-    media::TrackInfo,
+}
-    scanner::FileScanner,
-};
 
 fn tracks(index: &Arc<impl MusicIndex>) -> Vec<TrackInfo> {
     match index.list_tracks() {
@@ -25,15 +31,6 @@ struct Static(PathBuf);
 
 impl Static {
     fn read(self, root: PathBuf) -> String {
-        /*
-        let mut path = root;
-        match self {
-            Bundle::Index => path.push(PathBuf::from("index.html")),
-            Bundle::App => path.push(PathBuf::from("bundle.js")),
-            Bundle::Styles => path.push(PathBuf::from("styles.css")),
-        };
-        std::fs::read_to_string(path).expect("to find the file")
-        */
         let mut path = root;
         path.push(self.0);
         println!("path: {:?}", path);
@@ -55,13 +52,15 @@ pub async fn main() {
         .unwrap();
 
     let index = Arc::new(MemoryIndex::new());
-    let scanner = FileScanner::new(vec![music_root.clone()]);
+    let scanner = Arc::new(FileScanner::new(vec![music_root.clone()]));
-    let _core = match Core::new(index.clone(), scanner) {
+    let (core, api) = match Core::new(index.clone(), scanner) {
-        Flow::Ok(core) => core,
+        Flow::Ok((core, api)) => (core, api),
         Flow::Err(error) => panic!("error: {}", error),
         Flow::Fatal(error) => panic!("fatal: {}", error),
     };
 
+    let _handle = thread::spawn(move || core.start());
+
     println!("config: {:?} {:?} {:?}", dev, bundle_root, music_root);
 
     let root = warp::path!().and(warp::get()).map({
@@ -102,6 +101,18 @@ pub async fn main() {
         move || warp::reply::json(&tracks(&index))
     });
 
+    let play_track = warp::path!("api" / "v1" / "play")
+        .and(warp::post())
+        .and(warp::body::json())
+        .map({
+            let api = api.clone();
+            move |body: TrackRequest| {
+                let result = api.play_track(TrackId::from(body.id));
+                println!("Play result: {:?}", result);
+                warp::reply::json(&("ok".to_owned()))
+            }
+        });
+
     /*
     let tracks_for_artist = warp::path!("api" / "v1" / "artist" / String)
         .and(warp::get())
@@ -128,7 +139,7 @@ pub async fn main() {
         .or(queue)
         .or(playing_status);
         */
-    let routes = root.or(assets).or(track_list);
+    let routes = root.or(assets).or(track_list).or(play_track);
     let server = warp::serve(routes);
     server
         .run(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 8002))

music-player/server/src/core.rs

@@ -1,9 +1,16 @@
-use crate::{database::MusicIndex, media::TrackInfo, scanner::MusicScanner, Error, FatalError};
+use crate::{
-use flow::{ok, Flow};
+    database::MusicIndex,
+    media::{TrackId, TrackInfo},
+    playback::{Playback, PlaybackControl, PlaybackStatus},
+    scanner::MusicScanner,
+    Error, FatalError,
+};
+use flow::{ok, return_error, Flow};
+use gstreamer::{format::ClockTime, prelude::*, MessageView};
 use std::{
     sync::{
         mpsc::{channel, Receiver, RecvTimeoutError, Sender},
-        Arc,
+        Arc, Mutex,
     },
     thread,
     thread::JoinHandle,
@@ -14,7 +21,9 @@ fn scan_frequency() -> Duration {
     Duration::from_secs(60)
 }
 
+#[derive(Clone)]
 pub enum ControlMsg {
+    PlayTrack(TrackId),
     Exit,
 }
 
@@ -32,74 +41,119 @@ pub enum PlaybackMsg {
 
 pub struct Core {
     db: Arc<dyn MusicIndex>,
-    _track_handle: JoinHandle<()>,
+    scanner: Arc<dyn MusicScanner>,
-    _track_rx: Receiver<TrackMsg>,
+    control_rx: Receiver<ControlMsg>,
-    _playback_handle: JoinHandle<()>,
+
-    _playback_rx: Receiver<PlaybackMsg>,
+    playback_controller: Playback,
-    control_tx: Sender<ControlMsg>,
 }
 
 impl Core {
     pub fn new(
         db: Arc<dyn MusicIndex>,
-        scanner: impl MusicScanner + 'static,
+        scanner: Arc<dyn MusicScanner>,
-    ) -> Flow<Core, FatalError, Error> {
+    ) -> Flow<(Core, CoreAPI), FatalError, Error> {
         let (control_tx, control_rx) = channel::<ControlMsg>();
         let db = db;
 
-        let (_track_handle, _track_rx) = {
+        let playback_controller = Playback::new();
-            let (track_tx, track_rx) = channel();
+
-            let db = db.clone();
+        ok((
-            let track_handle = thread::spawn(move || {
+            Core {
+                db,
+                scanner,
+                control_rx,
+                playback_controller,
+            },
+            CoreAPI {
+                control_tx: Arc::new(Mutex::new(control_tx)),
+            },
+        ))
+    }
+
+    pub fn start(&self) -> Flow<(), FatalError, Error> {
+        gstreamer::init();
+        let (scanner_tx, _scanner_rx) = channel();
         let mut next_scan = Instant::now();
         loop {
             if Instant::now() >= next_scan {
-                        let _ = track_tx.send(TrackMsg::UpdateInProgress);
+                let scan_start = Instant::now();
-                        for track in scanner.scan() {
+                let _ = scanner_tx.send(TrackMsg::UpdateInProgress);
+                for track in self.scanner.scan() {
                     match track {
-                                Ok(track) => db.add_track(track),
+                        Ok(track) => self.db.add_track(track),
                         Err(_) => ok(()),
                     };
                 }
-                        let _ = track_tx.send(TrackMsg::UpdateComplete);
+                let _ = scanner_tx.send(TrackMsg::UpdateComplete);
                 next_scan = Instant::now() + scan_frequency();
+                println!("scan duration: {:?}", Instant::now() - scan_start);
             }
-                    match control_rx.recv_timeout(Duration::from_millis(1000)) {
+            match self.control_rx.recv_timeout(Duration::from_millis(1000)) {
-                        Ok(ControlMsg::Exit) => return,
+                Ok(ControlMsg::PlayTrack(id)) => {
+                    let _ = self.play_track(id);
+                }
+                Ok(ControlMsg::Exit) => return ok(()),
                 Err(RecvTimeoutError::Timeout) => (),
-                        Err(RecvTimeoutError::Disconnected) => return,
+                Err(RecvTimeoutError::Disconnected) => return ok(()),
             }
         }
-            });
-            (track_handle, track_rx)
-        };
-
-        let (_playback_handle, _playback_rx) = {
-            let (_playback_tx, playback_rx) = channel();
-            let playback_handle = thread::spawn(move || {});
-            (playback_handle, playback_rx)
-        };
-
-        ok(Core {
-            db,
-            _track_handle,
-            _track_rx,
-            _playback_handle,
-            _playback_rx,
-            control_tx,
-        })
     }
 
     pub fn list_tracks<'a>(&'a self) -> Flow<Vec<TrackInfo>, FatalError, Error> {
         self.db.list_tracks().map_err(Error::DatabaseError)
     }
 
-    pub fn exit(&self) {
+    pub fn play_track<'a>(&'a self, id: TrackId) -> Flow<(), FatalError, Error> {
-        let _ = self.control_tx.send(ControlMsg::Exit);
         /*
-        self.track_handle.join();
+        println!("play_track: {}", id.as_ref());
-        self.playback_handle.join();
+        let pipeline = return_error!(Flow::from(
+            gstreamer::parse_launch(&format!("playbin uri={}", id.as_str()))
+                .map_err(|err| Error::CannotPlay(err.to_string()),)
+        ));
+        return_error!(Flow::from(
+            pipeline
+                .set_state(gstreamer::State::Playing)
+                .map_err(|err| Error::CannotPlay(err.to_string()))
+        ));
+        {
+            let pipeline = pipeline.clone();
+            thread::spawn(move || {
+                println!("starting");
+                let bus = pipeline.bus().unwrap();
+                for msg in bus.iter_timed(gstreamer::ClockTime::NONE) {
+                    match msg.view() {
+                        MessageView::Eos(_) => (),
+                        MessageView::Error(err) => {
+                            println!(
+                                "Error from {:?}: {} ({:?})",
+                                err.src().map(|s| s.path_string()),
+                                err.error(),
+                                err.debug()
+                            );
+                        }
+                        msg => println!("{:?}", msg),
+                    }
+                }
+            });
+        }
         */
+        self.playback_controller.play_track(id);
+        ok(())
+    }
+}
+
+#[derive(Clone)]
+pub struct CoreAPI {
+    control_tx: Arc<Mutex<Sender<ControlMsg>>>,
+}
+
+impl CoreAPI {
+    pub fn play_track(&self, id: TrackId) -> () {
+        self.control_tx
+            .lock()
+            .unwrap()
+            .send(ControlMsg::PlayTrack(id))
+            .unwrap()
     }
 }
 
@@ -115,8 +169,8 @@ mod test {
     {
         let index = MemoryIndex::new();
         let scanner = MockScanner::new();
-        match Core::new(Arc::new(index), scanner) {
+        match Core::new(Arc::new(index), Arc::new(scanner)) {
-            Flow::Ok(core) => {
+            Flow::Ok((core, api)) => {
                 thread::sleep(Duration::from_millis(10));
                 f(core)
             }

music-player/server/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod core;
 pub mod database;
 pub mod media;
+pub mod playback;
 pub mod scanner;
 use database::DatabaseError;
 use thiserror::Error;
@@ -9,6 +10,15 @@ use thiserror::Error;
 pub enum Error {
     #[error("Database error: {0}")]
     DatabaseError(DatabaseError),
+
+    #[error("Cannot play track")]
+    CannotPlay,
+
+    #[error("Cannot stop playback")]
+    CannotStop,
+
+    #[error("Unmatched glib error: {0}")]
+    GlibError(gstreamer::glib::Error),
 }
 
 impl From<DatabaseError> for Error {

music-player/server/src/media.rs

@@ -59,6 +59,12 @@ impl From<url::ParseError> for AudioError {
 #[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize)]
 pub struct TrackId(String);
 
+impl TrackId {
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
 impl Default for TrackId {
     fn default() -> Self {
         Self(uuid::Uuid::new_v4().as_hyphenated().to_string())

music-player/server/src/playback.rs

@@ -0,0 +1,131 @@
+use crate::{media::TrackId, Error, FatalError};
+use flow::{ok, return_error, Flow};
+use gstreamer::{format::ClockTime, prelude::*, MessageView, StateChangeError};
+use std::{
+    path::PathBuf,
+    sync::mpsc::{channel, Receiver, Sender},
+    thread::{self, JoinHandle},
+    time::Duration,
+};
+use urlencoding::encode;
+
+pub enum PlaybackControl {
+    PlayTrack(TrackId),
+    Stop,
+    Exit,
+}
+
+pub enum PlaybackStatus {
+    Stopped,
+}
+
+pub struct Playback {
+    handle: JoinHandle<Flow<(), FatalError, Error>>,
+    control_tx: Sender<PlaybackControl>,
+}
+
+impl Playback {
+    pub fn new() -> Playback {
+        let (control_tx, control_rx) = channel::<PlaybackControl>();
+
+        let handle = thread::spawn(move || {
+            let mut pipeline = None;
+            loop {
+                match control_rx.recv().unwrap() {
+                    PlaybackControl::PlayTrack(id) => match play_track(id) {
+                        Flow::Ok(pipeline_) => pipeline = Some(pipeline_),
+                        Flow::Fatal(err) => panic!("fatal error: {:?}", err),
+                        Flow::Err(err) => panic!("playback error: {:?}", err),
+                    },
+                    PlaybackControl::Stop => {
+                        if let Some(ref pipeline) = pipeline {
+                            return_error!(Flow::from(
+                                pipeline
+                                    .set_state(gstreamer::State::Paused)
+                                    .map_err(|_| Error::CannotStop)
+                            ));
+                        }
+                    }
+                    PlaybackControl::Exit => return ok(()),
+                }
+            }
+        });
+
+        Self { handle, control_tx }
+    }
+
+    pub fn play_track(&self, id: TrackId) {
+        self.control_tx
+            .send(PlaybackControl::PlayTrack(id))
+            .unwrap();
+    }
+}
+
+fn play_track(id: TrackId) -> Flow<gstreamer::Element, FatalError, Error> {
+    let pb = PathBuf::from(id.as_ref());
+    let path = pb
+        .iter()
+        .skip(1)
+        .map(|component| encode(&component.to_string_lossy()).into_owned())
+        .collect::<PathBuf>();
+    let playbin = format!("playbin uri=file:///{}", path.display());
+    println!("setting up to play {}", playbin);
+    let pipeline = return_error!(Flow::from(
+        gstreamer::parse_launch(&playbin).map_err(|err| Error::GlibError(err))
+    ));
+    println!("ready to play");
+    return_error!(Flow::from(
+        pipeline
+            .set_state(gstreamer::State::Playing)
+            .map_err(|_| Error::CannotPlay)
+    ));
+    println!("playing started");
+    ok(pipeline)
+}
+
+/*
+fn play_track(id: TrackId) -> Flow<(), FatalError, Error> {
+    let playbin = format!("playbin uri=file://{}", id.as_ref());
+    let pipeline = return_error!(Flow::from(
+        gstreamer::parse_launch(&playbin).map_err(|err| Error::GlibError(err))
+    ));
+    return_error!(Flow::from(
+        pipeline
+            .set_state(gstreamer::State::Playing)
+            .map_err(|_| Error::CannotPlay)
+    ));
+
+    let message_handler = {
+        let pipeline = pipeline.clone();
+        thread::spawn(move || {
+            let bus = pipeline.bus().unwrap();
+            for msg in bus.iter_timed(gstreamer::ClockTime::NONE) {
+                match msg.view() {
+                    MessageView::Eos(_) => (),
+                    MessageView::Error(err) => {
+                        println!(
+                            "Error from {:?}: {} ({:?})",
+                            err.src().map(|s| s.path_string()),
+                            err.error(),
+                            err.debug()
+                        );
+                    }
+                    msg => println!("{:?}", msg),
+                }
+            }
+        })
+    };
+
+    let query_handler = {
+        let pipeline = pipeline.clone();
+        thread::spawn(move || loop {
+            let position: Option<ClockTime> = pipeline.query_position();
+            let duration: Option<ClockTime> = pipeline.query_duration();
+            println!("Position {:?} {:?}", position, duration);
+            thread::sleep(Duration::from_millis(100));
+        })
+    };
+
+    ok(())
+}
+*/

music-player/server/src/scanner.rs

@@ -37,7 +37,7 @@ impl From<id3::Error> for ScannerError {
     }
 }
 
-pub trait MusicScanner: Send {
+pub trait MusicScanner: Sync + Send {
     fn scan<'a>(&'a self) -> Box<dyn Iterator<Item = Result<TrackInfo, ScannerError>> + 'a>;
 }
 
@@ -58,11 +58,6 @@ pub struct FileIterator {
 
 impl FileIterator {
     fn scan_file(&self, path: PathBuf) -> Result<TrackInfo, ScannerError> {
-        println!(
-            "[{:?}] {}",
-            mime_guess::from_path(path.clone()).first(),
-            path.to_str().unwrap()
-        );
         let mimetype = mime_guess::from_path(path.clone())
             .first()
             .ok_or(ScannerError::CannotScan)?;

music-player/streamer/src/main.rs

@@ -34,7 +34,7 @@ fn main() {
         thread::spawn(move || loop {
             let position: Option<ClockTime> = pipeline.query_position();
             let duration: Option<ClockTime> = pipeline.query_duration();
-            println!("{:?} {:?}", position, duration);
+            println!("Position {:?} {:?}", position, duration);
             thread::sleep(Duration::from_millis(100));
         })
     };