player_plugin/
audio.rs

1use std::collections::VecDeque;
2
3use serde::{Deserialize, Serialize};
4use thiserror::Error;
5
6use crate::{DecoderFrameFormat, DecoderPcmFrame, DecoderPcmFrameMetadata};
7
8const MAX_AUDIO_PROCESSOR_QUEUE_FRAMES: usize = 256;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
11pub enum AudioProcessorSubmitStatus {
12    Accepted,
13    Backpressure,
14}
15
16/// Pitch behavior requested from an audio processing chain.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
18pub enum AudioPitchMode {
19    /// Preserve the original pitch while changing playback speed.
20    PreservePitch,
21    /// Let pitch follow the playback rate.
22    FollowRate,
23}
24
25impl AudioPitchMode {
26    pub const fn wire_name(self) -> &'static str {
27        match self {
28            Self::PreservePitch => "preservePitch",
29            Self::FollowRate => "followRate",
30        }
31    }
32}
33
34/// Playback policy applied to an audio processor chain.
35#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
36pub struct AudioPlaybackPolicy {
37    pub playback_rate: f32,
38    pub pitch_mode: AudioPitchMode,
39}
40
41impl AudioPlaybackPolicy {
42    pub const fn normal() -> Self {
43        Self {
44            playback_rate: 1.0,
45            pitch_mode: AudioPitchMode::FollowRate,
46        }
47    }
48
49    pub fn validate(self) -> Result<(), AudioProcessorError> {
50        if !self.playback_rate.is_finite() || self.playback_rate <= 0.0 {
51            return Err(AudioProcessorError::InvalidPlaybackRate {
52                rate: self.playback_rate,
53            });
54        }
55        Ok(())
56    }
57}
58
59#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
60pub struct AudioProcessorCapabilities {
61    pub accepted_formats: Vec<DecoderFrameFormat>,
62    pub output_format: Option<DecoderFrameFormat>,
63    pub supports_flush: bool,
64    pub max_in_flight_frames: Option<u32>,
65    pub playback_rate_min: Option<f32>,
66    pub playback_rate_max: Option<f32>,
67    pub pitch_modes: Vec<AudioPitchMode>,
68}
69
70impl AudioProcessorCapabilities {
71    pub fn supports_input_format(&self, format: &DecoderFrameFormat) -> bool {
72        self.accepted_formats.is_empty() || self.accepted_formats.iter().any(|item| item == format)
73    }
74
75    pub fn supports_playback_policy(&self, policy: AudioPlaybackPolicy) -> bool {
76        let rate_supported = self
77            .playback_rate_min
78            .is_none_or(|minimum| policy.playback_rate >= minimum)
79            && self
80                .playback_rate_max
81                .is_none_or(|maximum| policy.playback_rate <= maximum);
82        let pitch_supported =
83            self.pitch_modes.is_empty() || self.pitch_modes.contains(&policy.pitch_mode);
84        rate_supported && pitch_supported
85    }
86}
87
88/// Configuration used to open one native audio processor session.
89#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
90pub struct AudioProcessorSessionConfig {
91    pub processor_index: usize,
92    pub input_metadata: DecoderPcmFrameMetadata,
93    pub playback_policy: AudioPlaybackPolicy,
94    #[serde(default)]
95    pub max_in_flight_frames: Option<u32>,
96}
97
98/// Metadata returned after opening one audio processor session.
99#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
100pub struct AudioProcessorSessionInfo {
101    pub processor_name: Option<String>,
102    pub selected_backend: Option<String>,
103    pub output_format: Option<DecoderFrameFormat>,
104    pub max_in_flight_frames: Option<u32>,
105}
106
107/// Empty success payload used by configure, flush, and close operations.
108#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
109pub struct AudioProcessorOperationStatus {
110    pub completed: bool,
111}
112
113#[derive(Debug, Error, Clone, PartialEq, Serialize, Deserialize)]
114pub enum AudioProcessorError {
115    #[error(
116        "audio processor queue capacity must be between 1 and {MAX_AUDIO_PROCESSOR_QUEUE_FRAMES}"
117    )]
118    InvalidCapacity,
119    #[error("audio processor chain is closed")]
120    Closed,
121    #[error("invalid PCM frame: {0}")]
122    InvalidPcm(String),
123    #[error("invalid playback rate: {rate}")]
124    InvalidPlaybackRate { rate: f32 },
125    #[error("audio processor does not support the requested playback policy")]
126    UnsupportedPlaybackPolicy,
127    #[error("audio processor payload codec error: {0}")]
128    PayloadCodec(String),
129    #[error("audio processor ABI violation: {0}")]
130    AbiViolation(String),
131    #[error("audio processor backpressure: {0}")]
132    Backpressure(String),
133    #[error("audio processor timeout: {0}")]
134    Timeout(String),
135    #[error("audio processor failure: {0}")]
136    Processor(String),
137}
138
139impl AudioProcessorError {
140    pub fn payload_codec(message: impl Into<String>) -> Self {
141        Self::PayloadCodec(message.into())
142    }
143
144    pub fn abi_violation(message: impl Into<String>) -> Self {
145        Self::AbiViolation(message.into())
146    }
147}
148
149/// Safe factory exported by native Rust audio processor plugins.
150pub trait AudioProcessorPluginFactory: Send + Sync {
151    fn name(&self) -> &str;
152
153    fn capabilities(&self) -> AudioProcessorCapabilities;
154
155    fn open_session(
156        &self,
157        config: &AudioProcessorSessionConfig,
158    ) -> Result<Box<dyn AudioProcessorSession>, AudioProcessorError>;
159}
160
161pub trait AudioProcessorSession: Send {
162    fn name(&self) -> &str;
163
164    fn capabilities(&self) -> AudioProcessorCapabilities;
165
166    fn session_info(&self) -> AudioProcessorSessionInfo {
167        let capabilities = self.capabilities();
168        AudioProcessorSessionInfo {
169            processor_name: Some(self.name().to_owned()),
170            selected_backend: None,
171            output_format: capabilities.output_format,
172            max_in_flight_frames: capabilities.max_in_flight_frames,
173        }
174    }
175
176    fn configure(&mut self, _policy: AudioPlaybackPolicy) -> Result<(), AudioProcessorError> {
177        Ok(())
178    }
179
180    /// Processes one PCM frame while preserving the host-owned PTS and discontinuity marker.
181    fn process(&mut self, frame: DecoderPcmFrame) -> Result<DecoderPcmFrame, AudioProcessorError>;
182
183    fn flush(&mut self) -> Result<(), AudioProcessorError>;
184
185    fn close(&mut self) -> Result<(), AudioProcessorError>;
186}
187
188pub struct AudioProcessorChain {
189    capacity: usize,
190    pending: VecDeque<DecoderPcmFrame>,
191    processors: Vec<Box<dyn AudioProcessorSession>>,
192    playback_policy: AudioPlaybackPolicy,
193    closed: bool,
194}
195
196impl AudioProcessorChain {
197    pub fn new(capacity: usize) -> Result<Self, AudioProcessorError> {
198        if capacity == 0 || capacity > MAX_AUDIO_PROCESSOR_QUEUE_FRAMES {
199            return Err(AudioProcessorError::InvalidCapacity);
200        }
201        Ok(Self {
202            capacity,
203            pending: VecDeque::with_capacity(capacity),
204            processors: Vec::new(),
205            playback_policy: AudioPlaybackPolicy::normal(),
206            closed: false,
207        })
208    }
209
210    pub fn with_processors(
211        capacity: usize,
212        processors: Vec<Box<dyn AudioProcessorSession>>,
213    ) -> Result<Self, AudioProcessorError> {
214        let mut chain = Self::new(capacity)?;
215        chain.processors = processors;
216        Ok(chain)
217    }
218
219    pub fn playback_policy(&self) -> AudioPlaybackPolicy {
220        self.playback_policy
221    }
222
223    pub fn set_playback_policy(
224        &mut self,
225        policy: AudioPlaybackPolicy,
226    ) -> Result<(), AudioProcessorError> {
227        if self.closed {
228            return Err(AudioProcessorError::Closed);
229        }
230        policy.validate()?;
231        if self
232            .processors
233            .iter()
234            .any(|processor| !processor.capabilities().supports_playback_policy(policy))
235        {
236            return Err(AudioProcessorError::UnsupportedPlaybackPolicy);
237        }
238        for processor in &mut self.processors {
239            processor.configure(policy)?;
240        }
241        self.playback_policy = policy;
242        Ok(())
243    }
244
245    pub fn submit(
246        &mut self,
247        frame: DecoderPcmFrame,
248    ) -> Result<AudioProcessorSubmitStatus, AudioProcessorError> {
249        if self.closed {
250            return Err(AudioProcessorError::Closed);
251        }
252        frame
253            .validate()
254            .map_err(|error| AudioProcessorError::InvalidPcm(error.to_string()))?;
255        if self.pending.len() >= self.capacity {
256            return Ok(AudioProcessorSubmitStatus::Backpressure);
257        }
258        let mut processed = frame;
259        for processor in &mut self.processors {
260            let input_pts_us = processed.metadata.pts_us;
261            let input_discontinuity = processed.metadata.discontinuity;
262            let output = processor.process(processed)?;
263            output.validate().map_err(|error| {
264                AudioProcessorError::abi_violation(format!(
265                    "processor returned invalid PCM: {error}"
266                ))
267            })?;
268            if output.metadata.pts_us != input_pts_us {
269                return Err(AudioProcessorError::abi_violation(
270                    "processor changed the host-owned PCM presentation timestamp",
271                ));
272            }
273            if output.metadata.discontinuity != input_discontinuity {
274                return Err(AudioProcessorError::abi_violation(
275                    "processor changed the host-owned PCM discontinuity marker",
276                ));
277            }
278            processed = output;
279        }
280        self.pending.push_back(processed);
281        Ok(AudioProcessorSubmitStatus::Accepted)
282    }
283
284    pub fn receive(&mut self) -> Result<Option<DecoderPcmFrame>, AudioProcessorError> {
285        if self.closed {
286            return Err(AudioProcessorError::Closed);
287        }
288        Ok(self.pending.pop_front())
289    }
290
291    pub fn flush(&mut self) -> Result<(), AudioProcessorError> {
292        if self.closed {
293            return Err(AudioProcessorError::Closed);
294        }
295        self.pending.clear();
296        for processor in &mut self.processors {
297            processor.flush()?;
298        }
299        Ok(())
300    }
301
302    pub fn close(&mut self) -> Result<(), AudioProcessorError> {
303        if !self.closed {
304            self.pending.clear();
305            let mut first_error = None;
306            for processor in self.processors.iter_mut().rev() {
307                if let Err(error) = processor.close()
308                    && first_error.is_none()
309                {
310                    first_error = Some(error);
311                }
312            }
313            self.closed = true;
314            if let Some(error) = first_error {
315                return Err(error);
316            }
317        }
318        Ok(())
319    }
320
321    pub fn queue_depth(&self) -> usize {
322        self.pending.len()
323    }
324}
325
326#[cfg(test)]
327mod tests {
328    use super::{
329        AudioPitchMode, AudioPlaybackPolicy, AudioProcessorCapabilities, AudioProcessorChain,
330        AudioProcessorError, AudioProcessorSession, AudioProcessorSubmitStatus,
331    };
332    use crate::{
333        DecoderFrameFormat, DecoderPcmFrame, DecoderPcmFrameMetadata, DecoderPcmSampleLayout,
334    };
335
336    fn frame() -> DecoderPcmFrame {
337        let metadata = DecoderPcmFrameMetadata::audio(
338            "aac",
339            DecoderFrameFormat::F32,
340            48_000,
341            2,
342            DecoderPcmSampleLayout::Interleaved,
343            2,
344        );
345        DecoderPcmFrame {
346            metadata,
347            data: vec![0; 16],
348        }
349    }
350
351    #[test]
352    fn audio_chain_is_bounded_and_flushes_pending_output() {
353        let mut chain = AudioProcessorChain::new(1).expect("bounded chain");
354        assert_eq!(
355            chain.submit(frame()).unwrap(),
356            AudioProcessorSubmitStatus::Accepted
357        );
358        assert_eq!(
359            chain.submit(frame()).unwrap(),
360            AudioProcessorSubmitStatus::Backpressure
361        );
362        chain.flush().expect("flush chain");
363        assert!(chain.receive().unwrap().is_none());
364    }
365
366    struct AddOneProcessor;
367
368    impl AudioProcessorSession for AddOneProcessor {
369        fn name(&self) -> &str {
370            "add-one"
371        }
372
373        fn capabilities(&self) -> AudioProcessorCapabilities {
374            AudioProcessorCapabilities {
375                accepted_formats: vec![DecoderFrameFormat::F32],
376                output_format: Some(DecoderFrameFormat::F32),
377                supports_flush: true,
378                max_in_flight_frames: Some(1),
379                playback_rate_min: Some(0.5),
380                playback_rate_max: Some(2.0),
381                pitch_modes: vec![AudioPitchMode::PreservePitch, AudioPitchMode::FollowRate],
382            }
383        }
384
385        fn process(
386            &mut self,
387            mut frame: DecoderPcmFrame,
388        ) -> Result<DecoderPcmFrame, AudioProcessorError> {
389            frame.data[0] = frame.data[0].saturating_add(1);
390            Ok(frame)
391        }
392
393        fn flush(&mut self) -> Result<(), AudioProcessorError> {
394            Ok(())
395        }
396
397        fn close(&mut self) -> Result<(), AudioProcessorError> {
398            Ok(())
399        }
400    }
401
402    #[test]
403    fn audio_chain_applies_processors_in_linear_order() {
404        let mut chain = AudioProcessorChain::with_processors(
405            2,
406            vec![Box::new(AddOneProcessor), Box::new(AddOneProcessor)],
407        )
408        .expect("processor chain");
409        let mut input = frame();
410        input.data[0] = 0;
411        assert_eq!(
412            chain.submit(input).unwrap(),
413            AudioProcessorSubmitStatus::Accepted
414        );
415        assert_eq!(chain.receive().unwrap().expect("output").data[0], 2);
416        chain.close().expect("close chain");
417    }
418
419    #[test]
420    fn audio_chain_rejects_invalid_or_unsupported_playback_policy() {
421        let mut chain = AudioProcessorChain::with_processors(2, vec![Box::new(AddOneProcessor)])
422            .expect("processor chain");
423        assert!(matches!(
424            chain.set_playback_policy(AudioPlaybackPolicy {
425                playback_rate: 0.0,
426                pitch_mode: AudioPitchMode::FollowRate,
427            }),
428            Err(AudioProcessorError::InvalidPlaybackRate { .. })
429        ));
430        assert_eq!(chain.playback_policy(), AudioPlaybackPolicy::normal());
431
432        let mut chain = AudioProcessorChain::new(2).expect("processor chain");
433        let mut processor = AddOneProcessor;
434        processor
435            .configure(AudioPlaybackPolicy::normal())
436            .expect("default policy");
437        chain
438            .set_playback_policy(AudioPlaybackPolicy {
439                playback_rate: 1.5,
440                pitch_mode: AudioPitchMode::PreservePitch,
441            })
442            .expect("empty chain accepts policy");
443    }
444
445    struct FollowRateOnlyProcessor;
446
447    impl AudioProcessorSession for FollowRateOnlyProcessor {
448        fn name(&self) -> &str {
449            "follow-rate-only"
450        }
451
452        fn capabilities(&self) -> AudioProcessorCapabilities {
453            AudioProcessorCapabilities {
454                accepted_formats: vec![DecoderFrameFormat::F32],
455                output_format: Some(DecoderFrameFormat::F32),
456                supports_flush: true,
457                max_in_flight_frames: Some(1),
458                playback_rate_min: Some(0.5),
459                playback_rate_max: Some(2.0),
460                pitch_modes: vec![AudioPitchMode::FollowRate],
461            }
462        }
463
464        fn process(
465            &mut self,
466            frame: DecoderPcmFrame,
467        ) -> Result<DecoderPcmFrame, AudioProcessorError> {
468            Ok(frame)
469        }
470
471        fn flush(&mut self) -> Result<(), AudioProcessorError> {
472            Ok(())
473        }
474
475        fn close(&mut self) -> Result<(), AudioProcessorError> {
476            Ok(())
477        }
478    }
479
480    #[test]
481    fn audio_chain_rejects_rate_and_pitch_modes_outside_processor_capabilities() {
482        let mut chain =
483            AudioProcessorChain::with_processors(2, vec![Box::new(FollowRateOnlyProcessor)])
484                .expect("processor chain");
485        assert_eq!(
486            chain.set_playback_policy(AudioPlaybackPolicy {
487                playback_rate: 2.5,
488                pitch_mode: AudioPitchMode::FollowRate,
489            }),
490            Err(AudioProcessorError::UnsupportedPlaybackPolicy)
491        );
492        assert_eq!(
493            chain.set_playback_policy(AudioPlaybackPolicy {
494                playback_rate: 1.5,
495                pitch_mode: AudioPitchMode::PreservePitch,
496            }),
497            Err(AudioProcessorError::UnsupportedPlaybackPolicy)
498        );
499    }
500
501    struct TimestampMutatingProcessor {
502        pts_us: Option<i64>,
503    }
504
505    impl AudioProcessorSession for TimestampMutatingProcessor {
506        fn name(&self) -> &str {
507            "timestamp-mutator"
508        }
509
510        fn capabilities(&self) -> AudioProcessorCapabilities {
511            AudioProcessorCapabilities::default()
512        }
513
514        fn process(
515            &mut self,
516            mut frame: DecoderPcmFrame,
517        ) -> Result<DecoderPcmFrame, AudioProcessorError> {
518            frame.metadata.pts_us = self.pts_us;
519            Ok(frame)
520        }
521
522        fn flush(&mut self) -> Result<(), AudioProcessorError> {
523            Ok(())
524        }
525
526        fn close(&mut self) -> Result<(), AudioProcessorError> {
527            Ok(())
528        }
529    }
530
531    #[test]
532    fn audio_chain_rejects_negative_or_mutated_host_owned_timestamps() {
533        for mutated_pts in [Some(-1), Some(1_001)] {
534            let mut input = frame();
535            input.metadata.pts_us = Some(1_000);
536            let mut chain = AudioProcessorChain::with_processors(
537                1,
538                vec![Box::new(TimestampMutatingProcessor {
539                    pts_us: mutated_pts,
540                })],
541            )
542            .expect("processor chain");
543
544            assert!(matches!(
545                chain.submit(input),
546                Err(AudioProcessorError::AbiViolation(message))
547                    if message.contains("presentation timestamp")
548            ));
549            assert_eq!(chain.queue_depth(), 0);
550        }
551    }
552}
player_plugin/audio.rs

player_plugin/
audio.rs
