feat(rdpsnd): add Opus audio client decoding (#661)

Signed-off-by: Marc-André Lureau <marcandre.lureau@redhat.com>
Co-authored-by: Benoît Cortier <3809077+CBenoit@users.noreply.github.com>

Cargo.lock

@@ -2656,8 +2656,10 @@ name = "ironrdp-rdpsnd-native"
 version = "0.1.2"
 dependencies = [
  "anyhow",
+ "bytemuck",
  "cpal",
  "ironrdp-rdpsnd",
+ "opus",
  "tracing",
  "tracing-subscriber",
 ]

crates/ironrdp-rdpsnd-native/Cargo.toml

@@ -14,10 +14,16 @@ categories.workspace = true
 doctest = false
 test = false
 
+[features]
+default = ["opus"]
+opus = ["dep:opus", "dep:bytemuck"]
+
 [dependencies]
 anyhow = "1"
+bytemuck = { version = "1.21", optional = true }
 cpal = "0.15.3"
 ironrdp-rdpsnd.workspace = true
+opus = { version = "0.3", optional = true }
 tracing.workspace = true
 
 [dev-dependencies]

crates/ironrdp-rdpsnd-native/examples/cpal.rs

@@ -1,3 +1,5 @@
+#![allow(unused_crate_dependencies)] // opus, false negative because it's a separate binary :/
+
 use core::time::Duration;
 use std::sync::mpsc;
 use std::thread;
@@ -5,7 +7,7 @@ use std::thread;
 use anyhow::Context;
 use cpal::traits::StreamTrait;
 use ironrdp_rdpsnd::pdu::{AudioFormat, WaveFormat};
-use ironrdp_rdpsnd_native::cpal::make_stream;
+use ironrdp_rdpsnd_native::cpal::DecodeStream;
 use tracing::debug;
 
 fn setup_logging() -> anyhow::Result<()> {
@@ -41,7 +43,7 @@ fn main() -> anyhow::Result<()> {
         data: None,
     };
     let (tx, rx) = mpsc::channel();
-    let stream = make_stream(&rx_format, rx).unwrap();
+    let stream = DecodeStream::new(&rx_format, rx).unwrap();
 
     let producer = thread::spawn(move || {
         let data_chunks = vec![vec![1u8, 2, 3], vec![4, 5, 6], vec![7, 8, 9]];
@@ -52,7 +54,7 @@ fn main() -> anyhow::Result<()> {
         }
     });
 
-    stream.play()?;
+    stream.stream.play()?;
     thread::sleep(Duration::from_secs(3));
     let _ = producer.join();
 

crates/ironrdp-rdpsnd-native/src/cpal.rs

@@ -1,3 +1,4 @@
+use core::mem::size_of;
 use core::sync::atomic::{AtomicBool, Ordering};
 use core::time::Duration;
 use std::borrow::Cow;
@@ -44,6 +45,30 @@ impl Drop for RdpsndBackend {
 }
 
 impl RdpsndClientHandler for RdpsndBackend {
+    fn get_formats(&self) -> &[AudioFormat] {
+        &[
+            #[cfg(feature = "opus")]
+            AudioFormat {
+                format: 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,
+            },
+            AudioFormat {
+                format: WaveFormat::PCM,
+                n_channels: 2,
+                n_samples_per_sec: 44100,
+                n_avg_bytes_per_sec: 176400,
+                n_block_align: 4,
+                bits_per_sample: 16,
+                data: None,
+            },
+        ]
+    }
+
     fn wave(&mut self, format: &AudioFormat, _ts: u32, data: Cow<'_, [u8]>) {
         if Some(format) != self.format.as_ref() {
             debug!(?format, "New audio format");
@@ -58,7 +83,7 @@ impl RdpsndClientHandler for RdpsndBackend {
             self.stream_ended.store(false, Ordering::Relaxed);
             let stream_ended = Arc::clone(&self.stream_ended);
             self.stream_handle = Some(thread::spawn(move || {
-                let stream = match make_stream(&format, rx) {
+                let stream = match DecodeStream::new(&format, rx) {
                     Ok(stream) => stream,
                     Err(e) => {
                         error!(error = format!("{e:#}"));
@@ -100,48 +125,81 @@ impl RdpsndClientHandler for RdpsndBackend {
 }
 
 #[doc(hidden)]
-pub fn make_stream(rx_format: &AudioFormat, rx: Receiver<Vec<u8>>) -> anyhow::Result<Stream> {
-    if rx_format.format != WaveFormat::PCM {
-        bail!("only PCM formats supported");
-    }
-    let sample_format = match rx_format.bits_per_sample {
-        8 => SampleFormat::U8,
-        16 => SampleFormat::I16,
-        _ => {
-            bail!("only PCM 8/16 bits formats supported");
+pub struct DecodeStream {
+    _dec_thread: Option<JoinHandle<()>>,
+    pub stream: Stream,
+}
+
+impl DecodeStream {
+    pub fn new(rx_format: &AudioFormat, mut rx: Receiver<Vec<u8>>) -> anyhow::Result<Self> {
+        let mut dec_thread = None;
+        match rx_format.format {
+            #[cfg(feature = "opus")]
+            WaveFormat::OPUS => {
+                let chan = match rx_format.n_channels {
+                    1 => opus::Channels::Mono,
+                    2 => opus::Channels::Stereo,
+                    _ => bail!("unsupported #channels for Opus"),
+                };
+                let (dec_tx, dec_rx) = mpsc::channel();
+                let mut dec = opus::Decoder::new(rx_format.n_samples_per_sec, chan)?;
+                dec_thread = Some(thread::spawn(move || {
+                    while let Ok(pkt) = rx.recv() {
+                        let nb_samples = dec.get_nb_samples(&pkt).unwrap();
+                        let mut pcm = vec![0u8; nb_samples * chan as usize * size_of::<i16>()];
+                        dec.decode(&pkt, bytemuck::cast_slice_mut(pcm.as_mut_slice()), false)
+                            .unwrap();
+                        dec_tx.send(pcm).unwrap();
+                    }
+                }));
+                rx = dec_rx;
+            }
+            WaveFormat::PCM => {}
+            _ => bail!("audio format not supported"),
         }
-    };
 
-    let host = cpal::default_host();
-    let device = host.default_output_device().context("no default output device")?;
-    let _supported_configs_range = device
-        .supported_output_configs()
-        .context("no supported output config")?;
-    let default_config = device.default_output_config()?;
-    debug!(?default_config);
+        let sample_format = match rx_format.bits_per_sample {
+            8 => SampleFormat::U8,
+            16 => SampleFormat::I16,
+            _ => {
+                bail!("only PCM 8/16 bits formats supported");
+            }
+        };
 
-    let mut rx = RxBuffer::new(rx);
-    let config = StreamConfig {
-        channels: rx_format.n_channels,
-        sample_rate: cpal::SampleRate(rx_format.n_samples_per_sec),
-        buffer_size: cpal::BufferSize::Default,
-    };
-    debug!(?config);
+        let host = cpal::default_host();
+        let device = host.default_output_device().context("no default output device")?;
+        let _supported_configs_range = device
+            .supported_output_configs()
+            .context("no supported output config")?;
+        let default_config = device.default_output_config()?;
+        debug!(?default_config);
 
-    let stream = device
-        .build_output_stream_raw(
-            &config,
-            sample_format,
-            move |data, _info: &cpal::OutputCallbackInfo| {
-                let data = data.bytes_mut();
-                rx.fill(data)
-            },
-            |error| error!(%error),
-            None,
-        )
-        .context("failed to setup output stream")?;
+        let mut rx = RxBuffer::new(rx);
+        let config = StreamConfig {
+            channels: rx_format.n_channels,
+            sample_rate: cpal::SampleRate(rx_format.n_samples_per_sec),
+            buffer_size: cpal::BufferSize::Default,
+        };
+        debug!(?config);
 
-    Ok(stream)
+        let stream = device
+            .build_output_stream_raw(
+                &config,
+                sample_format,
+                move |data, _info: &cpal::OutputCallbackInfo| {
+                    let data = data.bytes_mut();
+                    rx.fill(data)
+                },
+                |error| error!(%error),
+                None,
+            )
+            .context("failed to setup output stream")?;
+
+        Ok(Self {
+            _dec_thread: dec_thread,
+            stream,
+        })
+    }
 }
 
 struct RxBuffer {

crates/ironrdp-rdpsnd/src/client.rs

@@ -10,6 +10,8 @@ use crate::pdu::{self, AudioFormat, PitchPdu, ServerAudioFormatPdu, TrainingPdu,
 use crate::server::RdpsndSvcMessages;
 
 pub trait RdpsndClientHandler: Send + core::fmt::Debug {
+    fn get_formats(&self) -> &[AudioFormat];
+
     fn wave(&mut self, format: &AudioFormat, ts: u32, data: Cow<'_, [u8]>);
 
     fn set_volume(&mut self, volume: VolumePdu);
@@ -23,6 +25,10 @@ pub trait RdpsndClientHandler: Send + core::fmt::Debug {
 pub struct NoopRdpsndBackend;
 
 impl RdpsndClientHandler for NoopRdpsndBackend {
+    fn get_formats(&self) -> &[AudioFormat] {
+        &[]
+    }
+
     fn wave(&mut self, _format: &AudioFormat, _ts: u32, _data: Cow<'_, [u8]>) {}
 
     fn set_volume(&mut self, _volume: VolumePdu) {}
@@ -83,15 +89,10 @@ impl Rdpsnd {
     }
 
     pub fn client_formats(&mut self) -> PduResult<RdpsndSvcMessages> {
-        let server_format = self
-            .server_format
-            .as_ref()
-            .ok_or_else(|| pdu_other_err!("invalid state - no format"))?;
-
         let pdu = pdu::ClientAudioFormatPdu {
             version: self.version()?,
             flags: pdu::AudioFormatFlags::empty(),
-            formats: server_format.formats.clone(),
+            formats: self.handler.get_formats().to_vec(),
             volume_left: 0xFFFF,
             volume_right: 0xFFFF,
             pitch: 0x00010000,