Set up a set of pipelines that mix two different file sources

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

@@ -1,115 +1,75 @@
 use std::time::Duration;
 
 use gstreamer::{
-    prelude::*, Bus, Element, EventType, MessageType, MessageView, Pad, PadProbeData, PadProbeInfo, PadProbeReturn, Pipeline
+    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 srcfactory = gstreamer::ElementFactory::find("filesrc").unwrap().load().unwrap();
+    let pipeline = gstreamer::Pipeline::new();
-    let source = gstreamer::ElementFactory::find("filesrc")
+    let pipeline_object = pipeline.clone().upcast::<gstreamer::Object>();
-        .unwrap()
+
-        .load()
-        .unwrap()
-        .create()
-        .name("source")
-        .property("location", "/home/savanni/Music/tavern-music.ogg")
-        // .property(
-        //     "location",
-        //     "/home/savanni/Music/Night at Work _ Instrumental Chill Music Mix [n9Y2Eb4BaSg].m4a",
-        // )
-        .build()
-        .unwrap();
-    let decoder = gstreamer::ElementFactory::find("decodebin")
-        .unwrap()
-        .load()
-        .unwrap()
-        .create()
-        .name("decoder")
-        .build()
-        .unwrap();
     let sinkfactory = gstreamer::ElementFactory::find("pulsesink")
         .unwrap()
         .load()
         .unwrap();
-    let defaultsink = sinkfactory.create().name("sink").build().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();
-    let convert = gstreamer::ElementFactory::find("audioconvert")
-        .unwrap()
-        .load()
-        .unwrap()
-        .create()
-        .name("convert")
-        .build()
-        .unwrap();
-    let resample = gstreamer::ElementFactory::find("audioresample")
-        .unwrap()
-        .load()
-        .unwrap()
-        .create()
-        .name("resample")
-        .build()
-        .unwrap();
-
-    let pipeline = gstreamer::Pipeline::new();
-    pipeline.add(&source).unwrap();
-    pipeline.add(&defaultsink).unwrap();
-    pipeline.add(&decoder).unwrap();
-    pipeline.add(&convert).unwrap();
-    pipeline.add(&resample).unwrap();
-
-    source.link(&decoder).unwrap();
-    convert.link(&resample).unwrap();
-    resample.link(&defaultsink).unwrap();
-
-    decoder.connect_pad_added(move |element, pad| handle_pad_added(element, pad, &convert));
-
-    // println!("Sink target: {:?}", sink.property::<String>("current-device"));
-    // println!("Sink target: {:?}", sink.property::<String>("device-name"));
-
-    /*
-    println!("source: {:?}", source);
-    source.foreach_pad(|_, pad| {
-        println!("\tpad: {:?} {}, {:?}", pad, pad.name(), pad.direction());
-        true
-    });
-    let source_pad = source.get_pad(pad.name());
-    source_pad
         */
 
-    // source.connect_pad_added(handle_pad_added);
-    // source.link(&convert).unwrap();
-
     pipeline.set_state(gstreamer::State::Playing).unwrap();
     let pipeline_object = pipeline.clone().upcast::<gstreamer::Object>();
 
     /*
-    std::thread::sleep(Duration::from_secs(5));
-    println!("switching audio outputs");
-
-    pipeline.set_state(gstreamer::State::Paused).unwrap();
-    resample.unlink(&defaultsink);
-    pipeline.remove(&defaultsink).unwrap();
-    pipeline.add(&btsink).unwrap();
-    resample.link(&btsink).unwrap();
-    pipeline.set_state(gstreamer::State::Playing).unwrap();
-    println!("switch complete");
-    */
-
-    let bus = pipeline.bus().unwrap();
-    // let msg = bus.timed_pop_filtered(
-    //     gstreamer::ClockTime::NONE,
-    //     &[gstreamer::MessageType::Error, gstreamer::MessageType::Eos],
-    // );
-    /*
-     */
     std::thread::spawn({
         let bus = bus.clone();
         let pipeline = pipeline.clone();
@@ -118,6 +78,9 @@ fn main() {
             swap_audio_output(bus, pipeline, resample, defaultsink, btsink);
         }
     });
+    */
+
+    pipeline.set_state(gstreamer::State::Playing).unwrap();
 
     let mut playing = false;
 
@@ -163,67 +126,75 @@ fn main() {
     pipeline.set_state(gstreamer::State::Null).unwrap();
 }
 
-fn handle_pad_added(element: &Element, pad: &Pad, converter: &Element) {
+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 converter_pad = converter.static_pad("sink").unwrap();
-    pad.link(&converter_pad).unwrap();
-}
-
-fn swap_audio_output(
-    bus: Bus,
-    pipeline: Pipeline,
-    upstream: Element,
-    oldoutput: Element,
-    newoutput: Element,
-) {
-    let source_pad = upstream.static_pad("src").unwrap();
-    source_pad.add_probe(gstreamer::PadProbeType::BLOCK, move |pad, info| {
-        println!("upstream pad blocked");
     /*
-        if let Some(probe_id) = info.id {
+    let audio_caps = gstreamer::caps::Caps::builder()
-            pad.remove_probe(probe_id);
+        .field("audio", "audio/x-raw,
-        }
+        .build();
         */
-
     /*
-        if let Some(old_element_output) = oldelement.static_pad("src") {
+    let audio_pad_template = PadTemplate::new(
-            old_element_output.add_probe(gstreamer::PadProbeType::BLOCK, wait_for_eos);
+        "audio-pad-template",
-            let old_element_input = oldelement.static_pad("sink").unwrap();
+        PadDirection::Sink,
-            old_element_input.send_event(gstreamer::event::Eos::new());
+        PadPresence::Request,
-        } else {
+        &pad.current_caps().unwrap(),
-            /* This is a final sink, so we don't have to wait for EOS to go through. */
+    )
-        }
+    .unwrap();
     */
-        let oldoutput_pad = oldoutput.static_pad("sink").unwrap();
+    let next_pad = next_element.request_pad(template, None, None).unwrap();
-        // oldoutput_pad.send_event(gstreamer::event::Eos::new());
+    // let converter_pad = converter.static_pad("sink").unwrap();
-        // oldoutput.set_state(gstreamer::State::Null).unwrap();
+    pad.link(&next_pad).unwrap();
-        pipeline.remove(&oldoutput).unwrap();
-
-        pipeline.add(&newoutput).unwrap();
-        upstream.link(&newoutput).unwrap();
-
-        PadProbeReturn::Remove
-    });
 }
 
-fn wait_for_eos(pad: &Pad, info: &mut PadProbeInfo) -> PadProbeReturn {
+fn setup_file_reader(pipeline: &Pipeline, dest: Element, path: &str) {
-/*
+    let source = gstreamer::ElementFactory::find("filesrc")
-    match &info.data {
+        .unwrap()
-        Some(PadProbeData::Event(event)) if event.type_() == EventType::Eos => {
+        .load()
-            /* Do a lot of things now that this pad is empty of data. */
+        .unwrap()
-            element.set_state(gstreamer::State::Null);
+        .create()
-            pipeline.remove(element);
+        .property("location", path)
-            pipeline.add(new_element);
+        .build()
-            let new_pad = new_element.static_pad("src");
+        .unwrap();
-            pad.link(new_pad);
 
-            println!("EOS detected");
+    let decoder = gstreamer::ElementFactory::find("decodebin")
-        }
+        .unwrap()
-        _ => return PadProbeReturn::Ok,
+        .load()
-    };
+        .unwrap()
-*/
+        .create()
+        .build()
+        .unwrap();
 
-    return PadProbeReturn::Ok;
+    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();
 }