Start setting up the audio control system

gm-dash/server/src/app.rs

@@ -1,27 +1,44 @@
-use std::{collections::HashSet, sync::{Arc, RwLock}};
+use crate::audio_control::AudioControl;
+use std::{
+    collections::HashSet,
+    sync::{Arc, RwLock},
+};
 
-struct State_ {
+struct AppState {
     device_list: Vec<String>,
     track_list: Vec<String>,
     currently_playing: HashSet<String>,
+
+    audio_control: AudioControl,
 }
 
-#[derive(Clone)]
+impl Default for AppState {
-pub struct State {
+    fn default() -> Self {
-    internal: Arc<RwLock<State_>>,
+        Self {
-}
-
-impl State {
-    fn new() -> State {
-        let internal = State_ {
             device_list: vec![],
             track_list: vec![
                 "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/01 - A Day to Rebuild.mp3.mp3".to_owned(),
+                "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/02 - Against the Clock.mp3.mp3".to_owned(),
+                "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/03 - Alleyway Cutthroat.mp3.mp3".to_owned(),
+                "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/04 - Beasts Of Legend.mp3.mp3".to_owned(),
                 "/home/savanni/Music/Travis Savoie/RPG Toolkit Volume II/05 - Books and Spellcrafting.mp3.mp3".to_owned(),
             ],
             currently_playing: HashSet::default(),
-        };
+
-        State {
+            audio_control: AudioControl::default(),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct App {
+    internal: Arc<RwLock<AppState>>,
+}
+
+impl App {
+    fn new() -> App {
+        let internal = AppState::default();
+        App {
             internal: Arc::new(RwLock::new(internal)),
         }
     }
@@ -41,11 +58,17 @@ impl State {
         st.track_list.clone()
     }
 
-    pub fn play(&self, track: String) -> Result<(), String> {
+    pub fn play_pause(&self) -> Result<(), String> {
+        self.internal.write().unwrap().audio_control.play_pause();
+        Ok(())
+    }
+
+    pub fn add_track(&self, track: String) -> Result<(), String> {
         let mut st = self.internal.write().unwrap();
         if st.track_list.contains(&track) {
-            st.currently_playing.insert(track);
+            st.currently_playing.insert(track.clone());
         }
+        st.audio_control.add_track(track);
         Ok(())
     }
 
@@ -67,10 +90,8 @@ impl State {
     }
 }
 
-impl Default for State {
+impl Default for App {
-    fn default() -> State {
+    fn default() -> App {
-        State::new()
+        App::new()
     }
 }
-
-

gm-dash/server/src/audio_control.rs

@@ -0,0 +1,138 @@
+use gstreamer::{prelude::*, ClockTime, MessageType, MessageView};
+use std::sync::{Arc, RwLock};
+
+pub struct AudioControl {
+    bus: gstreamer::Bus,
+    pipeline: gstreamer::Pipeline,
+    mixer: gstreamer::Element,
+    audio_sink: gstreamer::Element,
+
+    bus_monitor: std::thread::JoinHandle<()>,
+
+    playing: Arc<RwLock<bool>>,
+}
+
+impl Default for AudioControl {
+    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 playing = Arc::new(RwLock::new(false));
+
+        let bus_monitor = std::thread::spawn({
+            let pipeline_object = pipeline.clone().upcast::<gstreamer::Object>();
+            let playing = playing.clone();
+            let bus = bus.clone();
+            move || loop {
+                if let Some(msg) = bus.timed_pop_filtered(
+                    ClockTime::NONE,
+                    &[
+                        MessageType::Error,
+                        MessageType::Eos,
+                        MessageType::StateChanged,
+                    ],
+                ) {
+                    match msg.view() {
+                        MessageView::StateChanged(st) => {
+                            if msg.src() == Some(&pipeline_object) {
+                                *playing.write().unwrap() = st.current() == gstreamer::State::Playing;
+                            }
+                        }
+                        MessageView::Error(err) => {
+                            println!("error: {:?}", err);
+                        }
+                        MessageView::Eos(_) => {
+                            println!("EOS");
+                        }
+                        _ => {
+                            unreachable!();
+                        }
+                    }
+                }
+            }
+        });
+
+        Self {
+            bus,
+            pipeline,
+            mixer,
+            audio_sink,
+
+            bus_monitor,
+
+            playing,
+        }
+    }
+}
+
+impl AudioControl {
+    pub fn play_pause(&self) {
+        if *self.playing.read().unwrap() {
+            self.pipeline.set_state(gstreamer::State::Paused).unwrap();
+        } else {
+            self.pipeline.set_state(gstreamer::State::Playing).unwrap();
+        }
+    }
+
+    pub fn add_track(&mut self, path: String) {
+        let source = gstreamer::ElementFactory::find("filesrc")
+            .unwrap()
+            .load()
+            .unwrap()
+            .create()
+            .property("location", path)
+            .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", 0.75)
+            .build()
+            .unwrap();
+        self.pipeline.add(&volume).unwrap();
+        volume.link(&self.mixer).unwrap();
+
+        decoder.connect_pad_added(move |_, pad| {
+            let next_pad = volume.static_pad("sink").unwrap();
+            pad.link(&next_pad).unwrap();
+        });
+    }
+
+    pub fn remove_track(&mut self, path: String) {
+        /* Need to run EOS through to a probe on the trailing end of the volume element */
+    }
+}

gm-dash/server/src/main.rs

@@ -4,15 +4,17 @@ use std::net::{Ipv6Addr, SocketAddrV6};
 use tokio::task::spawn_blocking;
 use warp::{serve, Filter};
 
-mod state;
+mod audio_control;
-use state::State;
+
+mod app;
+use app::App;
 
 #[derive(Deserialize)]
 struct PlayTrackParams {
     track_name: String,
 }
 
-async fn server_main(state: State) {
+async fn server_main(state: App) {
     let localhost: Ipv6Addr = "::1".parse().unwrap();
     let server_addr = SocketAddrV6::new(localhost, 3001, 0, 0);
 
@@ -76,7 +78,7 @@ async fn server_main(state: State) {
     serve(routes).run(server_addr).await;
 }
 
-fn handle_add_audio_device(state: State, props: &pipewire::spa::utils::dict::DictRef) {
+fn handle_add_audio_device(state: 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());
@@ -84,7 +86,7 @@ fn handle_add_audio_device(state: State, props: &pipewire::spa::utils::dict::Dic
     }
 }
 
-fn pipewire_loop(state: State) -> Result<(), Box<dyn std::error::Error>> {
+fn pipewire_loop(state: App) -> Result<(), Box<dyn std::error::Error>> {
     let mainloop = MainLoop::new(None)?;
     let context = Context::new(&mainloop)?;
     let core = context.connect(None)?;
@@ -107,13 +109,13 @@ 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();
+    let state = App::default();
 
     spawn_blocking({
         let state = state.clone();