music-player/client/src/client.ts

@@ -1,8 +1,3 @@
-export type Response<A> =
-  | { type: "Success"; content: A }
-  | { type: "Failure"; content: string }
-  | { type: "Fatal"; content: string };
-
 export interface TrackInfo {
   id: string;
   track_number?: number;
@@ -13,37 +8,11 @@ export interface TrackInfo {
 }
 
 export const getTracks = (): Promise<TrackInfo[]> =>
-  fetch("/api/v1/tracks")
+  fetch("/api/v1/tracks").then((r) => r.json());
-    .then((r) => r.json())
-    .then((result: Response<TrackInfo[]>) => {
-      switch (result.type) {
-        case "Success":
-          return result.content || [];
-        case "Failure":
-          console.log("failed: ", result.content);
-          return [];
-        case "Fatal":
-          console.log("fatal: ", result.content);
-          return [];
-      }
-    });
 
-export const playTrack = (id: string): Promise<void> =>
+export const playTrack = (id: string): Promise<Response> =>
   fetch("/api/v1/play", {
     method: "POST",
     headers: { "content-type": "application/json" },
     body: JSON.stringify({ id: id }),
-  })
-    .then((r) => r.json())
-    .then((result: Response<null>) => {
-      console.log("result: ", result);
-    });
-
-export const stopPlayback = (): Promise<void> =>
-  fetch("api/v1/stop", {
-    method: "POST",
-  })
-    .then((r) => r.json())
-    .then((result: Response<null>) => {
-      console.log("result: ", result);
   });

music-player/client/src/components/NowPlaying.ts

@@ -1,7 +1,6 @@
 import { TextField } from "./TextField";
 
 export class NowPlaying extends HTMLElement {
-  onStop: () => void;
   nameContainer: TextField;
   albumContainer: TextField;
   artistContainer: TextField;
@@ -21,8 +20,6 @@ export class NowPlaying extends HTMLElement {
 
     this.artistContainer = document.createElement("text-field");
     this.artistContainer.classList.add("now-playing__artist");
-
-    this.onStop = () => {};
   }
 
   get name(): string | null {
@@ -77,13 +74,6 @@ export class NowPlaying extends HTMLElement {
     container.appendChild(this.albumContainer);
     container.appendChild(this.artistContainer);
 
-    const stopButton = document.createElement("button");
-    stopButton.innerHTML = "Stop";
-    stopButton.addEventListener("click", (_) => {
-      this.onStop();
-    });
-
     this.appendChild(container);
-    this.appendChild(stopButton);
   }
 }

music-player/client/src/main.ts

@@ -1,5 +1,5 @@
 import * as _ from "lodash";
-import { TrackInfo, getTracks, playTrack, stopPlayback } from "./client";
+import { TrackInfo, getTracks, playTrack } from "./client";
 import { DataCard } from "./components/DataCard";
 import { NowPlaying } from "./components/NowPlaying";
 import { TextField } from "./components/TextField";
@@ -55,7 +55,6 @@ const updateNowPlaying = (track: TrackInfo) => {
     card.album = track.album || null;
     card.artist = track.artist || null;
     track_list.appendChild(card);
-    card.onStop = () => stopPlayback();
   }
 };
 

music-player/server/Cargo.lock

@@ -664,7 +664,6 @@ dependencies = [
  "mime_guess",
  "rusqlite",
  "serde",
- "serde_json",
  "thiserror",
  "tokio",
  "url",

music-player/server/Cargo.toml

@@ -13,7 +13,6 @@ mime_guess  = { version = "2.0" }
 mime        = { version = "0.3" }
 rusqlite    = { version = "0.28" }
 serde       = { version = "1.0", features = ["derive"] }
-serde_json  = { version = "1.0" }
 thiserror   = { version = "1.0" }
 tokio       = { version = "1.24", features = ["full"] }
 url         = { version = "2.3" }
@@ -21,7 +20,4 @@ uuid        = { version = "1", features = ["v4"] }
 warp        = { version = "0.3" }
 urlencoding = { version = "2.1" }
 
-[target.armv7-unknown-linux-gnueabi]
-linker = "arm-linux-gnueabi-gcc"
-
 [lib]

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

@@ -4,47 +4,26 @@ use music_player::{
     database::{MemoryIndex, MusicIndex},
     media::{TrackId, TrackInfo},
     scanner::FileScanner,
-    Error, FatalError,
 };
-use serde::{Deserialize, Serialize};
+use serde::Deserialize;
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::PathBuf,
-    sync::{Arc, RwLock},
+    sync::Arc,
     thread,
 };
 use warp::Filter;
 
 #[derive(Clone, Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
 struct TrackRequest {
     id: String,
 }
 
-#[derive(Serialize)]
+fn tracks(index: &Arc<impl MusicIndex>) -> Vec<TrackInfo> {
-#[serde(tag = "type", content = "content")]
+    match index.list_tracks() {
-enum Response<A: Serialize> {
+        Flow::Ok(tracks) => tracks,
-    Success(A),
+        Flow::Err(err) => panic!("error: {}", err),
-    Failure(String),
+        Flow::Fatal(err) => panic!("fatal: {}", err),
-    Fatal(String),
-}
-
-impl<A: Serialize> From<Result<A, Error>> for Response<A> {
-    fn from(res: Result<A, Error>) -> Self {
-        match res {
-            Ok(val) => Self::Success(val),
-            Err(err) => Self::Failure(format!("{}", err)),
-        }
-    }
-}
-
-impl<A: Serialize> From<Flow<A, FatalError, Error>> for Response<A> {
-    fn from(res: Flow<A, FatalError, Error>) -> Self {
-        match res {
-            Flow::Ok(val) => Self::Success(val),
-            Flow::Fatal(fatal) => Self::Fatal(format!("{}", fatal)),
-            Flow::Err(err) => Self::Failure(format!("{}", err)),
-        }
     }
 }
 
@@ -54,7 +33,8 @@ impl Static {
     fn read(self, root: PathBuf) -> String {
         let mut path = root;
         path.push(self.0);
-        std::fs::read_to_string(path.clone()).expect(&format!("to find {:?}", path))
+        println!("path: {:?}", path);
+        std::fs::read_to_string(path).expect("to find the file")
     }
 }
 
@@ -71,18 +51,16 @@ pub async fn main() {
         .map(|b| PathBuf::from(b))
         .unwrap();
 
-    match gstreamer::init() {
-        Ok(()) => (),
-        Err(err) => panic!("failed to initialize gstreamer: {}", err),
-    }
-
     let index = Arc::new(MemoryIndex::new());
     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) {
-        Ok(core) => Arc::new(RwLock::new(core)),
+        Flow::Ok((core, api)) => (core, api),
-        Err(error) => panic!("core failed to initialize: {}", error),
+        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({
@@ -101,6 +79,8 @@ pub async fn main() {
                 .map(|m| m.essence_str().to_owned())
                 .unwrap_or("text/plain".to_owned());
             println!("mime_type: {:?}", mime_type);
+            // let mut path = PathBuf::from("assets");
+            // path.push(filename);
             warp::http::Response::builder()
                 .header("content-type", mime_type)
                 .body(Static(PathBuf::from(filename)).read(bundle_root.clone()))
@@ -118,32 +98,21 @@ pub async fn main() {
 
     let track_list = warp::path!("api" / "v1" / "tracks").and(warp::get()).map({
         let index = index.clone();
-        move || {
+        move || warp::reply::json(&tracks(&index))
-            warp::reply::json(&Response::from(
-                index
-                    .list_tracks()
-                    .map_err(|db_err| Error::DatabaseError(db_err)),
-            ))
-        }
     });
 
     let play_track = warp::path!("api" / "v1" / "play")
         .and(warp::post())
         .and(warp::body::json())
         .map({
-            let core = core.clone();
+            let api = api.clone();
             move |body: TrackRequest| {
-                warp::reply::json(&Response::from(
+                let result = api.play_track(TrackId::from(body.id));
-                    core.write().unwrap().play_track(TrackId::from(body.id)),
+                println!("Play result: {:?}", result);
-                ))
+                warp::reply::json(&("ok".to_owned()))
             }
         });
 
-    let stop_playback = warp::path!("api" / "v1" / "stop").and(warp::post()).map({
-        let core = core.clone();
-        move || warp::reply::json(&Response::from(core.write().unwrap().stop_playback()))
-    });
-
     /*
     let tracks_for_artist = warp::path!("api" / "v1" / "artist" / String)
         .and(warp::get())
@@ -170,11 +139,7 @@ pub async fn main() {
         .or(queue)
         .or(playing_status);
         */
-    let routes = root
+    let routes = root.or(assets).or(track_list).or(play_track);
-        .or(assets)
-        .or(track_list)
-        .or(play_track)
-        .or(stop_playback);
     let server = warp::serve(routes);
     server
         .run(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 8002))

music-player/server/src/core.rs

@@ -8,32 +8,30 @@ use crate::{
 use flow::{ok, return_error, Flow};
 use gstreamer::{format::ClockTime, prelude::*, MessageView};
 use std::{
-    sync::{Arc, Mutex, RwLock},
+    sync::{
+        mpsc::{channel, Receiver, RecvTimeoutError, Sender},
+        Arc, Mutex,
+    },
     thread,
+    thread::JoinHandle,
     time::{Duration, Instant},
 };
-use tokio::{
-    sync::mpsc::{channel, Receiver, Sender},
-    task::AbortHandle,
-};
 
 fn scan_frequency() -> Duration {
     Duration::from_secs(60)
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub enum ControlMsg {
     PlayTrack(TrackId),
     Exit,
 }
 
-#[derive(Clone, Debug)]
 pub enum TrackMsg {
     UpdateInProgress,
     UpdateComplete,
 }
 
-#[derive(Clone, Debug)]
 pub enum PlaybackMsg {
     PositionUpdate,
     Playing,
@@ -43,77 +41,119 @@ pub enum PlaybackMsg {
 
 pub struct Core {
     db: Arc<dyn MusicIndex>,
-    playback_controller: Option<Playback>,
+    scanner: Arc<dyn MusicScanner>,
-    next_scan: Arc<RwLock<Instant>>,
+    control_rx: Receiver<ControlMsg>,
-    scanner_handle: thread::JoinHandle<()>,
+
+    playback_controller: Playback,
 }
 
 impl Core {
     pub fn new(
         db: Arc<dyn MusicIndex>,
         scanner: Arc<dyn MusicScanner>,
-    ) -> Result<Core, FatalError> {
+    ) -> Flow<(Core, CoreAPI), FatalError, Error> {
+        let (control_tx, control_rx) = channel::<ControlMsg>();
         let db = db;
 
-        let next_scan = Arc::new(RwLock::new(Instant::now()));
+        let playback_controller = Playback::new();
-        let scanner_handle = {
-            let db = db.clone();
-            let next_scan = next_scan.clone();
-            thread::spawn(move || scan_loop(scanner, db, next_scan))
-        };
 
-        Ok(Core {
+        ok((
+            Core {
                 db,
-            playback_controller: None,
+                scanner,
-            next_scan,
+                control_rx,
-            scanner_handle,
+                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 scan_start = Instant::now();
+                let _ = scanner_tx.send(TrackMsg::UpdateInProgress);
+                for track in self.scanner.scan() {
+                    match track {
+                        Ok(track) => self.db.add_track(track),
+                        Err(_) => ok(()),
+                    };
+                }
+                let _ = scanner_tx.send(TrackMsg::UpdateComplete);
+                next_scan = Instant::now() + scan_frequency();
+                println!("scan duration: {:?}", Instant::now() - scan_start);
+            }
+            match self.control_rx.recv_timeout(Duration::from_millis(1000)) {
+                Ok(ControlMsg::PlayTrack(id)) => {
+                    let _ = self.play_track(id);
+                }
+                Ok(ControlMsg::Exit) => return ok(()),
+                Err(RecvTimeoutError::Timeout) => (),
+                Err(RecvTimeoutError::Disconnected) => return ok(()),
+            }
+        }
     }
 
     pub fn list_tracks<'a>(&'a self) -> Flow<Vec<TrackInfo>, FatalError, Error> {
         self.db.list_tracks().map_err(Error::DatabaseError)
     }
 
-    pub fn play_track<'a>(&'a mut self, id: TrackId) -> Result<(), Error> {
+    pub fn play_track<'a>(&'a self, id: TrackId) -> Flow<(), FatalError, Error> {
-        self.stop_playback()?;
+        /*
-        self.playback_controller = Some(Playback::new(id)?);
+        println!("play_track: {}", id.as_ref());
-        Ok(())
+        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),
-    pub fn stop_playback<'a>(&'a mut self) -> Result<(), Error> {
-        match self.playback_controller {
-            Some(ref controller) => controller.stop()?,
-            None => (),
                     }
-        self.playback_controller = None;
+                }
-        Ok(())
+            });
+        }
+        */
+        self.playback_controller.play_track(id);
+        ok(())
     }
 }
 
-pub fn scan_loop(
+#[derive(Clone)]
-    scanner: Arc<dyn MusicScanner>,
+pub struct CoreAPI {
-    db: Arc<dyn MusicIndex>,
+    control_tx: Arc<Mutex<Sender<ControlMsg>>>,
-    next_scan: Arc<RwLock<Instant>>,
+}
-) {
+
-    loop {
+impl CoreAPI {
-        if Instant::now() >= *next_scan.read().unwrap() {
+    pub fn play_track(&self, id: TrackId) -> () {
-            let scan_start = Instant::now();
+        self.control_tx
-            let mut counter = 0;
+            .lock()
-            for track in scanner.scan() {
+            .unwrap()
-                counter += 1;
+            .send(ControlMsg::PlayTrack(id))
-                match track {
+            .unwrap()
-                    Ok(track) => db.add_track(track),
-                    Err(_) => ok(()),
-                };
-            }
-            *next_scan.write().unwrap() = Instant::now() + scan_frequency();
-            println!(
-                "scanned {} files in {:?}",
-                counter,
-                Instant::now() - scan_start
-            );
-        }
-        thread::sleep(Duration::from_secs(1));
     }
 }
 
@@ -130,11 +170,12 @@ mod test {
         let index = MemoryIndex::new();
         let scanner = MockScanner::new();
         match Core::new(Arc::new(index), Arc::new(scanner)) {
-            Ok(core) => {
+            Flow::Ok((core, api)) => {
                 thread::sleep(Duration::from_millis(10));
                 f(core)
             }
-            Err(error) => panic!("{:?}", error),
+            Flow::Err(error) => panic!("{:?}", error),
+            Flow::Fatal(error) => panic!("{:?}", error),
         }
     }
 

music-player/server/src/database.rs

@@ -4,7 +4,6 @@ use crate::{
 };
 use flow::{error, ok, Flow};
 use rusqlite::Connection;
-use serde::Serialize;
 use std::collections::HashMap;
 use std::{
     path::PathBuf,
@@ -12,12 +11,12 @@ use std::{
 };
 use thiserror::Error;
 
-#[derive(Debug, Error, PartialEq, Serialize)]
+#[derive(Debug, Error, PartialEq)]
 pub enum DatabaseError {
     #[error("database is unreadable")]
     DatabaseUnreadable,
     #[error("unhandled database problem: {0}")]
-    UnhandledError(String),
+    UnhandledError(rusqlite::Error),
 }
 
 pub trait MusicIndex: Sync + Send {
@@ -95,7 +94,7 @@ impl Database {
     pub fn new(path: PathBuf) -> Flow<Database, FatalError, DatabaseError> {
         let connection = match Connection::open(path.clone()) {
             Ok(connection) => connection,
-            Err(err) => return error(DatabaseError::UnhandledError(err.to_string())),
+            Err(err) => return error(DatabaseError::UnhandledError(err)),
         };
         ok(Database {
             path,

music-player/server/src/lib.rs

@@ -12,7 +12,7 @@ pub enum Error {
     DatabaseError(DatabaseError),
 
     #[error("Cannot play track")]
-    CannotPlay(String),
+    CannotPlay,
 
     #[error("Cannot stop playback")]
     CannotStop,

music-player/server/src/playback.rs

@@ -3,10 +3,10 @@ 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 tokio::sync::mpsc::{channel, Receiver, Sender};
 use urlencoding::encode;
 
 pub enum PlaybackControl {
@@ -20,57 +20,84 @@ pub enum PlaybackStatus {
 }
 
 pub struct Playback {
-    events_handle: tokio::task::JoinHandle<()>,
+    handle: JoinHandle<Flow<(), FatalError, Error>>,
-    pipeline: gstreamer::Element,
+    control_tx: Sender<PlaybackControl>,
 }
 
 impl Playback {
-    pub fn new(id: TrackId) -> Result<Self, Error> {
+    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 pipeline = gstreamer::parse_launch(&format!("playbin uri=file:///{}", path.display()))
+    let playbin = format!("playbin uri=file:///{}", path.display());
-            .map_err(|err| Error::GlibError(err))?;
+    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(|err| Error::CannotPlay(err.to_string()))?;
+            .map_err(|_| Error::CannotPlay)
-        let events_handle = tokio::task::spawn_blocking({
+    ));
+    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();
-            move || pipeline_events(pipeline)
+        thread::spawn(move || {
-        });
-
-        Ok(Self {
-            events_handle,
-            pipeline,
-        })
-    }
-
-    pub fn stop(&self) -> Result<(), Error> {
-        self.events_handle.abort();
-        self.pipeline
-            .set_state(gstreamer::State::Paused)
-            .map_err(|_| Error::CannotStop)?;
-        Ok(())
-    }
-
-    pub fn position(&self) -> (Option<ClockTime>, Option<ClockTime>) {
-        let position = self.pipeline.query_position();
-        let duration = self.pipeline.query_duration();
-        (position, duration)
-    }
-}
-
-impl Drop for Playback {
-    fn drop(&mut self) {
-        self.stop();
-    }
-}
-
-fn pipeline_events(pipeline: gstreamer::Element) {
             let bus = pipeline.bus().unwrap();
             for msg in bus.iter_timed(gstreamer::ClockTime::NONE) {
                 match msg.view() {
@@ -86,4 +113,19 @@ fn pipeline_events(pipeline: gstreamer::Element) {
                     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(())
 }
+*/