Set up to be able to add a track to a running pipeline

All of the examples are for switching out elements in the middle of a
pipeline. In this case I am trying to switch out the trailing element.
The element refuses to be removed, and a new one can't be added until
the old audio sink is removed. I think that the audio sink can't be
removed because it is still holding on to data, and I don't know how to
detect the EOS signal as it passes through.

flake.lock

@@ -20,16 +20,16 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1704732714,
+        "lastModified": 1724316499,
-        "narHash": "sha256-ABqK/HggMYA/jMUXgYyqVAcQ8QjeMyr1jcXfTpSHmps=",
+        "narHash": "sha256-Qb9MhKBUTCfWg/wqqaxt89Xfi6qTD3XpTzQ9eXi3JmE=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "6723fa4e4f1a30d42a633bef5eb01caeb281adc3",
+        "rev": "797f7dc49e0bc7fab4b57c021cdf68f595e47841",
         "type": "github"
       },
       "original": {
         "id": "nixpkgs",
-        "ref": "nixos-23.11",
+        "ref": "nixos-24.05",
         "type": "indirect"
       }
     },

flake.nix

@@ -2,7 +2,7 @@
   description = "Lumenescent Dreams Tools";
 
   inputs = {
-    nixpkgs.url = "nixpkgs/nixos-23.11";
+    nixpkgs.url = "nixpkgs/nixos-24.05";
     unstable.url = "nixpkgs/nixos-unstable";
     typeshare.url = "github:1Password/typeshare";
   };
@@ -30,6 +30,9 @@
             pkgs.gst_all_1.gst-plugins-good
             pkgs.gst_all_1.gst-plugins-ugly
             pkgs.gst_all_1.gstreamer
+            pkgs.gst_all_1.gstreamer.dev
+            pkgs.gst_all_1.gst-libav
+            pkgs.gst_all_1.gst-vaapi
             pkgs.gtk4
             pkgs.libadwaita
             pkgs.librsvg

gm-dash/server/Cargo.toml

@@ -6,8 +6,13 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-pipewire = "0.8.0"
+gstreamer   = { version = "0.23.0", features = ["serde", "v1_24"] }
+pipewire    = { version = "0.8.0" }
 serde       = { version = "1.0.209", features = ["alloc", "derive"] }
-serde_json = "1.0.127"
+serde_json  = { version = "1.0.127" }
 tokio       = { version = "1.39.3", features = ["full"] }
-warp = "0.3.7"
+warp        = { version = "0.3.7" }
+glib        = { version = "0.18" }
+thiserror = "1.0.63"
+cool_asserts = "2.0.3"
+

gm-dash/server/src/app/mod.rs

@@ -0,0 +1,163 @@
+use std::{
+    collections::HashMap,
+    path::PathBuf,
+    sync::{Arc, RwLock},
+};
+
+use tokio::{
+    sync::mpsc::{Receiver, Sender},
+    task::JoinHandle,
+};
+
+use crate::{
+    audio_control::AudioControl,
+    types::{
+        AppError, AudioControlMessage, AudioState, AudioStatusMessage, TrackInfo, TrackSpec, Volume,
+    },
+};
+
+#[cfg(test)]
+mod tests;
+
+struct AppState {
+    playing: bool,
+
+    device_list: Vec<String>,
+    track_list: Vec<PathBuf>,
+
+    track_status: Vec<TrackInfo>,
+}
+
+impl Default for AppState {
+    fn default() -> Self {
+        Self {
+            playing: false,
+
+            device_list: vec![],
+            track_list: vec![],
+
+            track_status: vec![],
+        }
+    }
+}
+
+pub struct App {
+    state: Arc<RwLock<AppState>>,
+
+    audio_control: Sender<AudioControlMessage>,
+
+    listener: JoinHandle<()>,
+}
+
+impl App {
+    pub fn new(
+        audio_control: Sender<AudioControlMessage>,
+        mut audio_status: Receiver<AudioStatusMessage>,
+    ) -> App {
+        let state = Arc::new(RwLock::new(AppState::default()));
+
+        let listener = tokio::spawn({
+            let state = state.clone();
+            async move {
+                println!("listener started");
+                while let Some(msg) = audio_status.recv().await {
+                    match msg {
+                        AudioStatusMessage::Playing => {
+                            state.write().unwrap().playing = true;
+                        }
+                        AudioStatusMessage::Status(AudioState {
+                            playing: _playing,
+                            tracks,
+                        }) => {
+                            state.write().unwrap().track_status = tracks;
+                        }
+                        msg => println!("message received from audio controller: {:?}", msg),
+                    }
+                }
+                println!("listener exiting");
+            }
+        });
+
+        Self {
+            state,
+            audio_control,
+            listener,
+        }
+    }
+
+    pub fn playing(&self) -> bool {
+        self.state.read().unwrap().playing
+    }
+
+    pub fn add_audio(&self, device: String) {
+        /*
+        let mut st = self.internal.write().unwrap();
+        st.device_list.push(device);
+        */
+    }
+
+    pub fn audio_devices(&self) -> Vec<String> {
+        /*
+        let st = self.internal.read().unwrap();
+        st.device_list.clone()
+        */
+        vec![]
+    }
+
+    pub fn enabled_tracks(&self) -> Vec<PathBuf> {
+        let st = self.state.read().unwrap();
+        st.track_status.iter().map(|ti| ti.path.clone()).collect()
+    }
+
+    pub async fn enable_track(&self, path: PathBuf) -> Result<(), AppError> {
+        println!("enabling track: {}", path.display());
+        self.audio_control
+            .send(AudioControlMessage::EnableTrack(TrackSpec {
+                path,
+                volume: Volume::try_from(1.0).unwrap(),
+            }))
+            .await
+            .expect("audio control send to succeed");
+        /*
+        let st = self.internal.write().unwrap();
+        st.audio_control.add_track(TrackSpec{ path })?;
+        */
+
+        Ok(())
+    }
+
+    pub async fn disable_track(&self, _track: &str) -> Result<(), AppError> {
+        /*
+        let mut st = self.internal.write().unwrap();
+        if st.currently_playing.contains_key(track) {
+            st.currently_playing.remove(track);
+        }
+        Ok(())
+        */
+        Ok(())
+    }
+
+    pub async fn play(&self) -> Result<(), AppError> {
+        self.audio_control
+            .send(AudioControlMessage::Play)
+            .await
+            .expect("audio control send to succeed");
+        Ok(())
+    }
+
+    pub async fn stop(&self) -> Result<(), AppError> {
+        self.audio_control
+            .send(AudioControlMessage::Stop)
+            .await
+            .expect("audio control send to succeed");
+        Ok(())
+    }
+
+    pub async fn pause(&self) -> Result<(), AppError> {
+        self.audio_control
+            .send(AudioControlMessage::Pause)
+            .await
+            .expect("audio control send to succeed");
+        Ok(())
+    }
+}

gm-dash/server/src/app/tests.rs

@@ -0,0 +1,126 @@
+use std::{future::Future, path::PathBuf, time::Duration};
+
+use cool_asserts::assert_matches;
+use tokio::sync::mpsc::{Receiver, Sender};
+
+use crate::{
+    app::{App, AppError},
+    types::{AudioControlMessage, AudioStatusMessage, Progress, TrackInfo, TrackSpec, Volume},
+};
+
+fn memory_app() -> (
+    App,
+    Receiver<AudioControlMessage>,
+    Sender<AudioStatusMessage>,
+) {
+    let (audio_control_tx, audio_control_rx) = tokio::sync::mpsc::channel(5);
+    let (audio_status_tx, audio_status_rx) = tokio::sync::mpsc::channel(5);
+    let app = App::new(audio_control_tx, audio_status_rx);
+
+    (app, audio_control_rx, audio_status_tx)
+}
+
+#[tokio::test]
+async fn app_starts_in_stopped_state() {
+    let (app, _control_rx, _status_tx) = memory_app();
+
+    assert!(!app.playing());
+}
+
+#[tokio::test]
+async fn can_add_a_track_without_starting_playback() {
+    let (app, mut control_rx, status_tx) = memory_app();
+    let path_1 = PathBuf::from("/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/05 - Books and Spellcrafting.mp3.mp3");
+    let path_2 = PathBuf::from(
+        "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/01 - A Day to Rebuild.mp3.mp3",
+    );
+
+    {
+        app.enable_track(path_1.clone())
+            .await
+            .expect("to enable a track");
+
+        assert_matches!(control_rx.recv().await, Some(AudioControlMessage::EnableTrack(trackspec)) => {
+            assert_eq!(trackspec, TrackSpec{ path: path_1.clone(), volume: Volume::try_from(1.0).unwrap() });
+        });
+
+        status_tx
+            .send(AudioStatusMessage::Status(vec![TrackInfo {
+                path: path_1.clone(),
+                volume: Volume::try_from(1.0).unwrap(),
+                progress: Progress {
+                    current: Duration::from_secs(0),
+                    length: Duration::from_secs(100),
+                },
+            }]))
+            .await
+            .expect("status send to work");
+
+        tokio::time::sleep(Duration::from_millis(1)).await;
+
+        let tracks = app.enabled_tracks();
+        tracks.iter().find(|p| **p == path_1);
+        assert!(!app.playing());
+    }
+
+    {
+        app.enable_track(path_2.clone())
+            .await
+            .expect("to enable a track");
+
+        assert_matches!(control_rx.recv().await, Some(AudioControlMessage::EnableTrack(trackspec)) => {
+            assert_eq!(trackspec, TrackSpec{ path: path_2.clone(), volume: Volume::try_from(1.0).unwrap() });
+        });
+
+        status_tx
+            .send(AudioStatusMessage::Status(vec![
+                TrackInfo {
+                    path: path_1.clone(),
+                    volume: Volume::try_from(1.0).unwrap(),
+                    progress: Progress {
+                        current: Duration::from_secs(0),
+                        length: Duration::from_secs(100),
+                    },
+                },
+                TrackInfo {
+                    path: path_2.clone(),
+                    volume: Volume::try_from(1.0).unwrap(),
+                    progress: Progress {
+                        current: Duration::from_secs(0),
+                        length: Duration::from_secs(100),
+                    },
+                },
+            ]))
+            .await
+            .expect("status send to work");
+
+        tokio::time::sleep(Duration::from_millis(1)).await;
+        let tracks = app.enabled_tracks();
+        tracks.iter().find(|p| **p == path_1);
+        tracks.iter().find(|p| **p == path_2);
+    }
+}
+
+#[tokio::test]
+async fn cannot_start_playback_with_no_tracks() {
+    let (app, control_rx, status_tx) = memory_app();
+    // assert_matches!(app.play(), Err(AppError::NoTracks));
+    unimplemented!()
+}
+
+#[tokio::test]
+async fn can_add_a_track_during_playback() {
+    let (app, control_rx, status_tx) = memory_app();
+    /*
+    app.enable_track(PathBuf::from("/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/05 - Books and Spellcrafting.mp3.mp3")).expect("to enable a track");
+    app.play().expect("to start playback");
+
+    assert!(app.playing());
+
+    app.enable_track(PathBuf::from(
+        "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/01 - A Day to Rebuild.mp3.mp3",
+    ))
+    .expect("to enable another track during playback");
+    */
+    unimplemented!()
+}

gm-dash/server/src/audio_control/mod.rs

@@ -0,0 +1,398 @@
+use std::{
+    collections::HashMap,
+    path::PathBuf,
+    sync::{Arc, RwLock},
+    time::Duration,
+};
+
+use gstreamer::{prelude::*, ClockTime, MessageType, MessageView};
+use thiserror::Error;
+use tokio::sync::mpsc::{Receiver, Sender};
+
+use crate::types::{
+    AudioControlMessage, AudioState, AudioStatusMessage, Progress, TrackInfo, TrackSpec,
+};
+
+#[derive(Debug, Error, PartialEq)]
+pub enum AudioError {
+    #[error("No tracks are available to play")]
+    NoTracks,
+
+    #[error("Cannot perform operation in the current state")]
+    InvalidState,
+}
+
+pub struct AudioControl {
+    backend: Box<dyn AudioControlBackend>,
+}
+
+impl AudioControl {
+    pub fn new(backend: impl AudioControlBackend + 'static) -> Self {
+        Self {
+            backend: Box::new(backend),
+        }
+    }
+
+    pub async fn listen(&self, mut control_rx: Receiver<AudioControlMessage>) {
+        while let Some(msg) = control_rx.recv().await {
+            println!("control message: {:?}", msg);
+            match msg {
+                AudioControlMessage::Play => {
+                    self.backend.play().unwrap();
+                }
+                AudioControlMessage::Stop => {
+                    self.backend.stop().unwrap();
+                }
+                AudioControlMessage::Pause => {
+                    self.backend.pause().unwrap();
+                }
+                AudioControlMessage::EnableTrack(spec) => {
+                    self.backend.add_track(spec).unwrap();
+                }
+                AudioControlMessage::DisableTrack(_) => {
+                    unimplemented!()
+                }
+                AudioControlMessage::ReportStatus => {
+                    unimplemented!()
+                }
+            }
+        }
+    }
+
+    pub async fn report(&self, status_tx: Sender<AudioStatusMessage>) {
+        loop {
+            status_tx
+                .send(AudioStatusMessage::Status(AudioState {
+                    playing: self.backend.playing(),
+                    tracks: self.backend.tracks(),
+                }))
+                .await
+                .expect("to successfully send a message");
+            let _ = tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    }
+}
+
+/*
+    pub fn playing(&self) -> bool {
+        self.backend.read().unwrap().playing()
+    }
+
+    pub fn tracks(&self) -> Vec<TrackSpec> {
+        self.backend.read().unwrap().tracks()
+    }
+
+    pub fn play(&self) -> Result<(), AudioError> {
+        self.backend.read().unwrap().play()
+    }
+
+    pub fn stop(&self) -> Result<(), AudioError> {
+        self.backend.read().unwrap().stop()
+    }
+
+    pub fn add_track(&self, track: TrackSpec) -> Result<(), AudioError> {
+        self.backend.write().unwrap().add_track(track)
+    }
+}
+*/
+
+pub trait AudioControlBackend: Send + Sync {
+    fn playing(&self) -> bool;
+
+    fn tracks(&self) -> Vec<TrackInfo>;
+
+    fn play(&self) -> Result<(), AudioError>;
+
+    fn stop(&self) -> Result<(), AudioError>;
+
+    fn pause(&self) -> Result<(), AudioError>;
+
+    fn add_track(&self, track: TrackSpec) -> Result<(), AudioError>;
+
+    fn remove_track(&self, track: TrackSpec) -> Result<(), AudioError>;
+}
+
+/*
+pub struct MemoryBackend {
+    playing: Arc<RwLock<bool>>,
+    tracks: HashMap<PathBuf, TrackSpec>,
+}
+
+impl Default for MemoryBackend {
+    fn default() -> Self {
+        Self {
+            playing: Arc::new(RwLock::new(false)),
+            tracks: HashMap::new(),
+        }
+    }
+}
+
+impl AudioControlBackend for MemoryBackend {
+    fn playing(&self) -> bool {
+        *self.playing.read().unwrap()
+    }
+
+    fn tracks(&self) -> Vec<TrackSpec> {
+        /*
+        self.tracks.iter().cloned().collect()
+        */
+        vec![]
+    }
+
+    fn play(&self) -> Result<(), AudioError> {
+        if self.tracks.is_empty() {
+            return Err(AudioError::NoTracks);
+        }
+
+        let mut playing = self.playing.write().unwrap();
+        if *playing {
+            return Err(AudioError::InvalidState);
+        }
+
+        *playing = true;
+        Ok(())
+    }
+
+    fn stop(&self) -> Result<(), AudioError> {
+        let mut playing = self.playing.write().unwrap();
+        if *playing {
+            *playing = false;
+            Ok(())
+        } else {
+            Err(AudioError::InvalidState)
+        }
+    }
+
+    fn add_track(&mut self, track: TrackSpec) -> Result<(), AudioError> {
+        /*
+        self.tracks.insert(track);
+        */
+        Ok(())
+    }
+
+    fn remove_track(&mut self, track: TrackSpec) -> Result<(), AudioError> {
+        /*
+        self.tracks.remove(&track);
+        */
+        Ok(())
+    }
+}
+*/
+
+struct GStreamerBackendState {
+    playing: bool,
+    tracks: HashMap<PathBuf, TrackInfo>,
+}
+
+impl Default for GStreamerBackendState {
+    fn default() -> Self {
+        Self {
+            playing: false,
+            tracks: HashMap::new(),
+        }
+    }
+}
+
+pub struct GStreamerBackend {
+    bus: gstreamer::Bus,
+    pipeline: gstreamer::Pipeline,
+    mixer: gstreamer::Element,
+    audio_sink: gstreamer::Element,
+    monitor: std::thread::JoinHandle<()>,
+
+    state: Arc<RwLock<GStreamerBackendState>>,
+}
+
+impl Default for GStreamerBackend {
+    fn default() -> Self {
+        let pipeline = gstreamer::Pipeline::new();
+        let bus = pipeline.bus().unwrap();
+
+        let mixer = gstreamer::ElementFactory::find("audiomixer")
+            .unwrap()
+            .load()
+            .unwrap()
+            .create()
+            .build()
+            .unwrap();
+        pipeline.add(&mixer).unwrap();
+
+        let audio_sink = gstreamer::ElementFactory::find("pulsesink")
+            .unwrap()
+            .load()
+            .unwrap()
+            .create()
+            .build()
+            .unwrap();
+        pipeline.add(&audio_sink).unwrap();
+        mixer.link(&audio_sink).unwrap();
+
+        let state = Arc::new(RwLock::new(GStreamerBackendState::default()));
+
+        let monitor = std::thread::spawn({
+            let pipeline = pipeline.clone();
+            let pipeline_object = pipeline.clone().upcast::<gstreamer::Object>();
+            let state = state.clone();
+            let bus = bus.clone();
+            move || loop {
+                /*
+                if let Some(msg) = bus.timed_pop_filtered(
+                    gstreamer::ClockTime::from_mseconds(100),
+                    &[
+                        MessageType::Error,
+                        MessageType::Eos,
+                        MessageType::StateChanged,
+                    ],
+                ) {
+                */
+                if let Some(msg) = bus.timed_pop(gstreamer::ClockTime::from_mseconds(100)) {
+                    match msg.view() {
+                        MessageView::StateChanged(st) => {
+                            println!("state changed: {:?}", st);
+                            if msg.src() == Some(&pipeline_object) {
+                                state.write().unwrap().playing =
+                                    st.current() == gstreamer::State::Playing;
+                            }
+                        }
+                        MessageView::Error(err) => {
+                            println!("error: {:?}", err);
+                        }
+                        MessageView::Eos(_) => {
+                            println!("EOS");
+                        }
+                        msg => {
+                            println!("{:?}", msg);
+                        }
+                    }
+                } else {
+                    if state.read().unwrap().playing {
+                        let mut q = gstreamer::query::Position::new(gstreamer::Format::Time);
+                        pipeline.query(&mut q);
+                        println!("Position: {:?}", q.result());
+                    }
+                }
+            }
+        });
+
+        Self {
+            bus,
+            pipeline,
+            mixer,
+            audio_sink,
+
+            monitor,
+
+            state,
+        }
+    }
+}
+
+impl AudioControlBackend for GStreamerBackend {
+    fn playing(&self) -> bool {
+        self.state.read().unwrap().playing
+    }
+
+    fn tracks(&self) -> Vec<TrackInfo> {
+        vec![]
+    }
+
+    fn play(&self) -> Result<(), AudioError> {
+        if !self.playing() {
+            self.pipeline.set_state(gstreamer::State::Playing).unwrap();
+            Ok(())
+        } else {
+            Err(AudioError::InvalidState)
+        }
+    }
+
+    fn stop(&self) -> Result<(), AudioError> {
+        if self.playing() {
+            self.pipeline.set_state(gstreamer::State::Ready).unwrap();
+            Ok(())
+        } else {
+            Err(AudioError::InvalidState)
+        }
+    }
+
+    fn pause(&self) -> Result<(), AudioError> {
+        if self.playing() {
+            self.pipeline.set_state(gstreamer::State::Paused).unwrap();
+            Ok(())
+        } else {
+            Err(AudioError::InvalidState)
+        }
+    }
+
+    fn add_track(&self, track: TrackSpec) -> Result<(), AudioError> {
+        let mut st = self.state.write().unwrap();
+        st.tracks.insert(
+            track.path.clone(),
+            TrackInfo {
+                path: track.path.clone(),
+                volume: track.volume,
+                progress: Progress {
+                    current: Duration::from_secs(0),
+                    length: Duration::from_secs(1),
+                },
+            },
+        );
+
+        if st.playing {
+            self.pipeline.set_state(gstreamer::State::Paused);
+        }
+
+        let source = gstreamer::ElementFactory::find("filesrc")
+            .unwrap()
+            .load()
+            .unwrap()
+            .create()
+            .property("location", track.path.to_str().unwrap())
+            .build()
+            .unwrap();
+        self.pipeline.add(&source).unwrap();
+
+        let decoder = gstreamer::ElementFactory::find("decodebin")
+            .unwrap()
+            .load()
+            .unwrap()
+            .create()
+            .build()
+            .unwrap();
+        self.pipeline.add(&decoder).unwrap();
+        source.link(&decoder).unwrap();
+
+        let volume = gstreamer::ElementFactory::find("volume")
+            .unwrap()
+            .load()
+            .unwrap()
+            .create()
+            .property("mute", false)
+            .property("volume", track.volume.as_f64())
+            .build()
+            .unwrap();
+        self.pipeline.add(&volume).unwrap();
+        volume.link(&self.mixer).unwrap();
+
+        decoder.connect_pad_added({
+            let volume = volume.clone();
+            move |_, pad| {
+                let next_pad = volume.static_pad("sink").unwrap();
+                pad.link(&next_pad).unwrap();
+            }
+        });
+
+        if st.playing {
+            source.set_state(gstreamer::State::Paused).unwrap();
+            decoder.set_state(gstreamer::State::Paused).unwrap();
+            volume.set_state(gstreamer::State::Paused).unwrap();
+            self.pipeline.set_state(gstreamer::State::Playing).unwrap();
+        }
+
+        Ok(())
+    }
+
+    fn remove_track(&self, _path: TrackSpec) -> Result<(), AudioError> {
+        unimplemented!()
+        /* Need to run EOS through to a probe on the trailing end of the volume element */
+    }
+}

gm-dash/server/src/bin/enumerate-audio-sinks.rs

@@ -10,8 +10,10 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         .add_listener_local()
         .global(|global| {
             if global.props.and_then(|p| p.get("media.class")) == Some("Audio/Sink"){
+                // println!("{:?}", global.props.map(|p| p));
                 println!(
-                    "\t{:?} {:?}",
+                    "\t{:?} {:?} {:?}",
+                    global.props.and_then(|p| p.get("node.name")),
                     global.props.and_then(|p| p.get("node.description")),
                     global.props.and_then(|p| p.get("media.class"))
                 );

gm-dash/server/src/bin/play-track.rs

@@ -0,0 +1,200 @@
+use std::time::Duration;
+
+use gstreamer::{
+    prelude::*, Bus, Element, EventType, MessageType, MessageView, Pad, PadDirection, PadPresence,
+    PadProbeData, PadProbeInfo, PadProbeReturn, PadTemplate, Pipeline,
+};
+use pipewire::{context::Context, main_loop::MainLoop};
+
+fn main() {
+    gstreamer::init();
+
+    let pipeline = gstreamer::Pipeline::new();
+    let pipeline_object = pipeline.clone().upcast::<gstreamer::Object>();
+
+    let sinkfactory = gstreamer::ElementFactory::find("pulsesink")
+        .unwrap()
+        .load()
+        .unwrap();
+    let audio_template = sinkfactory
+        .static_pad_templates()
+        .iter()
+        .next()
+        .map(|template| template.get())
+        .unwrap();
+
+    let audio_output = sinkfactory.create().name("sink").build().unwrap();
+    pipeline.add(&audio_output).unwrap();
+
+    let funnel = gstreamer::ElementFactory::find("audiomixer")
+        .unwrap()
+        .load()
+        .unwrap()
+        .create()
+        .build()
+        .unwrap();
+    pipeline.add(&funnel).unwrap();
+
+    let convert = gstreamer::ElementFactory::find("audioconvert")
+        .unwrap()
+        .load()
+        .unwrap()
+        .create()
+        .build()
+        .unwrap();
+    pipeline.add(&convert).unwrap();
+    funnel.link(&convert).unwrap();
+    convert.link(&audio_output).unwrap();
+
+    /*
+    setup_file_reader(
+        &pipeline,
+        funnel.clone(),
+        "/home/savanni/Music/technical-station.ogg",
+    );
+    */
+    setup_file_reader(&pipeline, funnel.clone(), "/home/savanni/Music/techno-city-day.ogg");
+
+    let bus = pipeline.bus().unwrap();
+
+    /*
+    let btsink = sinkfactory
+        .create()
+        .name("sink")
+        .property("device", "bluez_output.0C_A6_94_75_6E_8F.1")
+        .build()
+        .unwrap();
+        */
+
+    pipeline.set_state(gstreamer::State::Playing).unwrap();
+    let pipeline_object = pipeline.clone().upcast::<gstreamer::Object>();
+
+    /*
+    std::thread::spawn({
+        let bus = bus.clone();
+        let pipeline = pipeline.clone();
+        move || {
+            std::thread::sleep(Duration::from_secs(5));
+            swap_audio_output(bus, pipeline, resample, defaultsink, btsink);
+        }
+    });
+    */
+
+    pipeline.set_state(gstreamer::State::Playing).unwrap();
+
+    let mut playing = false;
+
+    loop {
+        if let Some(msg) = bus.timed_pop_filtered(
+            gstreamer::ClockTime::from_mseconds(100),
+            &[
+                MessageType::Error,
+                MessageType::Eos,
+                MessageType::Progress,
+                MessageType::StateChanged,
+                MessageType::StructureChange,
+            ],
+        ) {
+            match msg.view() {
+                MessageView::Progress(prog) => {
+                    println!("progress: {:?}", prog);
+                }
+                MessageView::StateChanged(st) => {
+                    if msg.src() == Some(&pipeline_object) {
+                        println!("State changed from {:?} to {:?}", st.old(), st.current());
+                        playing = st.current() == gstreamer::State::Playing;
+                    }
+                }
+                MessageView::StructureChange(change) => {
+                    println!("structure change: {:?}", change);
+                }
+                _ => {
+                    println!("{:?}", msg);
+                }
+            }
+        } else {
+            if playing {
+                let mut q = gstreamer::query::Position::new(gstreamer::Format::Time);
+                pipeline.query(&mut q);
+                println!("Position result: {:?}", q.result());
+            } else {
+                break;
+            }
+        }
+    }
+
+    pipeline.set_state(gstreamer::State::Null).unwrap();
+}
+
+fn handle_pad_added(element: &Element, pad: &Pad, next_element: &Element, template: &PadTemplate) {
+    println!("handle_pad_added");
+    println!("\t{:?}", element);
+    println!("\t{:?}, {:?}", pad, pad.current_caps());
+    /*
+    let audio_caps = gstreamer::caps::Caps::builder()
+        .field("audio", "audio/x-raw,
+        .build();
+        */
+    /*
+    let audio_pad_template = PadTemplate::new(
+        "audio-pad-template",
+        PadDirection::Sink,
+        PadPresence::Request,
+        &pad.current_caps().unwrap(),
+    )
+    .unwrap();
+    */
+    let next_pad = next_element.request_pad(template, None, None).unwrap();
+    // let converter_pad = converter.static_pad("sink").unwrap();
+    pad.link(&next_pad).unwrap();
+}
+
+fn setup_file_reader(pipeline: &Pipeline, dest: Element, path: &str) {
+    let source = gstreamer::ElementFactory::find("filesrc")
+        .unwrap()
+        .load()
+        .unwrap()
+        .create()
+        .property("location", path)
+        .build()
+        .unwrap();
+
+    let decoder = gstreamer::ElementFactory::find("decodebin")
+        .unwrap()
+        .load()
+        .unwrap()
+        .create()
+        .build()
+        .unwrap();
+
+    let volume = gstreamer::ElementFactory::find("volume")
+        .unwrap()
+        .load()
+        .unwrap()
+        .create()
+        .property("mute", false)
+        .property("volume", 0.5)
+        .build()
+        .unwrap();
+
+    pipeline.add(&source).unwrap();
+    pipeline.add(&decoder).unwrap();
+    pipeline.add(&volume).unwrap();
+
+    source.link(&decoder).unwrap();
+
+    let next_pad = dest.request_pad_simple("sink_%u").unwrap();
+    let volume_output = volume.static_pad("src").unwrap();
+    volume_output.link(&next_pad).unwrap();
+
+    decoder.connect_pad_added(
+        move |element, pad| handle_decoder_started(element, pad, volume.clone())
+    );
+}
+
+fn handle_decoder_started(_: &Element, pad: &Pad, next: Element) {
+    println!("connecting file decoder to converter stream");
+    // let next_pad = next.request_pad_simple("sink_%u").unwrap();
+    let next_pad = next.static_pad("sink").unwrap();
+    pad.link(&next_pad).unwrap();
+}

gm-dash/server/src/main.rs

@@ -1,75 +1,128 @@
-use pipewire::{context::Context, main_loop::MainLoop};
 use std::{
+    convert::Infallible,
     net::{Ipv6Addr, SocketAddrV6},
-    sync::{Arc, RwLock},
+    path::PathBuf,
+    sync::Arc,
 };
+
+use app::App;
+use audio_control::{AudioControl, GStreamerBackend};
+use pipewire::{context::Context, main_loop::MainLoop};
+use serde::Deserialize;
 use tokio::task::spawn_blocking;
 use warp::{serve, Filter};
 
-struct State_ {
+mod app;
-    device_list: Vec<String>,
+mod audio_control;
+mod types;
+
+#[derive(Deserialize)]
+struct PlayTrackParams {
+    track_name: String,
 }
 
-#[derive(Clone)]
+fn with_app(app: Arc<App>) -> impl Filter<Extract = (Arc<App>,), Error = Infallible> + Clone {
-struct State {
+    warp::any().map(move || app.clone())
-    internal: Arc<RwLock<State_>>,
 }
 
-impl State {
+async fn server_main(app: Arc<App>) {
-    fn new() -> State {
-        let internal = State_ {
-            device_list: vec![],
-        };
-        State {
-            internal: Arc::new(RwLock::new(internal)),
-        }
-    }
-
-    fn add_audio(&self, device: String) {
-        let mut st = self.internal.write().unwrap();
-        (*st).device_list.push(device);
-    }
-
-    fn audio_devices(&self) -> Vec<String> {
-        let st = self.internal.read().unwrap();
-        (*st).device_list.clone()
-    }
-}
-
-impl Default for State {
-    fn default() -> State {
-        State::new()
-    }
-}
-
-async fn server_main(state: State) {
     let localhost: Ipv6Addr = "::1".parse().unwrap();
     let server_addr = SocketAddrV6::new(localhost, 3001, 0, 0);
 
     let root = warp::path!().map(|| "ok".to_string());
     let list_output_devices = warp::path!("output_devices").map({
-        let state = state.clone();
+        let app = app.clone();
         move || {
-            let devices = state.audio_devices();
+            let devices = app.audio_devices();
             serde_json::to_string(&devices).unwrap()
         }
     });
 
-    let routes = root.or(list_output_devices);
+    /*
+    let list_tracks = warp::path!("tracks").map({
+        let app = app.clone();
+        move || serde_json::to_string(&app.tracks()).unwrap()
+    });
+    */
+
+    let enable_track = warp::put()
+        .and(warp::path!("playing"))
+        .and(warp::body::json())
+        .and(with_app(app.clone()))
+        .then(|params: PlayTrackParams, app: Arc<App>| async move {
+            println!("enable track");
+            let _ = app.enable_track(PathBuf::from(params.track_name)).await;
+            "".to_owned()
+        });
+
+    let disable_track = warp::delete()
+        .and(warp::path!("playing"))
+        .and(warp::body::json())
+        .and(with_app(app.clone()))
+        .then(|params: PlayTrackParams, app: Arc<App>| async move {
+            let _ = app.disable_track(&params.track_name);
+            "".to_owned()
+        });
+
+    let play_all = warp::post()
+        .and(warp::path!("play"))
+        .and(with_app(app.clone()))
+        .then({
+            |app: Arc<App>| async move {
+                println!("play_all");
+                let _ = app.play().await;
+                "".to_owned()
+            }
+        });
+
+    let stop_all = warp::post()
+        .and(warp::path!("stop"))
+        .and(with_app(app.clone()))
+        .then({
+            |app: Arc<App>| async move {
+                let _ = app.stop().await;
+                "".to_owned()
+            }
+        });
+
+    let pause = warp::post()
+        .and(warp::path!("pause"))
+        .and(with_app(app.clone()))
+        .then({
+            |app: Arc<App>| async move {
+                let _ = app.pause().await;
+                "".to_owned()
+            }
+        });
+
+    let now_playing = warp::path!("playing").map({
+        let app = app.clone();
+        move || serde_json::to_string(&app.playing()).unwrap()
+    });
+
+    let routes = root
+        .or(list_output_devices)
+        // .or(list_tracks)
+        .or(enable_track)
+        .or(disable_track)
+        .or(play_all)
+        .or(stop_all)
+        .or(pause)
+        .or(now_playing);
 
     serve(routes).run(server_addr).await;
 }
 
-fn handle_add_audio_device(state: State, props: &pipewire::spa::utils::dict::DictRef)
+fn handle_add_audio_device(app: App, props: &pipewire::spa::utils::dict::DictRef) {
-{
     if props.get("media.class") == Some("Audio/Sink") {
         if let Some(device_name) = props.get("node.description") {
-            state.add_audio(device_name.to_owned());
+            app.add_audio(device_name.to_owned());
         }
     }
 }
 
-fn pipewire_loop(state: State) -> Result<(), Box<dyn std::error::Error>> {
+/*
+fn pipewire_loop(app: App) -> Result<(), Box<dyn std::error::Error>> {
     let mainloop = MainLoop::new(None)?;
     let context = Context::new(&mainloop)?;
     let core = context.connect(None)?;
@@ -78,10 +131,10 @@ fn pipewire_loop(state: State) -> Result<(), Box<dyn std::error::Error>> {
     let _listener = registry
         .add_listener_local()
         .global({
-            let state = state.clone();
+            let app = app.clone();
             move |global_data| {
                 if let Some(props) = global_data.props {
-                    handle_add_audio_device(state.clone(), props);
+                    handle_add_audio_device(app.clone(), props);
                 }
             }
         })
@@ -91,19 +144,38 @@ fn pipewire_loop(state: State) -> Result<(), Box<dyn std::error::Error>> {
 
     Ok(())
 }
+*/
 
-fn pipewire_main(state: State) {
+/*
+fn pipewire_main(state: App) {
     pipewire_loop(state).expect("pipewire should not error");
 }
+*/
 
 #[tokio::main]
 async fn main() {
-    let state = State::default();
+    gstreamer::init();
 
-    spawn_blocking({
+    let (audio_control_tx, audio_control_rx) = tokio::sync::mpsc::channel(5);
-        let state = state.clone();
+    let (audio_status_tx, audio_status_rx) = tokio::sync::mpsc::channel(5);
-        move || pipewire_main(state)
+
+    let app = Arc::new(App::new(audio_control_tx, audio_status_rx));
+    let audio_controller = Arc::new(AudioControl::new(GStreamerBackend::default()));
+    tokio::spawn({
+        let audio_controller = audio_controller.clone();
+        async move { audio_controller.listen(audio_control_rx).await }
+    });
+    tokio::spawn({
+        let audio_controller = audio_controller.clone();
+        async move { audio_controller.report(audio_status_tx).await }
     });
 
-    server_main(state.clone()).await;
+    /*
+    spawn_blocking({
+        let app = app.clone();
+        move || pipewire_main(state)
+    });
+    */
+
+    server_main(app.clone()).await;
 }

gm-dash/server/src/types.rs

@@ -0,0 +1,102 @@
+use std::{ops::Deref, path::PathBuf, time::Duration};
+
+use thiserror::Error;
+
+use crate::audio_control::AudioError;
+
+#[derive(Debug, Error, PartialEq)]
+pub enum AppError {
+    #[error("Operation invalid with no tracks enabled")]
+    NoTracks,
+
+    #[error("Operation is invalid in the current state")]
+    InvalidState,
+}
+
+impl From<AudioError> for AppError {
+    fn from(err: AudioError) -> Self {
+        match err {
+            AudioError::NoTracks => Self::NoTracks,
+            AudioError::InvalidState => Self::InvalidState,
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, PartialOrd)]
+pub struct TrackSpec {
+    pub path: PathBuf,
+    pub volume: Volume,
+}
+
+#[derive(Clone, Debug)]
+pub struct Progress {
+    pub current: Duration,
+    pub length: Duration,
+}
+
+#[derive(Debug, Error, PartialEq)]
+pub enum VolumeError {
+    #[error("The specified volume is out of range and must be between 0.0 and 1.0")]
+    OutOfRange,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, PartialOrd)]
+pub struct Volume(f32);
+
+impl Volume {
+    pub fn as_f32(&self) -> f32 {
+        self.0
+    }
+
+    pub fn as_f64(&self) -> f64 {
+        self.0.into()
+    }
+}
+
+impl TryFrom<f32> for Volume {
+    type Error = VolumeError;
+    fn try_from(val: f32) -> Result<Self, Self::Error> {
+        if val < 0. || val > 1. {
+            return Err(VolumeError::OutOfRange);
+        }
+
+        Ok(Self(val))
+    }
+}
+
+impl From<Volume> for f32 {
+    fn from(val: Volume) -> f32 {
+        val.0
+    }
+}
+
+#[derive(Clone, Debug)]
+pub enum AudioControlMessage {
+    Play,
+    Pause,
+    Stop,
+    EnableTrack(TrackSpec),
+    DisableTrack(PathBuf),
+    ReportStatus,
+}
+
+#[derive(Clone, Debug)]
+pub struct TrackInfo {
+    pub path: PathBuf,
+    pub volume: Volume,
+    pub progress: Progress,
+}
+
+#[derive(Clone, Debug)]
+pub struct AudioState {
+    pub playing: bool,
+    pub tracks: Vec<TrackInfo>,
+}
+
+#[derive(Debug)]
+pub enum AudioStatusMessage {
+    Playing,
+    Pausing,
+    Status(AudioState),
+    AudioError(AudioError),
+}

rust-toolchain

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.77.0"
+channel = "1.80.1"
 targets = [ "wasm32-unknown-unknown", "thumbv6m-none-eabi" ]