From 7e7ccdd6d19d32651b9327eb4e5bb1d861ea761d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Marc-Andr=C3=A9=20Lureau?= <marcandre.lureau@redhat.com>
Date: Thu, 23 Jan 2025 00:16:42 +0400
Subject: [PATCH] feat(rdp): add Opus codec
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
---
 Cargo.lock                   |  22 +++++++
 qemu-rdp/Cargo.toml          |   1 +
 qemu-rdp/src/server/sound.rs | 107 ++++++++++++++++++++++++++++-------
 3 files changed, 110 insertions(+), 20 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index edcd748..df12993 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -450,6 +450,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "audiopus_sys"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62314a1546a2064e033665d658e88c620a62904be945f8147e6b16c3db9f8651"
+dependencies = [
+ "cmake",
+ "log 0.4.25",
+ "pkg-config",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.4.0"
@@ -3151,6 +3162,16 @@ dependencies = [
  "paste",
 ]
 
+[[package]]
+name = "opus"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6526409b274a7e98e55ff59d96aafd38e6cd34d46b7dbbc32ce126dffcd75e8e"
+dependencies = [
+ "audiopus_sys",
+ "libc",
+]
+
 [[package]]
 name = "ordered-stream"
 version = "0.2.0"
@@ -3688,6 +3709,7 @@ dependencies = [
  "futures-util",
  "ironrdp",
  "keycodemap",
+ "opus",
  "pixman-sys",
  "qemu-display",
  "rustc_tools_util",
diff --git a/qemu-rdp/Cargo.toml b/qemu-rdp/Cargo.toml
index 16fafff..6088980 100644
--- a/qemu-rdp/Cargo.toml
+++ b/qemu-rdp/Cargo.toml
@@ -34,6 +34,7 @@ zbus.workspace = true
 enumflags2 = "0.7.10"
 dirs = "6.0"
 rustc_tools_util = "0.4.0"
+opus = "0.3"
 
 [build-dependencies]
 rustc_tools_util = "0.4.0"
diff --git a/qemu-rdp/src/server/sound.rs b/qemu-rdp/src/server/sound.rs
index 55c7c5f..4cdb982 100644
--- a/qemu-rdp/src/server/sound.rs
+++ b/qemu-rdp/src/server/sound.rs
@@ -26,9 +26,11 @@ enum State {
 #[derive(Debug)]
 pub struct Inner {
     state: State,
+    dbus_nsamples: Option<u32>,
     start_time: Instant,
     ev_sender: Option<mpsc::UnboundedSender<ServerEvent>>,
-    rdp_started: bool,
+    client_fmt: Option<pdu::AudioFormat>,
+    opus_enc: opus::Encoder,
     _audio: Option<Audio>,
 }
 
@@ -102,14 +104,41 @@ impl AudioOutHandler for DBusHandler {
         debug!(?id, ?volume)
     }
 
-    async fn write(&mut self, id: u64, data: Vec<u8>) {
-        let inner = self.inner.lock().unwrap();
+    async fn write(&mut self, id: u64, mut data: Vec<u8>) {
+        let mut inner = self.inner.lock().unwrap();
 
-        if !inner.rdp_started || !matches!(inner.state, State::Ready(iid, _) if iid == id) {
+        let State::Ready(iid, fmt) = &inner.state else {
+            debug!("QEMU stream is not ready");
+            return;
+        };
+        let Some(client_fmt) = &inner.client_fmt else {
+            debug!("RDP client has not started and negotiated a compatible format");
+            return;
+        };
+        if *iid != id {
+            return;
+        }
+        // TODO: handle format conversion?
+        if client_fmt.n_channels != fmt.n_channels
+            || client_fmt.n_samples_per_sec != fmt.n_samples_per_sec
+            || client_fmt.bits_per_sample != fmt.bits_per_sample
+        {
+            debug!("Incompatible client and server formats");
             return;
         }
 
-        // TODO: handle format conversion
+        if client_fmt.format == pdu::WaveFormat::OPUS {
+            let input =
+                unsafe { std::slice::from_raw_parts(data.as_ptr() as *const i16, data.len() / 2) };
+            data = match inner.opus_enc.encode_vec(input, data.len()) {
+                Ok(data) => data,
+                Err(err) => {
+                    warn!("Failed to encode with Opus: {}", err);
+                    return;
+                }
+            }
+        }
+
         if let Some(sender) = inner.ev_sender.as_ref() {
             let ts = inner.start_time.elapsed().as_millis() as _;
             let _ = sender.send(ServerEvent::Rdpsnd(RdpsndServerMessage::Wave(data, ts)));
@@ -135,28 +164,63 @@ pub(crate) struct RDPSndHandler {
     inner: Arc<Mutex<Inner>>,
 }
 
+impl RDPSndHandler {
+    fn choose_format(&self, client_formats: &[pdu::AudioFormat]) -> Option<u16> {
+        for (n, fmt) in client_formats.iter().enumerate() {
+            if self.get_formats().contains(fmt) {
+                return u16::try_from(n).ok();
+            }
+        }
+        None
+    }
+}
+
+const OPUS_48K: pdu::AudioFormat = pdu::AudioFormat {
+    format: pdu::WaveFormat::OPUS,
+    n_channels: 2,
+    n_samples_per_sec: 48000,
+    n_avg_bytes_per_sec: 192000,
+    n_block_align: 4,
+    bits_per_sample: 16,
+    data: None,
+};
+
+const PCM_48K: pdu::AudioFormat = pdu::AudioFormat {
+    format: pdu::WaveFormat::PCM,
+    n_channels: 2,
+    n_samples_per_sec: 48000,
+    n_avg_bytes_per_sec: 192000,
+    n_block_align: 4,
+    bits_per_sample: 16,
+    data: None,
+};
+
 impl RdpsndServerHandler for RDPSndHandler {
     fn get_formats(&self) -> &[pdu::AudioFormat] {
-        &[pdu::AudioFormat {
-            format: pdu::WaveFormat::PCM,
-            n_channels: 2,
-            n_samples_per_sec: 48000,
-            n_avg_bytes_per_sec: 192000,
-            n_block_align: 4,
-            bits_per_sample: 16,
-            data: None,
-        }]
+        let inner = self.inner.lock().unwrap();
+
+        if let Some(nsamples) = &inner.dbus_nsamples {
+            if [120u32, 240, 480, 960].contains(nsamples) {
+                return &[OPUS_48K, PCM_48K];
+            }
+        }
+
+        debug!("No Opus codec since DBus has incompatible nsamples");
+        &[PCM_48K]
     }
 
-    fn start(&mut self, _client_format: &pdu::ClientAudioFormatPdu) -> Option<u16> {
-        let mut inner = self.inner.lock().unwrap();
-        inner.rdp_started = true;
-        Some(0)
+    fn start(&mut self, client_format: &pdu::ClientAudioFormatPdu) -> Option<u16> {
+        let fmt = self.choose_format(&client_format.formats);
+        if let Some(n) = fmt {
+            let mut inner = self.inner.lock().unwrap();
+            inner.client_fmt = Some(client_format.formats[usize::from(n)].clone());
+        }
+        fmt
     }
 
     fn stop(&mut self) {
         let mut inner = self.inner.lock().unwrap();
-        inner.rdp_started = false;
+        inner.client_fmt = None;
     }
 }
 
@@ -172,11 +236,14 @@ impl SoundHandler {
     pub async fn connect(display: &Display<'_>) -> Result<Self> {
         let mut audio =
             Audio::new(display.connection(), Some(display.destination().to_owned())).await?;
+        let dbus_nsamples = audio.proxy.nsamples().await.ok();
         let inner = Arc::new(Mutex::new(Inner {
             start_time: Instant::now(),
+            dbus_nsamples,
             state: State::Init,
             ev_sender: None,
-            rdp_started: false,
+            client_fmt: None,
+            opus_enc: opus::Encoder::new(48000, opus::Channels::Stereo, opus::Application::Audio)?,
             _audio: None,
         }));