Audio capture (#29)

* sender: init audio capture

* use default slint features

* remove HLS backend

* sender: x11 capture

* sender: pipewire audio capture

* remove testkit

* fmt

* chores

* fmt

* chore: linux guards

* update readme

Author: MAlba124

Cargo.lock

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = [ "fcast-lib", "common", "receiver", "sender", "scap", "scap-gstreamer", "http", "android-sender", "testkit"]
+members = [ "fcast-lib", "common", "receiver", "sender", "android-sender"]
 
 [workspace.dependencies]
 gst-video = { package = "gstreamer-video", version = "0.23.5" }
@@ -13,8 +13,8 @@ env_logger = "0.11.6"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 rand = "0.9.0"
-slint = "1.11.0"
-slint-build = "1.11.0"
+slint = "1.12.0"
+slint-build = "1.12.0"
 anyhow = "1.0.98"
 futures = "0.3"
 tokio-stream = "0.1.17"

Cargo.toml

@@ -14,12 +14,14 @@ The current platform support matrix looks like this:
 
 #### Desktop
 
-|            |Linux (Wayland)  |Linux (X11)  |Windows  |MacOS     |
-|------------|-----------------|-------------|---------|----------|
-|OMSender    |Yes              |Yes          |Yes      |No        |
-|OMReceiver  |Yes              |Yes          |Yes      |Untested  |
+|            |Linux (Wayland)  |Linux (X11)  |Windows       |MacOS     |
+|------------|-----------------|-------------|--------------|----------|
+|OMSender    |Yes              |Yes          |~~Yes~~ No^1  |No^1      |
+|OMReceiver  |Yes              |Yes          |Yes           |Untested  |
 
-OMSender can cast to other FCast receivers as well.
+1: Support is planned
+
+~~OMSender can cast to other FCast receivers as well.~~
 
 OMReceiver is also an FCast receiver.
 

README.md

@@ -77,7 +77,6 @@ fn main() {
     cargo_link!("gstaudio-1.0");
     cargo_link!("gstapp-1.0");
     cargo_link!("gstrtp-1.0");
-    cargo_link!("gstwebrtc-1.0");
 
     const DEFAULT_CLANG_VERSION: &str = "20";
     let clang_version =

android-sender/build.rs

@@ -18,12 +18,10 @@ gst-rtsp = { package = "gstreamer-rtsp", version = "0.23.5", optional = true}
 gst-rtsp-server = { package = "gstreamer-rtsp-server", version = "0.23.5", optional = true}
 
 gio = { version = "0.20.9", optional = true }
-m3u8-rs = { version = "6.0.0", optional = true }
 futures = { workspace = true, optional = true }
 tokio-stream = { workspace = true, optional = true }
 mdns-sd = { workspace = true, optional = true }
 
-http = { path = "../http", optional = true }
 fcast-lib = { path = "../fcast-lib", optional = true }
 crossbeam-channel = { version = "0.5.15", optional = true }
 
@@ -42,11 +40,5 @@ glutin_glx_sys = { version = "0.6.1", optional = true }
 
 [features]
 video = ["gst", "gst-app", "gst-video", "gst-gl", "gst-gl-egl", "gst-gl-x11", "glutin_egl_sys", "glutin_glx_sys",]
-sender = ["gst", "gst-video", "gio", "m3u8-rs", "gst-pbutils",
-        "http", "crossbeam-channel", "gst-app", "futures",
-        "fcast-lib", "tokio-stream", "mdns-sd", "gst-gl-egl", "gst-gl-x11",
-        "glutin_egl_sys", "glutin_glx_sys", "gst-rtsp", "gst-rtsp-server"]
-
-[dev-dependencies]
-quickcheck = "1"
-quickcheck_macros = "1"
+sender = ["gst", "gst-video", "gio", "gst-pbutils", "crossbeam-channel", "gst-app", "futures",
+        "fcast-lib", "tokio-stream", "mdns-sd", "gst-rtsp", "gst-rtsp-server",]

assets/icons/reload.svg

@@ -0,0 +1,29 @@
+// TODO: try pipewiredeviceprovider and use pulsedeviceprovider as fallback
+// TODO: monitor for changes
+#[cfg(target_os = "linux")]
+pub fn get_pulse_dev() -> anyhow::Result<gst::Device> {
+    use anyhow::bail;
+    use gst::prelude::*;
+
+    let provider = gst::DeviceProviderFactory::by_name("pulsedeviceprovider").ok_or(
+        anyhow::anyhow!("Could not find pulse device provider factory"),
+    )?;
+
+    provider.start()?;
+    let devices = provider.devices();
+    provider.stop();
+
+    for device in devices {
+        if !device.has_classes("Audio/Sink") {
+            continue;
+        }
+        let Some(props) = device.properties() else {
+            continue;
+        };
+        if props.get::<bool>("is-default") == Ok(true) {
+            return Ok(device);
+        }
+    }
+
+    bail!("No device found")
+}

common/Cargo.toml

@@ -1,3 +1,5 @@
+#[cfg(target_os = "linux")]
+pub mod audio;
 pub mod discovery;
 pub mod pipeline;
 pub mod session;

common/src/sender/audio.rs

@@ -17,7 +17,7 @@
 
 #[cfg(target_os = "android")]
 use super::transmission::rtp::RtpSink;
-use super::transmission::{self, TransmissionSink, hls::HlsSink};
+use super::transmission::{self, TransmissionSink};
 use anyhow::Result;
 use fcast_lib::models::PlayMessage;
 #[cfg(target_os = "android")]
@@ -28,6 +28,7 @@ use log::error;
 #[cfg(target_os = "android")]
 use std::future::Future;
 use std::net::IpAddr;
+use std::str::FromStr;
 
 pub use transmission::init;
 
@@ -37,13 +38,19 @@ pub enum Event {
     Error,
 }
 
+pub enum SourceConfig {
+    AudioVideo {
+        video: gst::Element,
+        audio: gst::Element,
+    },
+    Video(gst::Element),
+    Audio(gst::Element),
+}
+
 #[cfg(not(target_os = "android"))]
 pub struct Pipeline {
     inner: gst::Pipeline,
-    tx_sink: Option<Box<dyn TransmissionSink>>,
-    tee: gst::Element,
-    preview_queue: gst::Element,
-    preview_appsink: gst::Element,
+    tx_sink: Box<dyn TransmissionSink>,
 }
 
 #[cfg(target_os = "android")]
@@ -171,58 +178,69 @@ impl Pipeline {
         })
     }
 
+    fn setup_video_source(pipeline: &gst::Pipeline, src: gst::Element) -> Result<gst::Element> {
+        // TODO: needed?
+        let videoflip = gst::ElementFactory::make("videoflip")
+            .property_from_str("video-direction", "auto")
+            .build()?;
+        let videorate = gst::ElementFactory::make("videorate")
+            .property("skip-to-first", true)
+            .build()?;
+        let capsfilter = gst::ElementFactory::make("capsfilter")
+            .property("caps", gst::Caps::from_str("video/x-raw,framerate=30/1")?)
+            .build()?;
+
+        // pipeline.add_many([&src, &videorate, &capsfilter])?;
+        // gst::Element::link_many([&src, &videorate, &capsfilter])?;
+
+        pipeline.add_many([&src, &videoflip, &videorate, &capsfilter])?;
+        gst::Element::link_many([&src, &videoflip, &videorate, &capsfilter])?;
+
+        Ok(capsfilter)
+    }
+
+    fn setup_audio_source(pipeline: &gst::Pipeline, src: gst::Element) -> Result<gst::Element> {
+        let capsfilter = gst::ElementFactory::make("capsfilter")
+            .property(
+                "caps",
+                gst::Caps::from_str("audio/x-raw,channels=2,rate=48000")?,
+            )
+            .build()?;
+
+        pipeline.add_many([&src, &capsfilter])?;
+        gst::Element::link_many([&src, &capsfilter])?;
+
+        Ok(capsfilter)
+    }
+
     #[cfg(not(target_os = "android"))]
-    pub fn new<E, S>(preview_appsink: gst::Element, mut on_event: E, on_sources: S) -> Result<Self>
+    pub fn new_rtsp<E>(mut on_event: E, source: SourceConfig) -> Result<Self>
     where
         E: FnMut(Event) + Send + Clone + 'static,
-        S: Fn(&[gst::glib::Value]) -> Option<gst::glib::Value> + Send + Sync + 'static,
     {
-        let scapsrc = gst::ElementFactory::make("scapsrc")
-            .property("perform-internal-preroll", true)
-            .build()?;
-        let tee = gst::ElementFactory::make("tee").build()?;
-        let preview_queue = gst::ElementFactory::make("queue")
-            .name("preview_queue")
-            .property("max-size-time", 0u64)
-            .property("max-size-buffers", 0u32)
-            .property("max-size-bytes", 0u32)
-            .property_from_str("leaky", "downstream")
-            .property("silent", true) // Don't emit signals, can give better perf.
-            .build()?;
+        use crate::sender::transmission::rtsp::RtspSink;
 
         let pipeline = gst::Pipeline::new();
 
-        let tx_sink = None::<Box<dyn TransmissionSink>>;
-
-        // https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/3993
-        scapsrc.static_pad("src").unwrap().add_probe(
-            gst::PadProbeType::QUERY_UPSTREAM.union(gst::PadProbeType::PUSH),
-            |_pad, info| match info.query_mut().map(|query| query.view_mut()) {
-                Some(gst::QueryViewMut::Latency(latency)) => {
-                    let (_live, min, max) = latency.result();
-                    latency.set(false, min, max);
-                    gst::PadProbeReturn::Handled
-                }
-                _ => gst::PadProbeReturn::Pass,
+        let source = match source {
+            SourceConfig::AudioVideo { video, audio } => SourceConfig::AudioVideo {
+                video: Self::setup_video_source(&pipeline, video)?,
+                audio: Self::setup_audio_source(&pipeline, audio)?,
             },
-        );
-
-        scapsrc.connect("select-source", false, on_sources);
-
-        pipeline.add_many([&scapsrc, &tee, &preview_queue, &preview_appsink])?;
-        gst::Element::link_many([&scapsrc, &tee])?;
-        gst::Element::link_many([&preview_queue, &preview_appsink])?;
+            SourceConfig::Video(video) => {
+                SourceConfig::Video(Self::setup_video_source(&pipeline, video)?)
+            }
+            SourceConfig::Audio(audio) => {
+                SourceConfig::Audio(Self::setup_audio_source(&pipeline, audio)?)
+            }
+        };
 
-        let tee_preview_pad = tee
-            .request_pad_simple("src_%u")
-            .map_or_else(|| Err(anyhow::anyhow!("`request_pad_simple()` failed")), Ok)?;
-        let queue_preview_pad = preview_queue
-            .static_pad("sink")
-            .ok_or(anyhow::anyhow!("preview_queue is missing static sink pad"))?;
-        tee_preview_pad.link(&queue_preview_pad)?;
+        let rtsp = RtspSink::new(&pipeline, source, 3000)?;
+        let p = Self {
+            inner: pipeline.clone(),
+            tx_sink: Box::new(rtsp),
+        };
 
-        // Start the pipeline in background thread because `scapsrc` initialization will block until
-        // the user selects the input source.
         let _ = std::thread::spawn({
             let bus = pipeline
                 .bus()
@@ -236,9 +254,11 @@ impl Pipeline {
                         debug!("Failed to upgrade pipeline before starting");
                         return;
                     };
-                    debug!("Starting pipeline");
+                    debug!("Starting pipeline...");
                     if let Err(err) = pipeline.set_state(gst::State::Playing) {
                         error!("Failed to start pipeline: {err}");
+                    } else {
+                        debug!("Pipeline started");
                     }
                 }
 
@@ -278,20 +298,11 @@ impl Pipeline {
             }
         });
 
-        Ok(Self {
-            inner: pipeline,
-            tx_sink,
-            tee,
-            preview_queue,
-            preview_appsink,
-        })
+        Ok(p)
     }
 
     pub fn playing(&mut self) -> Result<()> {
-        match &mut self.tx_sink {
-            Some(sink) => sink.playing(),
-            None => Ok(()),
-        }
+        self.tx_sink.playing()
     }
 
     #[cfg(target_os = "android")]
@@ -308,27 +319,7 @@ impl Pipeline {
     #[cfg(not(target_os = "android"))]
     pub fn shutdown(&mut self) -> Result<()> {
         self.inner.set_state(gst::State::Null)?;
-
-        self.preview_queue.unlink(&self.preview_appsink);
-        self.inner.remove(&self.preview_appsink)?;
-
-        if let Some(sink) = &mut self.tx_sink {
-            sink.shutdown();
-        }
-
-        Ok(())
-    }
-
-    #[cfg(not(target_os = "android"))]
-    pub fn add_hls_sink(&mut self, port: u16) -> Result<()> {
-        let tee_pad = self
-            .tee
-            .request_pad_simple("src_%u")
-            .ok_or(anyhow::anyhow!("`request_pad_simple()` failed"))?;
-        let hls = HlsSink::new(&self.inner, tee_pad, port)?;
-        self.tx_sink = Some(Box::new(hls));
-
-        debug!("Added HLS sink");
+        self.tx_sink.shutdown();
 
         Ok(())
     }
@@ -348,20 +339,6 @@ impl Pipeline {
         Ok(())
     }
 
-    #[cfg(not(target_os = "android"))]
-    pub fn add_rtsp_sink(&mut self, port: u16) -> Result<()> {
-        let tee_pad = self
-            .tee
-            .request_pad_simple("src_%u")
-            .ok_or(anyhow::anyhow!("`request_pad_simple()` failed"))?;
-        let rtsp = transmission::rtsp::RtspSink::new(tee_pad, &self.inner, port)?;
-        self.tx_sink = Some(Box::new(rtsp));
-
-        debug!("Added RTSP sink");
-
-        Ok(())
-    }
-
     #[cfg(target_os = "android")]
     pub fn add_rtp_sink(&mut self, port: u16, receiver_addr: IpAddr) -> Result<()> {
         let appsrc_pad = self
@@ -376,26 +353,9 @@ impl Pipeline {
         Ok(())
     }
 
-    pub fn remove_transmission_sink(&mut self) -> Result<()> {
-        if let Some(sink) = &mut self.tx_sink {
-            sink.shutdown();
-            sink.unlink(&self.inner)?;
-        }
-
-        self.tx_sink = None;
-
-        debug!("Removed transmission sink");
-
-        Ok(())
-    }
-
     /// Get the message that should be sent to a receiver to consume the stream if a transmission
     /// sink is present
     pub fn get_play_msg(&self, addr: IpAddr) -> Option<PlayMessage> {
-        if let Some(sink) = &self.tx_sink {
-            sink.get_play_msg(addr)
-        } else {
-            None
-        }
+        self.tx_sink.get_play_msg(addr)
     }
 }