audio_streams/
capture.rs

1// Copyright 2019 The ChromiumOS Authors
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! ```
6//! use audio_streams::{BoxError, capture::CaptureBuffer, SampleFormat, StreamSource,
7//!     NoopStreamSource};
8//! use std::io::Read;
9//!
10//! const buffer_size: usize = 120;
11//! const num_channels: usize = 2;
12//!
13//! # fn main() -> std::result::Result<(),BoxError> {
14//! let mut stream_source = NoopStreamSource::new();
15//! let sample_format = SampleFormat::S16LE;
16//! let frame_size = num_channels * sample_format.sample_bytes();
17//!
18//! let (_, mut stream) = stream_source
19//!     .new_capture_stream(num_channels, sample_format, 48000, buffer_size, &[])?;
20//! // Capture 10 buffers of zeros.
21//! let mut buf = Vec::new();
22//! buf.resize(buffer_size * frame_size, 0xa5u8);
23//! for _ in 0..10 {
24//!     let mut copy_func = |stream_buffer: &mut CaptureBuffer| {
25//!         assert_eq!(stream_buffer.read(&mut buf)?, buffer_size * frame_size);
26//!         Ok(())
27//!     };
28//!     stream.read_capture_buffer(&mut copy_func)?;
29//! }
30//! # Ok (())
31//! # }
32//! ```
33
34use std::io;
35use std::io::Read;
36use std::io::Write;
37use std::time::Duration;
38use std::time::Instant;
39
40use async_trait::async_trait;
41use remain::sorted;
42use thiserror::Error;
43
44use super::async_api::AudioStreamsExecutor;
45use super::AsyncBufferCommit;
46use super::AudioBuffer;
47use super::BoxError;
48use super::BufferCommit;
49use super::NoopBufferCommit;
50use super::SampleFormat;
51
52/// `CaptureBufferStream` provides `CaptureBuffer`s to read with audio samples from capture.
53pub trait CaptureBufferStream: Send {
54    fn next_capture_buffer<'b, 's: 'b>(&'s mut self) -> Result<CaptureBuffer<'b>, BoxError>;
55
56    /// Call `f` with a `CaptureBuffer`, and trigger the buffer done call back after. `f` can read
57    /// the capture data from the given `CaptureBuffer`.
58    fn read_capture_buffer<'b, 's: 'b>(
59        &'s mut self,
60        f: &mut dyn FnMut(&mut CaptureBuffer<'b>) -> Result<(), BoxError>,
61    ) -> Result<(), BoxError> {
62        let mut buf = self.next_capture_buffer()?;
63        f(&mut buf)?;
64        buf.commit();
65        Ok(())
66    }
67}
68
69impl<S: CaptureBufferStream + ?Sized> CaptureBufferStream for &mut S {
70    fn next_capture_buffer<'b, 's: 'b>(&'s mut self) -> Result<CaptureBuffer<'b>, BoxError> {
71        (**self).next_capture_buffer()
72    }
73}
74
75#[async_trait(?Send)]
76pub trait AsyncCaptureBufferStream: Send {
77    async fn next_capture_buffer<'a>(
78        &'a mut self,
79        _ex: &dyn AudioStreamsExecutor,
80    ) -> Result<AsyncCaptureBuffer<'a>, BoxError>;
81
82    /// Starts the capture stream.
83    ///
84    /// For stream implementations that require explicit lifecycle management
85    /// or authorization (e.g. AAudio on Android), this method is called to
86    /// initiate the stream. It must return promptly and must not wait for
87    /// external events (such as a user permission prompt), because no capture
88    /// periods are produced while it runs. Default implementation is a no-op.
89    fn start(&mut self) -> Result<(), BoxError> {
90        Ok(())
91    }
92
93    /// Stops the capture stream.
94    ///
95    /// Like `start()`, it must return promptly. Default implementation is a
96    /// no-op.
97    fn stop(&mut self) -> Result<(), BoxError> {
98        Ok(())
99    }
100}
101
102#[async_trait(?Send)]
103impl<S: AsyncCaptureBufferStream + ?Sized> AsyncCaptureBufferStream for &mut S {
104    async fn next_capture_buffer<'a>(
105        &'a mut self,
106        ex: &dyn AudioStreamsExecutor,
107    ) -> Result<AsyncCaptureBuffer<'a>, BoxError> {
108        (**self).next_capture_buffer(ex).await
109    }
110
111    fn start(&mut self) -> Result<(), BoxError> {
112        (**self).start()
113    }
114
115    fn stop(&mut self) -> Result<(), BoxError> {
116        (**self).stop()
117    }
118}
119
120/// `CaptureBuffer` contains a block of audio samples got from capture stream. It provides
121/// temporary view to those samples and will notifies capture stream when dropped.
122/// Note that it'll always send `buffer.len() / frame_size` to drop function when it got destroyed
123/// since `CaptureBufferStream` assumes that users get all the samples from the buffer.
124pub struct CaptureBuffer<'a> {
125    buffer: AudioBuffer<'a>,
126    drop: &'a mut dyn BufferCommit,
127}
128
129/// Async version of 'CaptureBuffer`
130pub struct AsyncCaptureBuffer<'a> {
131    buffer: AudioBuffer<'a>,
132    trigger: &'a mut dyn AsyncBufferCommit,
133}
134
135/// Errors that are possible from a `CaptureBuffer`.
136#[sorted]
137#[derive(Error, Debug)]
138pub enum CaptureBufferError {
139    #[error("Invalid buffer length")]
140    InvalidLength,
141}
142
143impl<'a> CaptureBuffer<'a> {
144    /// Creates a new `CaptureBuffer` that holds a reference to the backing memory specified in
145    /// `buffer`.
146    pub fn new<F>(
147        frame_size: usize,
148        buffer: &'a mut [u8],
149        drop: &'a mut F,
150    ) -> Result<Self, CaptureBufferError>
151    where
152        F: BufferCommit,
153    {
154        if buffer.len() % frame_size != 0 {
155            return Err(CaptureBufferError::InvalidLength);
156        }
157
158        Ok(CaptureBuffer {
159            buffer: AudioBuffer {
160                buffer,
161                frame_size,
162                offset: 0,
163            },
164            drop,
165        })
166    }
167
168    /// Returns the number of audio frames that fit in the buffer.
169    pub fn frame_capacity(&self) -> usize {
170        self.buffer.frame_capacity()
171    }
172
173    /// This triggers the callback of `BufferCommit`. This should be called after the data is read
174    /// from the buffer.
175    ///
176    /// Always sends `frame_capacity`.
177    pub fn commit(&mut self) {
178        self.drop.commit(self.frame_capacity());
179    }
180
181    pub fn latency_bytes(&self) -> u32 {
182        self.drop.latency_bytes()
183    }
184
185    /// Reads up to `size` bytes directly from this buffer inside of the given callback function.
186    pub fn copy_cb<F: FnOnce(&[u8])>(&mut self, size: usize, cb: F) -> io::Result<usize> {
187        self.buffer.read_copy_cb(size, cb)
188    }
189}
190
191impl Read for CaptureBuffer<'_> {
192    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
193        self.buffer.read(buf)
194    }
195}
196
197impl<'a> AsyncCaptureBuffer<'a> {
198    /// Creates a new `AsyncCaptureBuffer` that holds a reference to the backing memory specified in
199    /// `buffer`.
200    pub fn new<F>(
201        frame_size: usize,
202        buffer: &'a mut [u8],
203        trigger: &'a mut F,
204    ) -> Result<Self, CaptureBufferError>
205    where
206        F: AsyncBufferCommit,
207    {
208        if buffer.len() % frame_size != 0 {
209            return Err(CaptureBufferError::InvalidLength);
210        }
211
212        Ok(AsyncCaptureBuffer {
213            buffer: AudioBuffer {
214                buffer,
215                frame_size,
216                offset: 0,
217            },
218            trigger,
219        })
220    }
221
222    /// Returns the number of audio frames that fit in the buffer.
223    pub fn frame_capacity(&self) -> usize {
224        self.buffer.frame_capacity()
225    }
226
227    /// This triggers the callback of `AsyncBufferCommit`. This should be called after the data is
228    /// read from the buffer.
229    ///
230    /// Always sends `frame_capacity`.
231    pub async fn commit(&mut self) {
232        self.trigger.commit(self.frame_capacity()).await;
233    }
234
235    pub fn latency_bytes(&self) -> u32 {
236        self.trigger.latency_bytes()
237    }
238
239    /// Reads up to `size` bytes directly from this buffer inside of the given callback function.
240    pub fn copy_cb<F: FnOnce(&[u8])>(&mut self, size: usize, cb: F) -> io::Result<usize> {
241        self.buffer.read_copy_cb(size, cb)
242    }
243
244    /// Copy data to an io::Write
245    pub fn copy_to(&mut self, writer: &mut dyn Write) -> io::Result<usize> {
246        self.buffer.copy_to(writer)
247    }
248}
249
250impl Read for AsyncCaptureBuffer<'_> {
251    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
252        self.buffer.read(buf)
253    }
254}
255
256/// Stream that provides null capture samples.
257pub struct NoopCaptureStream {
258    buffer: Vec<u8>,
259    frame_size: usize,
260    interval: Duration,
261    next_frame: Duration,
262    start_time: Option<Instant>,
263    buffer_drop: NoopBufferCommit,
264}
265
266impl NoopCaptureStream {
267    pub fn new(
268        num_channels: usize,
269        format: SampleFormat,
270        frame_rate: u32,
271        buffer_size: usize,
272    ) -> Self {
273        let frame_size = format.sample_bytes() * num_channels;
274        let interval = Duration::from_nanos(buffer_size as u64 * 1_000_000_000 / frame_rate as u64);
275        NoopCaptureStream {
276            buffer: vec![0; buffer_size * frame_size],
277            frame_size,
278            interval,
279            next_frame: interval,
280            start_time: None,
281            buffer_drop: NoopBufferCommit {
282                which_buffer: false,
283            },
284        }
285    }
286}
287
288impl CaptureBufferStream for NoopCaptureStream {
289    fn next_capture_buffer<'b, 's: 'b>(&'s mut self) -> Result<CaptureBuffer<'b>, BoxError> {
290        if let Some(start_time) = self.start_time {
291            let elapsed = start_time.elapsed();
292            if elapsed < self.next_frame {
293                std::thread::sleep(self.next_frame - elapsed);
294            }
295            self.next_frame += self.interval;
296        } else {
297            self.start_time = Some(Instant::now());
298            self.next_frame = self.interval;
299        }
300        Ok(CaptureBuffer::new(
301            self.frame_size,
302            &mut self.buffer,
303            &mut self.buffer_drop,
304        )?)
305    }
306}
307
308#[async_trait(?Send)]
309impl AsyncCaptureBufferStream for NoopCaptureStream {
310    async fn next_capture_buffer<'a>(
311        &'a mut self,
312        ex: &dyn AudioStreamsExecutor,
313    ) -> Result<AsyncCaptureBuffer<'a>, BoxError> {
314        if let Some(start_time) = self.start_time {
315            let elapsed = start_time.elapsed();
316            if elapsed < self.next_frame {
317                ex.delay(self.next_frame - elapsed).await?;
318            }
319            self.next_frame += self.interval;
320        } else {
321            self.start_time = Some(Instant::now());
322            self.next_frame = self.interval;
323        }
324        Ok(AsyncCaptureBuffer::new(
325            self.frame_size,
326            &mut self.buffer,
327            &mut self.buffer_drop,
328        )?)
329    }
330}
331
332/// Call `f` with a `AsyncCaptureBuffer`, and trigger the buffer done call back after. `f` can read
333/// the capture data from the given `AsyncCaptureBuffer`.
334///
335/// This cannot be a trait method because trait methods with generic parameters are not object safe.
336pub async fn async_read_capture_buffer<F>(
337    stream: &mut dyn AsyncCaptureBufferStream,
338    f: F,
339    ex: &dyn AudioStreamsExecutor,
340) -> Result<(), BoxError>
341where
342    F: FnOnce(&mut AsyncCaptureBuffer) -> Result<(), BoxError>,
343{
344    let mut buf = stream.next_capture_buffer(ex).await?;
345    f(&mut buf)?;
346    buf.commit().await;
347    Ok(())
348}
349
350#[cfg(test)]
351mod tests {
352    use futures::FutureExt;
353
354    use super::super::async_api::test::TestExecutor;
355    use super::super::*;
356    use super::*;
357
358    #[test]
359    fn invalid_buffer_length() {
360        // Capture buffers can't be created with a size that isn't divisible by the frame size.
361        let mut cp_buf = [0xa5u8; 480 * 2 * 2 + 1];
362        let mut buffer_drop = NoopBufferCommit {
363            which_buffer: false,
364        };
365        assert!(CaptureBuffer::new(2, &mut cp_buf, &mut buffer_drop).is_err());
366    }
367
368    #[test]
369    fn commit() {
370        struct TestCommit {
371            frame_count: usize,
372        }
373        impl BufferCommit for TestCommit {
374            fn commit(&mut self, nwritten: usize) {
375                self.frame_count += nwritten;
376            }
377        }
378        let mut test_commit = TestCommit { frame_count: 0 };
379        {
380            const FRAME_SIZE: usize = 4;
381            let mut buf = [0u8; 480 * FRAME_SIZE];
382            let mut cp_buf = CaptureBuffer::new(FRAME_SIZE, &mut buf, &mut test_commit).unwrap();
383            let mut local_buf = [0u8; 240 * FRAME_SIZE];
384            assert_eq!(cp_buf.read(&mut local_buf).unwrap(), 240 * FRAME_SIZE);
385            cp_buf.commit();
386        }
387        // This should be 480 no matter how many samples are read.
388        assert_eq!(test_commit.frame_count, 480);
389    }
390
391    #[test]
392    fn sixteen_bit_stereo() {
393        let mut server = NoopStreamSource::new();
394        let (_, mut stream) = server
395            .new_capture_stream(2, SampleFormat::S16LE, 48000, 480, &[])
396            .unwrap();
397        let mut copy_func = |b: &mut CaptureBuffer| {
398            assert_eq!(b.buffer.frame_capacity(), 480);
399            let mut pb_buf = [0xa5u8; 480 * 2 * 2];
400            assert_eq!(b.read(&mut pb_buf).unwrap(), 480 * 2 * 2);
401            Ok(())
402        };
403        stream.read_capture_buffer(&mut copy_func).unwrap();
404    }
405
406    #[test]
407    fn consumption_rate() {
408        let mut server = NoopStreamSource::new();
409        let (_, mut stream) = server
410            .new_capture_stream(2, SampleFormat::S16LE, 48000, 480, &[])
411            .unwrap();
412        let start = Instant::now();
413        {
414            let mut copy_func = |b: &mut CaptureBuffer| {
415                let mut cp_buf = [0xa5u8; 480 * 2 * 2];
416                assert_eq!(b.read(&mut cp_buf).unwrap(), 480 * 2 * 2);
417                for buf in cp_buf.iter() {
418                    assert_eq!(*buf, 0, "Read samples should all be zeros.");
419                }
420                Ok(())
421            };
422            stream.read_capture_buffer(&mut copy_func).unwrap();
423        }
424        // The second call should block until the first buffer is consumed.
425        let mut assert_func = |_: &mut CaptureBuffer| {
426            let elapsed = start.elapsed();
427            assert!(
428                elapsed > Duration::from_millis(10),
429                "next_capture_buffer didn't block long enough {}",
430                elapsed.subsec_millis()
431            );
432            Ok(())
433        };
434        stream.read_capture_buffer(&mut assert_func).unwrap();
435    }
436
437    #[test]
438    fn async_commit() {
439        struct TestCommit {
440            frame_count: usize,
441        }
442        #[async_trait(?Send)]
443        impl AsyncBufferCommit for TestCommit {
444            async fn commit(&mut self, nwritten: usize) {
445                self.frame_count += nwritten;
446            }
447        }
448        async fn this_test() {
449            let mut test_commit = TestCommit { frame_count: 0 };
450            {
451                const FRAME_SIZE: usize = 4;
452                let mut buf = [0u8; 480 * FRAME_SIZE];
453                let mut cp_buf =
454                    AsyncCaptureBuffer::new(FRAME_SIZE, &mut buf, &mut test_commit).unwrap();
455                let mut local_buf = [0u8; 240 * FRAME_SIZE];
456                assert_eq!(cp_buf.read(&mut local_buf).unwrap(), 240 * FRAME_SIZE);
457                cp_buf.commit().await;
458            }
459            // This should be 480 no matter how many samples are read.
460            assert_eq!(test_commit.frame_count, 480);
461        }
462
463        this_test().now_or_never();
464    }
465
466    #[test]
467    fn consumption_rate_async() {
468        async fn this_test(ex: &TestExecutor) {
469            let mut server = NoopStreamSource::new();
470            let (_, mut stream) = server
471                .new_async_capture_stream(2, SampleFormat::S16LE, 48000, 480, &[], ex)
472                .unwrap();
473            let start = Instant::now();
474            {
475                let copy_func = |buf: &mut AsyncCaptureBuffer| {
476                    let mut cp_buf = [0xa5u8; 480 * 2 * 2];
477                    assert_eq!(buf.read(&mut cp_buf).unwrap(), 480 * 2 * 2);
478                    for buf in cp_buf.iter() {
479                        assert_eq!(*buf, 0, "Read samples should all be zeros.");
480                    }
481                    Ok(())
482                };
483                async_read_capture_buffer(&mut *stream, copy_func, ex)
484                    .await
485                    .unwrap();
486            }
487            // The second call should block until the first buffer is consumed.
488            let assert_func = |_: &mut AsyncCaptureBuffer| {
489                let elapsed = start.elapsed();
490                assert!(
491                    elapsed > Duration::from_millis(10),
492                    "write_playback_buffer didn't block long enough {}",
493                    elapsed.subsec_millis()
494                );
495                Ok(())
496            };
497            async_read_capture_buffer(&mut *stream, assert_func, ex)
498                .await
499                .unwrap();
500        }
501
502        let ex = TestExecutor {};
503        this_test(&ex).now_or_never();
504    }
505
506    #[test]
507    fn async_capture_stream_default_start_stop() {
508        let ex = TestExecutor {};
509        let mut server = NoopStreamSource::new();
510        let (_, mut stream) = server
511            .new_async_capture_stream(2, SampleFormat::S16LE, 48000, 480, &[], &ex)
512            .unwrap();
513        assert!(stream.start().is_ok());
514        assert!(stream.stop().is_ok());
515
516        let trait_obj: &mut dyn AsyncCaptureBufferStream = &mut *stream;
517        assert!(trait_obj.start().is_ok());
518        assert!(trait_obj.stop().is_ok());
519    }
520}
audio_streams/capture.rs

audio_streams/
capture.rs
