Rodio audio (#37786)

Adds input to the experimental rodio_audio pipeline.

Enable with:
```json
"audio": {
  "experimental.rodio_audio": true
}
```

Additionally enables automatic volume 
control for incoming audio:
```json
"audio": {
  "experimental.control_output_volume": true
}
```

Release Notes:

- N/A

crates/audio/src/audio.rs

@@ -1,19 +1,54 @@
-use anyhow::{Context as _, Result, anyhow};
+use anyhow::{Context as _, Result};
 use collections::HashMap;
-use gpui::{App, BorrowAppContext, Global};
-use rodio::{Decoder, OutputStream, OutputStreamBuilder, Source, source::Buffered};
+use gpui::{App, AsyncApp, BackgroundExecutor, BorrowAppContext, Global};
+use libwebrtc::native::apm;
+use log::info;
+use parking_lot::Mutex;
+use rodio::{
+    Decoder, OutputStream, OutputStreamBuilder, Source,
+    cpal::Sample,
+    mixer::Mixer,
+    nz,
+    source::{Buffered, LimitSettings, UniformSourceIterator},
+};
 use settings::Settings;
-use std::io::Cursor;
+use std::{
+    io::Cursor,
+    num::NonZero,
+    path::PathBuf,
+    sync::{Arc, atomic::Ordering},
+    time::Duration,
+};
 use util::ResultExt;
 
 mod audio_settings;
+mod replays;
+mod rodio_ext;
 pub use audio_settings::AudioSettings;
+pub use rodio_ext::RodioExt;
+
+use crate::audio_settings::LIVE_SETTINGS;
+
+// NOTE: We used to use WebRTC's mixer which only supported
+// 16kHz, 32kHz and 48kHz. As 48 is the most common "next step up"
+// for audio output devices like speakers/bluetooth, we just hard-code
+// this; and downsample when we need to.
+//
+// Since most noise cancelling requires 16kHz we will move to
+// that in the future.
+pub const SAMPLE_RATE: NonZero<u32> = nz!(48000);
+pub const CHANNEL_COUNT: NonZero<u16> = nz!(2);
+pub const BUFFER_SIZE: usize = // echo canceller and livekit want 10ms of audio
+    (SAMPLE_RATE.get() as usize / 100) * CHANNEL_COUNT.get() as usize;
+
+pub const REPLAY_DURATION: Duration = Duration::from_secs(30);
 
 pub fn init(cx: &mut App) {
     AudioSettings::register(cx);
+    LIVE_SETTINGS.initialize(cx);
 }
 
-#[derive(Copy, Clone, Eq, Hash, PartialEq)]
+#[derive(Debug, Copy, Clone, Eq, Hash, PartialEq)]
 pub enum Sound {
     Joined,
     Leave,
@@ -38,32 +73,138 @@ impl Sound {
     }
 }
 
-#[derive(Default)]
 pub struct Audio {
     output_handle: Option<OutputStream>,
+    output_mixer: Option<Mixer>,
+    pub echo_canceller: Arc<Mutex<apm::AudioProcessingModule>>,
     source_cache: HashMap<Sound, Buffered<Decoder<Cursor<Vec<u8>>>>>,
+    replays: replays::Replays,
+}
+
+impl Default for Audio {
+    fn default() -> Self {
+        Self {
+            output_handle: Default::default(),
+            output_mixer: Default::default(),
+            echo_canceller: Arc::new(Mutex::new(apm::AudioProcessingModule::new(
+                true, false, false, false,
+            ))),
+            source_cache: Default::default(),
+            replays: Default::default(),
+        }
+    }
 }
 
 impl Global for Audio {}
 
 impl Audio {
-    fn ensure_output_exists(&mut self) -> Option<&OutputStream> {
+    fn ensure_output_exists(&mut self) -> Result<&Mixer> {
         if self.output_handle.is_none() {
-            self.output_handle = OutputStreamBuilder::open_default_stream().log_err();
+            self.output_handle = Some(
+                OutputStreamBuilder::open_default_stream()
+                    .context("Could not open default output stream")?,
+            );
+            if let Some(output_handle) = &self.output_handle {
+                let (mixer, source) = rodio::mixer::mixer(CHANNEL_COUNT, SAMPLE_RATE);
+                // or the mixer will end immediately as its empty.
+                mixer.add(rodio::source::Zero::new(CHANNEL_COUNT, SAMPLE_RATE));
+                self.output_mixer = Some(mixer);
+
+                let echo_canceller = Arc::clone(&self.echo_canceller);
+                let source = source.inspect_buffer::<BUFFER_SIZE, _>(move |buffer| {
+                    let mut buf: [i16; _] = buffer.map(|s| s.to_sample());
+                    echo_canceller
+                        .lock()
+                        .process_reverse_stream(
+                            &mut buf,
+                            SAMPLE_RATE.get() as i32,
+                            CHANNEL_COUNT.get().into(),
+                        )
+                        .expect("Audio input and output threads should not panic");
+                });
+                output_handle.mixer().add(source);
+            }
         }
 
-        self.output_handle.as_ref()
+        Ok(self
+            .output_mixer
+            .as_ref()
+            .expect("we only get here if opening the outputstream succeeded"))
     }
 
-    pub fn play_source(
+    pub fn save_replays(
+        &self,
+        executor: BackgroundExecutor,
+    ) -> gpui::Task<anyhow::Result<(PathBuf, Duration)>> {
+        self.replays.replays_to_tar(executor)
+    }
+
+    pub fn open_microphone(voip_parts: VoipParts) -> anyhow::Result<impl Source> {
+        let stream = rodio::microphone::MicrophoneBuilder::new()
+            .default_device()?
+            .default_config()?
+            .prefer_sample_rates([SAMPLE_RATE, SAMPLE_RATE.saturating_mul(nz!(2))])
+            .prefer_channel_counts([nz!(1), nz!(2)])
+            .prefer_buffer_sizes(512..)
+            .open_stream()?;
+        info!("Opened microphone: {:?}", stream.config());
+
+        let (replay, stream) = UniformSourceIterator::new(stream, CHANNEL_COUNT, SAMPLE_RATE)
+            .limit(LimitSettings::live_performance())
+            .process_buffer::<BUFFER_SIZE, _>(move |buffer| {
+                let mut int_buffer: [i16; _] = buffer.map(|s| s.to_sample());
+                if voip_parts
+                    .echo_canceller
+                    .lock()
+                    .process_stream(
+                        &mut int_buffer,
+                        SAMPLE_RATE.get() as i32,
+                        CHANNEL_COUNT.get() as i32,
+                    )
+                    .context("livekit audio processor error")
+                    .log_err()
+                    .is_some()
+                {
+                    for (sample, processed) in buffer.iter_mut().zip(&int_buffer) {
+                        *sample = (*processed).to_sample();
+                    }
+                }
+            })
+            .automatic_gain_control(1.0, 4.0, 0.0, 5.0)
+            .periodic_access(Duration::from_millis(100), move |agc_source| {
+                agc_source.set_enabled(LIVE_SETTINGS.control_input_volume.load(Ordering::Relaxed));
+            })
+            .replayable(REPLAY_DURATION)
+            .expect("REPLAY_DURATION is longer then 100ms");
+
+        voip_parts
+            .replays
+            .add_voip_stream("local microphone".to_string(), replay);
+        Ok(stream)
+    }
+
+    pub fn play_voip_stream(
         source: impl rodio::Source + Send + 'static,
+        speaker_name: String,
+        is_staff: bool,
         cx: &mut App,
     ) -> anyhow::Result<()> {
+        let (replay_source, source) = source
+            .automatic_gain_control(1.0, 4.0, 0.0, 5.0)
+            .periodic_access(Duration::from_millis(100), move |agc_source| {
+                agc_source.set_enabled(LIVE_SETTINGS.control_input_volume.load(Ordering::Relaxed));
+            })
+            .replayable(REPLAY_DURATION)
+            .expect("REPLAY_DURATION is longer then 100ms");
+
         cx.update_default_global(|this: &mut Self, _cx| {
-            let output_handle = this
+            let output_mixer = this
                 .ensure_output_exists()
-                .ok_or_else(|| anyhow!("Could not open audio output"))?;
-            output_handle.mixer().add(source);
+                .context("Could not get output mixer")?;
+            output_mixer.add(source);
+            if is_staff {
+                this.replays.add_voip_stream(speaker_name, replay_source);
+            }
             Ok(())
         })
     }
@@ -71,8 +212,12 @@ impl Audio {
     pub fn play_sound(sound: Sound, cx: &mut App) {
         cx.update_default_global(|this: &mut Self, cx| {
             let source = this.sound_source(sound, cx).log_err()?;
-            let output_handle = this.ensure_output_exists()?;
-            output_handle.mixer().add(source);
+            let output_mixer = this
+                .ensure_output_exists()
+                .context("Could not get output mixer")
+                .log_err()?;
+
+            output_mixer.add(source);
             Some(())
         });
     }
@@ -103,3 +248,21 @@ impl Audio {
         Ok(source)
     }
 }
+
+pub struct VoipParts {
+    echo_canceller: Arc<Mutex<apm::AudioProcessingModule>>,
+    replays: replays::Replays,
+}
+
+impl VoipParts {
+    pub fn new(cx: &AsyncApp) -> anyhow::Result<Self> {
+        let (apm, replays) = cx.try_read_default_global::<Audio, _>(|audio, _| {
+            (Arc::clone(&audio.echo_canceller), audio.replays.clone())
+        })?;
+
+        Ok(Self {
+            echo_canceller: apm,
+            replays,
+        })
+    }
+}

crates/audio/src/audio_settings.rs

@@ -1,14 +1,29 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+
 use anyhow::Result;
 use gpui::App;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
-use settings::{Settings, SettingsKey, SettingsSources, SettingsUi};
+use settings::{Settings, SettingsKey, SettingsSources, SettingsStore, SettingsUi};
 
 #[derive(Clone, Default, Serialize, Deserialize, JsonSchema, Debug, SettingsUi)]
 pub struct AudioSettings {
     /// Opt into the new audio system.
     #[serde(rename = "experimental.rodio_audio", default)]
     pub rodio_audio: bool, // default is false
+    /// Requires 'rodio_audio: true'
+    ///
+    /// Use the new audio systems automatic gain control for your microphone.
+    /// This affects how loud you sound to others.
+    #[serde(rename = "experimental.control_input_volume", default)]
+    pub control_input_volume: bool,
+    /// Requires 'rodio_audio: true'
+    ///
+    /// Use the new audio systems automatic gain control on everyone in the
+    /// call. This makes call members who are too quite louder and those who are
+    /// too loud quieter. This only affects how things sound for you.
+    #[serde(rename = "experimental.control_output_volume", default)]
+    pub control_output_volume: bool,
 }
 
 /// Configuration of audio in Zed.
@@ -16,9 +31,22 @@ pub struct AudioSettings {
 #[serde(default)]
 #[settings_key(key = "audio")]
 pub struct AudioSettingsContent {
-    /// Whether to use the experimental audio system
+    /// Opt into the new audio system.
     #[serde(rename = "experimental.rodio_audio", default)]
-    pub rodio_audio: bool,
+    pub rodio_audio: bool, // default is false
+    /// Requires 'rodio_audio: true'
+    ///
+    /// Use the new audio systems automatic gain control for your microphone.
+    /// This affects how loud you sound to others.
+    #[serde(rename = "experimental.control_input_volume", default)]
+    pub control_input_volume: bool,
+    /// Requires 'rodio_audio: true'
+    ///
+    /// Use the new audio systems automatic gain control on everyone in the
+    /// call. This makes call members who are too quite louder and those who are
+    /// too loud quieter. This only affects how things sound for you.
+    #[serde(rename = "experimental.control_output_volume", default)]
+    pub control_output_volume: bool,
 }
 
 impl Settings for AudioSettings {
@@ -30,3 +58,39 @@ impl Settings for AudioSettings {
 
     fn import_from_vscode(_vscode: &settings::VsCodeSettings, _current: &mut Self::FileContent) {}
 }
+
+pub(crate) struct LiveSettings {
+    pub(crate) control_input_volume: AtomicBool,
+    pub(crate) control_output_volume: AtomicBool,
+}
+
+impl LiveSettings {
+    pub(crate) fn initialize(&self, cx: &mut App) {
+        cx.observe_global::<SettingsStore>(move |cx| {
+            LIVE_SETTINGS.control_input_volume.store(
+                AudioSettings::get_global(cx).control_input_volume,
+                Ordering::Relaxed,
+            );
+            LIVE_SETTINGS.control_output_volume.store(
+                AudioSettings::get_global(cx).control_output_volume,
+                Ordering::Relaxed,
+            );
+        })
+        .detach();
+
+        let init_settings = AudioSettings::get_global(cx);
+        LIVE_SETTINGS
+            .control_input_volume
+            .store(init_settings.control_input_volume, Ordering::Relaxed);
+        LIVE_SETTINGS
+            .control_output_volume
+            .store(init_settings.control_output_volume, Ordering::Relaxed);
+    }
+}
+
+/// Allows access to settings from the audio thread. Updated by
+/// observer of SettingsStore.
+pub(crate) static LIVE_SETTINGS: LiveSettings = LiveSettings {
+    control_input_volume: AtomicBool::new(true),
+    control_output_volume: AtomicBool::new(true),
+};

crates/audio/src/replays.rs

@@ -0,0 +1,77 @@
+use anyhow::{Context, anyhow};
+use async_tar::{Builder, Header};
+use gpui::{BackgroundExecutor, Task};
+
+use collections::HashMap;
+use parking_lot::Mutex;
+use rodio::Source;
+use smol::fs::File;
+use std::{io, path::PathBuf, sync::Arc, time::Duration};
+
+use crate::{REPLAY_DURATION, rodio_ext::Replay};
+
+#[derive(Default, Clone)]
+pub(crate) struct Replays(Arc<Mutex<HashMap<String, Replay>>>);
+
+impl Replays {
+    pub(crate) fn add_voip_stream(&self, stream_name: String, source: Replay) {
+        let mut map = self.0.lock();
+        map.retain(|_, replay| replay.source_is_active());
+        map.insert(stream_name, source);
+    }
+
+    pub(crate) fn replays_to_tar(
+        &self,
+        executor: BackgroundExecutor,
+    ) -> Task<anyhow::Result<(PathBuf, Duration)>> {
+        let map = Arc::clone(&self.0);
+        executor.spawn(async move {
+            let recordings: Vec<_> = map
+                .lock()
+                .iter_mut()
+                .map(|(name, replay)| {
+                    let queued = REPLAY_DURATION.min(replay.duration_ready());
+                    (name.clone(), replay.take_duration(queued).record())
+                })
+                .collect();
+            let longest = recordings
+                .iter()
+                .map(|(_, r)| {
+                    r.total_duration()
+                        .expect("SamplesBuffer always returns a total duration")
+                })
+                .max()
+                .ok_or(anyhow!("There is no audio to capture"))?;
+
+            let path = std::env::current_dir()
+                .context("Could not get current dir")?
+                .join("replays.tar");
+            let tar = File::create(&path)
+                .await
+                .context("Could not create file for tar")?;
+
+            let mut tar = Builder::new(tar);
+
+            for (name, recording) in recordings {
+                let mut writer = io::Cursor::new(Vec::new());
+                rodio::wav_to_writer(recording, &mut writer).context("failed to encode wav")?;
+                let wav_data = writer.into_inner();
+                let path = name.replace(' ', "_") + ".wav";
+                let mut header = Header::new_gnu();
+                // rw permissions for everyone
+                header.set_mode(0o666);
+                header.set_size(wav_data.len() as u64);
+                tar.append_data(&mut header, path, wav_data.as_slice())
+                    .await
+                    .context("failed to apped wav to tar")?;
+            }
+            tar.into_inner()
+                .await
+                .context("Could not finish writing tar")?
+                .sync_all()
+                .await
+                .context("Could not flush tar file to disk")?;
+            Ok((path, longest))
+        })
+    }
+}

crates/audio/src/rodio_ext.rs

@@ -0,0 +1,593 @@
+use std::{
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Duration,
+};
+
+use crossbeam::queue::ArrayQueue;
+use rodio::{ChannelCount, Sample, SampleRate, Source};
+
+#[derive(Debug)]
+pub struct ReplayDurationTooShort;
+
+pub trait RodioExt: Source + Sized {
+    fn process_buffer<const N: usize, F>(self, callback: F) -> ProcessBuffer<N, Self, F>
+    where
+        F: FnMut(&mut [Sample; N]);
+    fn inspect_buffer<const N: usize, F>(self, callback: F) -> InspectBuffer<N, Self, F>
+    where
+        F: FnMut(&[Sample; N]);
+    fn replayable(
+        self,
+        duration: Duration,
+    ) -> Result<(Replay, Replayable<Self>), ReplayDurationTooShort>;
+    fn take_samples(self, n: usize) -> TakeSamples<Self>;
+}
+
+impl<S: Source> RodioExt for S {
+    fn process_buffer<const N: usize, F>(self, callback: F) -> ProcessBuffer<N, Self, F>
+    where
+        F: FnMut(&mut [Sample; N]),
+    {
+        ProcessBuffer {
+            inner: self,
+            callback,
+            buffer: [0.0; N],
+            next: N,
+        }
+    }
+    fn inspect_buffer<const N: usize, F>(self, callback: F) -> InspectBuffer<N, Self, F>
+    where
+        F: FnMut(&[Sample; N]),
+    {
+        InspectBuffer {
+            inner: self,
+            callback,
+            buffer: [0.0; N],
+            free: 0,
+        }
+    }
+    /// Maintains a live replay with a history of at least `duration` seconds.
+    ///
+    /// Note:
+    /// History can be 100ms longer if the source drops before or while the
+    /// replay is being read
+    ///
+    /// # Errors
+    /// If duration is smaller then 100ms
+    fn replayable(
+        self,
+        duration: Duration,
+    ) -> Result<(Replay, Replayable<Self>), ReplayDurationTooShort> {
+        if duration < Duration::from_millis(100) {
+            return Err(ReplayDurationTooShort);
+        }
+
+        let samples_per_second = self.sample_rate().get() as usize * self.channels().get() as usize;
+        let samples_to_queue = duration.as_secs_f64() * samples_per_second as f64;
+        let samples_to_queue =
+            (samples_to_queue as usize).next_multiple_of(self.channels().get().into());
+
+        let chunk_size =
+            (samples_per_second.div_ceil(10)).next_multiple_of(self.channels().get() as usize);
+        let chunks_to_queue = samples_to_queue.div_ceil(chunk_size);
+
+        let is_active = Arc::new(AtomicBool::new(true));
+        let queue = Arc::new(ReplayQueue::new(chunks_to_queue, chunk_size));
+        Ok((
+            Replay {
+                rx: Arc::clone(&queue),
+                buffer: Vec::new().into_iter(),
+                sleep_duration: duration / 2,
+                sample_rate: self.sample_rate(),
+                channel_count: self.channels(),
+                source_is_active: is_active.clone(),
+            },
+            Replayable {
+                tx: queue,
+                inner: self,
+                buffer: Vec::with_capacity(chunk_size),
+                chunk_size,
+                is_active,
+            },
+        ))
+    }
+    fn take_samples(self, n: usize) -> TakeSamples<S> {
+        TakeSamples {
+            inner: self,
+            left_to_take: n,
+        }
+    }
+}
+
+pub struct TakeSamples<S> {
+    inner: S,
+    left_to_take: usize,
+}
+
+impl<S: Source> Iterator for TakeSamples<S> {
+    type Item = Sample;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.left_to_take == 0 {
+            None
+        } else {
+            self.left_to_take -= 1;
+            self.inner.next()
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (0, Some(self.left_to_take))
+    }
+}
+
+impl<S: Source> Source for TakeSamples<S> {
+    fn current_span_len(&self) -> Option<usize> {
+        None // does not support spans
+    }
+
+    fn channels(&self) -> ChannelCount {
+        self.inner.channels()
+    }
+
+    fn sample_rate(&self) -> SampleRate {
+        self.inner.sample_rate()
+    }
+
+    fn total_duration(&self) -> Option<Duration> {
+        Some(Duration::from_secs_f64(
+            self.left_to_take as f64
+                / self.sample_rate().get() as f64
+                / self.channels().get() as f64,
+        ))
+    }
+}
+
+#[derive(Debug)]
+struct ReplayQueue {
+    inner: ArrayQueue<Vec<Sample>>,
+    normal_chunk_len: usize,
+    /// The last chunk in the queue may be smaller then
+    /// the normal chunk size. This is always equal to the
+    /// size of the last element in the queue.
+    /// (so normally chunk_size)
+    last_chunk: Mutex<Vec<Sample>>,
+}
+
+impl ReplayQueue {
+    fn new(queue_len: usize, chunk_size: usize) -> Self {
+        Self {
+            inner: ArrayQueue::new(queue_len),
+            normal_chunk_len: chunk_size,
+            last_chunk: Mutex::new(Vec::new()),
+        }
+    }
+    /// Returns the length in samples
+    fn len(&self) -> usize {
+        self.inner.len().saturating_sub(1) * self.normal_chunk_len
+            + self
+                .last_chunk
+                .lock()
+                .expect("Self::push_last can not poison this lock")
+                .len()
+    }
+
+    fn pop(&self) -> Option<Vec<Sample>> {
+        self.inner.pop() // removes element that was inserted first
+    }
+
+    fn push_last(&self, mut samples: Vec<Sample>) {
+        let mut last_chunk = self
+            .last_chunk
+            .lock()
+            .expect("Self::len can not poison this lock");
+        std::mem::swap(&mut *last_chunk, &mut samples);
+    }
+
+    fn push_normal(&self, samples: Vec<Sample>) {
+        let _pushed_out_of_ringbuf = self.inner.force_push(samples);
+    }
+}
+
+pub struct ProcessBuffer<const N: usize, S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&mut [Sample; N]),
+{
+    inner: S,
+    callback: F,
+    /// Buffer used for both input and output.
+    buffer: [Sample; N],
+    /// Next already processed sample is at this index
+    /// in buffer.
+    ///
+    /// If this is equal to the length of the buffer we have no more samples and
+    /// we must get new ones and process them
+    next: usize,
+}
+
+impl<const N: usize, S, F> Iterator for ProcessBuffer<N, S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&mut [Sample; N]),
+{
+    type Item = Sample;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.next += 1;
+        if self.next < self.buffer.len() {
+            let sample = self.buffer[self.next];
+            return Some(sample);
+        }
+
+        for sample in &mut self.buffer {
+            *sample = self.inner.next()?
+        }
+        (self.callback)(&mut self.buffer);
+
+        self.next = 0;
+        Some(self.buffer[0])
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
+    }
+}
+
+impl<const N: usize, S, F> Source for ProcessBuffer<N, S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&mut [Sample; N]),
+{
+    fn current_span_len(&self) -> Option<usize> {
+        None
+    }
+
+    fn channels(&self) -> rodio::ChannelCount {
+        self.inner.channels()
+    }
+
+    fn sample_rate(&self) -> rodio::SampleRate {
+        self.inner.sample_rate()
+    }
+
+    fn total_duration(&self) -> Option<std::time::Duration> {
+        self.inner.total_duration()
+    }
+}
+
+pub struct InspectBuffer<const N: usize, S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&[Sample; N]),
+{
+    inner: S,
+    callback: F,
+    /// Stores already emitted samples, once its full we call the callback.
+    buffer: [Sample; N],
+    /// Next free element in buffer. If this is equal to the buffer length
+    /// we have no more free lements.
+    free: usize,
+}
+
+impl<const N: usize, S, F> Iterator for InspectBuffer<N, S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&[Sample; N]),
+{
+    type Item = Sample;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let Some(sample) = self.inner.next() else {
+            return None;
+        };
+
+        self.buffer[self.free] = sample;
+        self.free += 1;
+
+        if self.free == self.buffer.len() {
+            (self.callback)(&self.buffer);
+            self.free = 0
+        }
+
+        Some(sample)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
+    }
+}
+
+impl<const N: usize, S, F> Source for InspectBuffer<N, S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&[Sample; N]),
+{
+    fn current_span_len(&self) -> Option<usize> {
+        None
+    }
+
+    fn channels(&self) -> rodio::ChannelCount {
+        self.inner.channels()
+    }
+
+    fn sample_rate(&self) -> rodio::SampleRate {
+        self.inner.sample_rate()
+    }
+
+    fn total_duration(&self) -> Option<std::time::Duration> {
+        self.inner.total_duration()
+    }
+}
+
+#[derive(Debug)]
+pub struct Replayable<S: Source> {
+    inner: S,
+    buffer: Vec<Sample>,
+    chunk_size: usize,
+    tx: Arc<ReplayQueue>,
+    is_active: Arc<AtomicBool>,
+}
+
+impl<S: Source> Iterator for Replayable<S> {
+    type Item = Sample;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if let Some(sample) = self.inner.next() {
+            self.buffer.push(sample);
+            if self.buffer.len() == self.chunk_size {
+                self.tx.push_normal(std::mem::take(&mut self.buffer));
+            }
+            Some(sample)
+        } else {
+            let last_chunk = std::mem::take(&mut self.buffer);
+            self.tx.push_last(last_chunk);
+            self.is_active.store(false, Ordering::Relaxed);
+            None
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.inner.size_hint()
+    }
+}
+
+impl<S: Source> Source for Replayable<S> {
+    fn current_span_len(&self) -> Option<usize> {
+        self.inner.current_span_len()
+    }
+
+    fn channels(&self) -> ChannelCount {
+        self.inner.channels()
+    }
+
+    fn sample_rate(&self) -> SampleRate {
+        self.inner.sample_rate()
+    }
+
+    fn total_duration(&self) -> Option<Duration> {
+        self.inner.total_duration()
+    }
+}
+
+#[derive(Debug)]
+pub struct Replay {
+    rx: Arc<ReplayQueue>,
+    buffer: std::vec::IntoIter<Sample>,
+    sleep_duration: Duration,
+    sample_rate: SampleRate,
+    channel_count: ChannelCount,
+    source_is_active: Arc<AtomicBool>,
+}
+
+impl Replay {
+    pub fn source_is_active(&self) -> bool {
+        // - source could return None and not drop
+        // - source could be dropped before returning None
+        self.source_is_active.load(Ordering::Relaxed) && Arc::strong_count(&self.rx) < 2
+    }
+
+    /// Duration of what is in the buffer and can be returned without blocking.
+    pub fn duration_ready(&self) -> Duration {
+        let samples_per_second = self.channels().get() as u32 * self.sample_rate().get();
+
+        let seconds_queued = self.samples_ready() as f64 / samples_per_second as f64;
+        Duration::from_secs_f64(seconds_queued)
+    }
+
+    /// Number of samples in the buffer and can be returned without blocking.
+    pub fn samples_ready(&self) -> usize {
+        self.rx.len() + self.buffer.len()
+    }
+}
+
+impl Iterator for Replay {
+    type Item = Sample;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if let Some(sample) = self.buffer.next() {
+            return Some(sample);
+        }
+
+        loop {
+            if let Some(new_buffer) = self.rx.pop() {
+                self.buffer = new_buffer.into_iter();
+                return self.buffer.next();
+            }
+
+            if !self.source_is_active() {
+                return None;
+            }
+
+            std::thread::sleep(self.sleep_duration);
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        ((self.rx.len() + self.buffer.len()), None)
+    }
+}
+
+impl Source for Replay {
+    fn current_span_len(&self) -> Option<usize> {
+        None // source is not compatible with spans
+    }
+
+    fn channels(&self) -> ChannelCount {
+        self.channel_count
+    }
+
+    fn sample_rate(&self) -> SampleRate {
+        self.sample_rate
+    }
+
+    fn total_duration(&self) -> Option<Duration> {
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rodio::{nz, static_buffer::StaticSamplesBuffer};
+
+    use super::*;
+
+    const SAMPLES: [Sample; 5] = [0.0, 1.0, 2.0, 3.0, 4.0];
+
+    fn test_source() -> StaticSamplesBuffer {
+        StaticSamplesBuffer::new(nz!(1), nz!(1), &SAMPLES)
+    }
+
+    mod process_buffer {
+        use super::*;
+
+        #[test]
+        fn callback_gets_all_samples() {
+            let input = test_source();
+
+            let _ = input
+                .process_buffer::<{ SAMPLES.len() }, _>(|buffer| assert_eq!(*buffer, SAMPLES))
+                .count();
+        }
+        #[test]
+        fn callback_modifies_yielded() {
+            let input = test_source();
+
+            let yielded: Vec<_> = input
+                .process_buffer::<{ SAMPLES.len() }, _>(|buffer| {
+                    for sample in buffer {
+                        *sample += 1.0;
+                    }
+                })
+                .collect();
+            assert_eq!(
+                yielded,
+                SAMPLES.into_iter().map(|s| s + 1.0).collect::<Vec<_>>()
+            )
+        }
+        #[test]
+        fn source_truncates_to_whole_buffers() {
+            let input = test_source();
+
+            let yielded = input
+                .process_buffer::<3, _>(|buffer| assert_eq!(buffer, &SAMPLES[..3]))
+                .count();
+            assert_eq!(yielded, 3)
+        }
+    }
+
+    mod inspect_buffer {
+        use super::*;
+
+        #[test]
+        fn callback_gets_all_samples() {
+            let input = test_source();
+
+            let _ = input
+                .inspect_buffer::<{ SAMPLES.len() }, _>(|buffer| assert_eq!(*buffer, SAMPLES))
+                .count();
+        }
+        #[test]
+        fn source_does_not_truncate() {
+            let input = test_source();
+
+            let yielded = input
+                .inspect_buffer::<3, _>(|buffer| assert_eq!(buffer, &SAMPLES[..3]))
+                .count();
+            assert_eq!(yielded, SAMPLES.len())
+        }
+    }
+
+    mod instant_replay {
+        use super::*;
+
+        #[test]
+        fn continues_after_history() {
+            let input = test_source();
+
+            let (mut replay, mut source) = input
+                .replayable(Duration::from_secs(3))
+                .expect("longer then 100ms");
+
+            source.by_ref().take(3).count();
+            let yielded: Vec<Sample> = replay.by_ref().take(3).collect();
+            assert_eq!(&yielded, &SAMPLES[0..3],);
+
+            source.count();
+            let yielded: Vec<Sample> = replay.collect();
+            assert_eq!(&yielded, &SAMPLES[3..5],);
+        }
+
+        #[test]
+        fn keeps_only_latest() {
+            let input = test_source();
+
+            let (mut replay, mut source) = input
+                .replayable(Duration::from_secs(2))
+                .expect("longer then 100ms");
+
+            source.by_ref().take(5).count(); // get all items but do not end the source
+            let yielded: Vec<Sample> = replay.by_ref().take(2).collect();
+            assert_eq!(&yielded, &SAMPLES[3..5]);
+            source.count(); // exhaust source
+            assert_eq!(replay.next(), None);
+        }
+
+        #[test]
+        fn keeps_correct_amount_of_seconds() {
+            let input = StaticSamplesBuffer::new(nz!(1), nz!(16_000), &[0.0; 40_000]);
+
+            let (replay, mut source) = input
+                .replayable(Duration::from_secs(2))
+                .expect("longer then 100ms");
+
+            // exhaust but do not yet end source
+            source.by_ref().take(40_000).count();
+
+            // take all samples we can without blocking
+            let ready = replay.samples_ready();
+            let n_yielded = replay.take_samples(ready).count();
+
+            let max = source.sample_rate().get() * source.channels().get() as u32 * 2;
+            let margin = 16_000 / 10; // 100ms
+            assert!(n_yielded as u32 >= max - margin);
+        }
+
+        #[test]
+        fn samples_ready() {
+            let input = StaticSamplesBuffer::new(nz!(1), nz!(16_000), &[0.0; 40_000]);
+            let (mut replay, source) = input
+                .replayable(Duration::from_secs(2))
+                .expect("longer then 100ms");
+            assert_eq!(replay.by_ref().samples_ready(), 0);
+
+            source.take(8000).count(); // half a second
+            let margin = 16_000 / 10; // 100ms
+            let ready = replay.samples_ready();
+            assert!(ready >= 8000 - margin);
+        }
+    }
+}