PipeWire implementation

.github/workflows/cpal.yml

@@ -14,6 +14,8 @@ jobs:
       run: sudo apt-get install libasound2-dev
     - name: Install libjack
       run: sudo apt-get install libjack-jackd2-dev libjack-jackd2-0
+    - name: Install libpipewire
+      run: sudo apt-get install libpipewire-0.3-dev
     - name: Install stable
       uses: dtolnay/rust-toolchain@stable
       with:
@@ -67,6 +69,8 @@ jobs:
       run: sudo apt-get install libasound2-dev
     - name: Install libjack
       run: sudo apt-get install libjack-jackd2-dev libjack-jackd2-0
+    - name: Install libpipewire
+      run: sudo apt-get install libpipewire-0.3-dev
     - name: Install stable
       uses: dtolnay/rust-toolchain@stable
     - name: Run without features

.gitignore

@@ -1,5 +1,5 @@
-/target
-/Cargo.lock
+target/
+Cargo.lock
 .cargo/
 .DS_Store
 recorded.wav

Cargo.toml

@@ -12,6 +12,8 @@ rust-version = "1.70"
 [features]
 asio = ["asio-sys", "num-traits"] # Only available on Windows. See README for setup instructions.
 oboe-shared-stdcxx = ["oboe/shared-stdcxx"] # Only available on Android. See README for what it does.
+jack = ["dep:jack"]
+pipewire = ["dep:pipewire-client"]
 
 [dependencies]
 dasp_sample = "0.11"
@@ -46,6 +48,7 @@ num-traits = { version = "0.2.6", optional = true }
 alsa = "0.9"
 libc = "0.2"
 jack = { version = "0.13.0", optional = true }
+pipewire-client = { version = "0.1", path = "pipewire-client", optional = true }
 
 [target.'cfg(any(target_os = "macos", target_os = "ios"))'.dependencies]
 core-foundation-sys = "0.8.2" # For linking to CoreFoundation.framework and handling device name `CFString`s.

examples/beep.rs

@@ -24,6 +24,20 @@ struct Opt {
     #[arg(short, long)]
     #[allow(dead_code)]
     jack: bool,
+
+    /// Use the PipeWire host
+    #[cfg(all(
+        any(
+            target_os = "linux",
+            target_os = "dragonfly",
+            target_os = "freebsd",
+            target_os = "netbsd"
+        ),
+        feature = "pipewire"
+    ))]
+    #[arg(short, long)]
+    #[allow(dead_code)]
+    pipewire: bool,
 }
 
 fn main() -> anyhow::Result<()> {
@@ -52,14 +66,40 @@ fn main() -> anyhow::Result<()> {
         cpal::default_host()
     };
 
+    // Conditionally compile with pipewire if the feature is specified.
+    #[cfg(all(
+        any(
+            target_os = "linux",
+            target_os = "dragonfly",
+            target_os = "freebsd",
+            target_os = "netbsd"
+        ),
+        feature = "pipewire"
+    ))]
+    // Manually check for flags. Can be passed through cargo with -- e.g.
+    // cargo run --release --example beep --features pipewire -- --pipewire
+    let host = if opt.pipewire {
+        cpal::host_from_id(cpal::available_hosts()
+            .into_iter()
+            .find(|id| *id == cpal::HostId::PipeWire)
+            .expect(
+                "make sure --features pipewire is specified. only works on OSes where pipewire is available",
+            )).expect("pipewire host unavailable")
+    } else {
+        cpal::default_host()
+    };
+
     #[cfg(any(
         not(any(
             target_os = "linux",
             target_os = "dragonfly",
             target_os = "freebsd",
             target_os = "netbsd"
         )),
-        not(feature = "jack")
+        not(any(
+            feature = "jack",
+            feature = "pipewire",
+        ))
     ))]
     let host = cpal::default_host();
 

examples/synth_tones.rs

@@ -114,6 +114,51 @@ where
 
 pub fn host_device_setup(
 ) -> Result<(cpal::Host, cpal::Device, cpal::SupportedStreamConfig), anyhow::Error> {
+    #[cfg(all(
+        any(
+            target_os = "linux",
+            target_os = "dragonfly",
+            target_os = "freebsd",
+            target_os = "netbsd"
+        ),
+        feature = "jack"
+    ))]
+    // Manually check for flags. Can be passed through cargo with -- e.g.
+    // cargo run --release --example beep --features jack -- --jack
+    let host = cpal::host_from_id(cpal::available_hosts()
+            .into_iter()
+            .find(|id| *id == cpal::HostId::Jack)
+            .expect(
+                "make sure --features jack is specified. only works on OSes where jack is available",
+            )).expect("jack host unavailable");
+    #[cfg(all(
+        any(
+            target_os = "linux",
+            target_os = "dragonfly",
+            target_os = "freebsd",
+            target_os = "netbsd"
+        ),
+        feature = "pipewire"
+    ))]
+    let host = cpal::host_from_id(cpal::available_hosts()
+        .into_iter()
+        .find(|id| *id == cpal::HostId::PipeWire)
+        .expect(
+            "make sure --features pipewire is specified. only works on OSes where pipewire is available",
+        )).expect("pipewire host unavailable");
+
+    #[cfg(any(
+        not(any(
+            target_os = "linux",
+            target_os = "dragonfly",
+            target_os = "freebsd",
+            target_os = "netbsd"
+        )),
+        not(any(
+            feature = "jack",
+            feature = "pipewire",
+        ))
+    ))]
     let host = cpal::default_host();
 
     let device = host

pipewire-client/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "pipewire-client"
+version = "0.1.0"
+edition = "2021"
+authors = ["Alexis Bekhdadi <[email protected]>"]
+description = "PipeWire Client"
+repository = "https://github.com/RustAudio/cpal/"
+documentation = ""
+license = "Apache-2.0"
+keywords = ["pipewire", "client"]
+
+[dependencies]
+pipewire = { version = "0.8" }
+pipewire-spa-utils = { version = "0.1", path = "../pipewire-spa-utils"}
+pipewire-common = { version = "0.1", path = "../pipewire-common" }
+serde_json = "1.0"
+crossbeam-channel = "0.5"
+uuid = { version = "1.12", features = ["v4"] }
+tokio = { version = "1", features = ["full"] }
+tokio-util = "0.7"
+libc = "0.2"
+
+[dev-dependencies]
+rstest = "0.24"
+serial_test = "3.2"
+ctor = "0.2"
+pipewire-test-utils = { version = "0.1", path = "../pipewire-test-utils" }

pipewire-client/src/client/api/core.rs

@@ -0,0 +1,90 @@
+use crate::client::api::internal::InternalApi;
+use crate::error::Error;
+use crate::messages::{MessageRequest, MessageResponse};
+use crate::states::{DefaultAudioNodesState, GlobalObjectState, SettingsState};
+use std::sync::Arc;
+
+pub struct CoreApi {
+    pub(crate) api: Arc<InternalApi>,
+}
+
+impl CoreApi {
+    pub(crate) fn new(api: Arc<InternalApi>) -> Self {
+        CoreApi {
+            api,
+        }
+    }
+
+    pub(crate) fn check_session_manager_registered(&self) -> Result<(), Error> {
+        let request = MessageRequest::CheckSessionManagerRegistered;
+        let response = self.api.send_request(&request);
+        match response {
+            Ok(MessageResponse::CheckSessionManagerRegistered{
+                   session_manager_registered,
+                   error
+            }) => {
+                if session_manager_registered {
+                    return Ok(());
+                }
+                Err(error.unwrap())
+            },
+            Err(value) => Err(value),
+            Ok(value) => Err(Error {
+                description: format!("Received unexpected response: {:?}", value),
+            }),
+        }
+    }
+
+    pub fn quit(&self) {
+        let request = MessageRequest::Quit;
+        self.api.send_request_without_response(&request).unwrap();
+    }
+
+    pub fn get_settings(&self) -> Result<SettingsState, Error> {
+        let request = MessageRequest::Settings;
+        let response = self.api.send_request(&request);
+        match response {
+            Ok(MessageResponse::Settings(value)) => Ok(value),
+            Err(value) => Err(value),
+            Ok(value) => Err(Error {
+                description: format!("Received unexpected response: {:?}", value),
+            }),
+        }
+    }
+
+    pub(crate) fn get_settings_state(&self) -> Result<GlobalObjectState, Error> {
+        let request = MessageRequest::SettingsState;
+        let response = self.api.send_request(&request);
+        match response {
+            Ok(MessageResponse::SettingsState(value)) => Ok(value),
+            Err(value) => Err(value),
+            Ok(value) => Err(Error {
+                description: format!("Received unexpected response: {:?}", value),
+            }),
+        }
+    }
+
+    pub fn get_default_audio_nodes(&self) -> Result<DefaultAudioNodesState, Error> {
+        let request = MessageRequest::DefaultAudioNodes;
+        let response = self.api.send_request(&request);
+        match response {
+            Ok(MessageResponse::DefaultAudioNodes(value)) => Ok(value),
+            Err(value) => Err(value),
+            Ok(value) => Err(Error {
+                description: format!("Received unexpected response: {:?}", value),
+            }),
+        }
+    }
+
+    pub(crate) fn get_default_audio_nodes_state(&self) -> Result<GlobalObjectState, Error> {
+        let request = MessageRequest::DefaultAudioNodesState;
+        let response = self.api.send_request(&request);
+        match response {
+            Ok(MessageResponse::DefaultAudioNodesState(value)) => Ok(value),
+            Err(value) => Err(value),
+            Ok(value) => Err(Error {
+                description: format!("Received unexpected response: {:?}", value),
+            }),
+        }
+    }
+}

pipewire-client/src/client/api/core_test.rs

@@ -0,0 +1,28 @@
+use crate::test_utils::fixtures::{isolated_client, shared_client, PipewireTestClient};
+use rstest::rstest;
+use serial_test::serial;
+
+#[rstest]
+#[serial]
+fn quit(#[from(isolated_client)] client: PipewireTestClient) {
+    client.core().quit();
+}
+
+#[rstest]
+#[serial]
+pub fn settings(#[from(shared_client)] client: PipewireTestClient) {
+    let settings = client.core().get_settings().unwrap();
+    assert_eq!(true, settings.sample_rate > u32::default());
+    assert_eq!(true, settings.default_buffer_size > u32::default());
+    assert_eq!(true, settings.min_buffer_size > u32::default());
+    assert_eq!(true, settings.max_buffer_size > u32::default());
+    assert_eq!(true, settings.allowed_sample_rates[0] > u32::default());
+}
+
+#[rstest]
+#[serial]
+pub fn default_audio_nodes(#[from(shared_client)] client: PipewireTestClient) {
+    let default_audio_nodes = client.core().get_default_audio_nodes().unwrap();
+    assert_eq!(false, default_audio_nodes.sink.is_empty());
+    assert_eq!(false, default_audio_nodes.source.is_empty());
+}

pipewire-client/src/client/api/internal.rs

@@ -0,0 +1,42 @@
+use crate::client::channel::ClientChannel;
+use crate::error::Error;
+use crate::messages::{MessageRequest, MessageResponse};
+use std::time::Duration;
+
+pub(crate) struct InternalApi {
+    pub(crate) channel: ClientChannel<MessageRequest, MessageResponse>,
+    pub(crate) timeout: Duration
+}
+
+impl InternalApi {
+    pub(crate) fn new(
+        channel: ClientChannel<MessageRequest, MessageResponse>,
+        timeout: Duration
+    ) -> Self {
+        InternalApi {
+            channel,
+            timeout,
+        }
+    }
+
+    pub(crate) fn wait_response_with_timeout(&self, timeout: Duration) -> Result<MessageResponse, Error> {
+        self.channel.receive_timeout(timeout)
+    }
+
+    pub(crate) fn send_request(&self, request: &MessageRequest) -> Result<MessageResponse, Error> {
+        let response = self.channel.send(request.clone());
+        match response {
+            Ok(value) => {
+                match value {
+                    MessageResponse::Error(value) => Err(value),
+                    _ => Ok(value)
+                }
+            }
+            Err(value) => Err(value)
+        }
+    }
+
+    pub(crate) fn send_request_without_response(&self, request: &MessageRequest) -> Result<(), Error> {
+        self.channel.fire(request.clone()).map(move |_| ())
+    }
+}

pipewire-client/src/client/api/mod.rs

@@ -0,0 +1,20 @@
+mod core;
+pub(crate) use core::CoreApi;
+#[cfg(test)]
+#[path = "core_test.rs"]
+mod core_test;
+
+mod node;
+pub(crate) use node::NodeApi;
+#[cfg(test)]
+#[path = "node_test.rs"]
+mod node_test;
+
+mod stream;
+pub(crate) use stream::StreamApi;
+#[cfg(test)]
+#[path = "stream_test.rs"]
+mod stream_test;
+
+mod internal;
+pub(crate) use internal::InternalApi;