diff --git a/Cargo.lock b/Cargo.lock index 7f877924..6ea55599 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2656,8 +2656,10 @@ name = "ironrdp-rdpsnd-native" version = "0.1.2" dependencies = [ "anyhow", + "bytemuck", "cpal", "ironrdp-rdpsnd", + "opus", "tracing", "tracing-subscriber", ] diff --git a/crates/ironrdp-rdpsnd-native/Cargo.toml b/crates/ironrdp-rdpsnd-native/Cargo.toml index 1abb9f06..c3bca454 100644 --- a/crates/ironrdp-rdpsnd-native/Cargo.toml +++ b/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] diff --git a/crates/ironrdp-rdpsnd-native/examples/cpal.rs b/crates/ironrdp-rdpsnd-native/examples/cpal.rs index d3d96d9c..ac96d11b 100644 --- a/crates/ironrdp-rdpsnd-native/examples/cpal.rs +++ b/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(); diff --git a/crates/ironrdp-rdpsnd-native/src/cpal.rs b/crates/ironrdp-rdpsnd-native/src/cpal.rs index 395ebbb1..149c5db5 100644 --- a/crates/ironrdp-rdpsnd-native/src/cpal.rs +++ b/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>) -> anyhow::Result { - 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>, + pub stream: Stream, +} + +impl DecodeStream { + pub fn new(rx_format: &AudioFormat, mut rx: Receiver>) -> anyhow::Result { + 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::()]; + 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 { diff --git a/crates/ironrdp-rdpsnd/src/client.rs b/crates/ironrdp-rdpsnd/src/client.rs index 82fcbecc..65733f06 100644 --- a/crates/ironrdp-rdpsnd/src/client.rs +++ b/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 { - 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,