refactor buffered + fix size_hint for TestSource

add tests for buffered, fix buffered `parameters_changed`

Author: dvdsk

src/source/buffered.rs

@@ -1,137 +1,45 @@
+/// A iterator that stores extracted data in memory while allowing
+/// concurrent reading in real time.
 use std::mem;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use super::SeekError;
 use crate::common::{ChannelCount, SampleRate};
+use crate::math::PrevMultipleOf;
 use crate::Source;
 
-/// Internal function that builds a `Buffered` object.
-#[inline]
-pub fn buffered<I>(input: I) -> Buffered<I>
-where
-    I: Source,
-{
-    let total_duration = input.total_duration();
-    let first_span = extract(input);
-
-    Buffered {
-        current_span: first_span,
-        position_in_span: 0,
-        total_duration,
-    }
-}
-
-/// Iterator that at the same time extracts data from the iterator and stores it in a buffer.
+/// Iterator that at the same time extracts data from the iterator and
+/// stores it in a buffer.
 pub struct Buffered<I>
 where
     I: Source,
 {
     /// Immutable reference to the next span of data. Cannot be `Span::Input`.
     current_span: Arc<Span<I>>,
 
+    parameters_changed: bool,
+
     /// The position in number of samples of this iterator inside `current_span`.
     position_in_span: usize,
 
     /// Obtained once at creation and never modified again.
     total_duration: Option<Duration>,
 }
 
-enum Span<I>
-where
-    I: Source,
-{
-    /// Data that has already been extracted from the iterator. Also contains a pointer to the
-    /// next span.
-    Data(SpanData<I>),
-
-    /// No more data.
-    End,
-
-    /// Unextracted data. The `Option` should never be `None` and is only here for easier data
-    /// processing.
-    Input(Mutex<Option<I>>),
-}
-
-struct SpanData<I>
-where
-    I: Source,
-{
-    data: Vec<I::Item>,
-    channels: ChannelCount,
-    rate: SampleRate,
-    next: Mutex<Arc<Span<I>>>,
-}
-
-impl<I> Drop for SpanData<I>
-where
-    I: Source,
-{
-    fn drop(&mut self) {
-        // This is necessary to prevent stack overflows deallocating long chains of the mutually
-        // recursive `Span` and `SpanData` types. This iteratively traverses as much of the
-        // chain as needs to be deallocated, and repeatedly "pops" the head off the list. This
-        // solves the problem, as when the time comes to actually deallocate the `SpanData`,
-        // the `next` field will contain a `Span::End`, or an `Arc` with additional references,
-        // so the depth of recursive drops will be bounded.
-        while let Ok(arc_next) = self.next.get_mut() {
-            if let Some(next_ref) = Arc::get_mut(arc_next) {
-                // This allows us to own the next Span.
-                let next = mem::replace(next_ref, Span::End);
-                if let Span::Data(next_data) = next {
-                    // Swap the current SpanData with the next one, allowing the current one
-                    // to go out of scope.
-                    *self = next_data;
-                } else {
-                    break;
-                }
-            } else {
-                break;
-            }
-        }
-    }
-}
-
-/// Builds a span from the input iterator.
-fn extract<I>(mut input: I) -> Arc<Span<I>>
-where
-    I: Source,
-{
-    if input.parameters_changed() {
-        return Arc::new(Span::End);
-    }
-
-    let channels = input.channels();
-    let rate = input.sample_rate();
+impl<I: Source> Buffered<I> {
+    pub(crate) fn new(input: I) -> Buffered<I> {
+        let total_duration = input.total_duration();
+        let first_span = extract(input);
 
-    let mut data = Vec::new();
-    loop {
-        let Some(element) = input.next() else { break };
-        data.push(element);
-        if input.parameters_changed() {
-            break;
-        }
-        if data.len() > 32768 {
-            break;
+        Buffered {
+            current_span: first_span,
+            position_in_span: 0,
+            total_duration,
+            parameters_changed: false,
         }
     }
 
-    if data.is_empty() {
-        return Arc::new(Span::End);
-    }
-
-    Arc::new(Span::Data(SpanData {
-        data,
-        channels,
-        rate,
-        next: Mutex::new(Arc::new(Span::Input(Mutex::new(Some(input))))),
-    }))
-}
-
-impl<I> Buffered<I>
-where
-    I: Source,
-{
     /// Advances to the next span.
     fn next_span(&mut self) {
         let next_span = {
@@ -145,6 +53,7 @@ where
                 Span::End => next_span_ptr.clone(),
                 Span::Input(input) => {
                     let input = input.lock().unwrap().take().unwrap();
+                    dbg!();
                     extract(input)
                 }
             };
@@ -169,7 +78,7 @@ where
         let current_sample;
         let advance_span;
 
-        match &*self.current_span {
+        match dbg!(&*self.current_span) {
             Span::Data(SpanData { data, .. }) => {
                 current_sample = Some(data[self.position_in_span]);
                 self.position_in_span += 1;
@@ -185,39 +94,31 @@ where
         };
 
         if advance_span {
+            dbg!();
+            self.parameters_changed = true;
             self.next_span();
+        } else {
+            self.parameters_changed = false;
         }
 
         current_sample
     }
-
-    #[inline]
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        // TODO:
-        (0, None)
-    }
 }
 
-// TODO: implement exactsize iterator when size_hint is done
-
 impl<I> Source for Buffered<I>
 where
     I: Source,
 {
     #[inline]
     fn parameters_changed(&self) -> bool {
-        match &*self.current_span {
-            Span::Data(_) => false,
-            Span::End => true,
-            Span::Input(_) => unreachable!(),
-        }
+        self.parameters_changed
     }
 
     #[inline]
     fn channels(&self) -> ChannelCount {
         match *self.current_span {
             Span::Data(SpanData { channels, .. }) => channels,
-            Span::End => 1,
+            Span::End => 0,
             Span::Input(_) => unreachable!(),
         }
     }
@@ -226,7 +127,7 @@ where
     fn sample_rate(&self) -> SampleRate {
         match *self.current_span {
             Span::Data(SpanData { rate, .. }) => rate,
-            Span::End => 44100,
+            Span::End => 0,
             Span::Input(_) => unreachable!(),
         }
     }
@@ -256,6 +157,105 @@ where
             current_span: self.current_span.clone(),
             position_in_span: self.position_in_span,
             total_duration: self.total_duration,
+            parameters_changed: self.parameters_changed,
+        }
+    }
+}
+
+enum Span<I>
+where
+    I: Source,
+{
+    /// Data that has already been extracted from the iterator.
+    /// Also contains a pointer to the next span.
+    Data(SpanData<I>),
+
+    /// No more data.
+    End,
+
+    /// Unextracted data. The `Option` should never be `None` and is only here for easier data
+    /// processing.
+    Input(Mutex<Option<I>>),
+}
+
+impl<I: Source> std::fmt::Debug for Span<I> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Span::Data(_) => f.write_str("Span::Data"),
+            Span::End => f.write_str("Span::End"),
+            Span::Input(_) => f.write_str("Span::Input"),
         }
     }
 }
+
+struct SpanData<I>
+where
+    I: Source,
+{
+    data: Vec<I::Item>,
+    channels: ChannelCount,
+    rate: SampleRate,
+    next: Mutex<Arc<Span<I>>>,
+}
+
+impl<I> Drop for SpanData<I>
+where
+    I: Source,
+{
+    fn drop(&mut self) {
+        // This is necessary to prevent stack overflows deallocating long chains of the mutually
+        // recursive `Span` and `SpanData` types. This iteratively traverses as much of the
+        // chain as needs to be deallocated, and repeatedly "pops" the head off the list. This
+        // solves the problem, as when the time comes to actually deallocate the `SpanData`,
+        // the `next` field will contain a `Span::End`, or an `Arc` with additional references,
+        // so the depth of recursive drops will be bounded.
+        while let Ok(arc_next) = self.next.get_mut() {
+            if let Some(next_ref) = Arc::get_mut(arc_next) {
+                // This allows us to own the next Span.
+                let next = mem::replace(next_ref, Span::End);
+                if let Span::Data(next_data) = next {
+                    // Swap the current SpanData with the next one, allowing the current one
+                    // to go out of scope.
+                    *self = next_data;
+                } else {
+                    break;
+                }
+            } else {
+                break;
+            }
+        }
+    }
+}
+
+/// Builds a span from the input iterator.
+fn extract<I>(mut input: I) -> Arc<Span<I>>
+where
+    I: Source,
+{
+    let channels = input.channels();
+    let rate = input.sample_rate();
+
+    let mut data = Vec::new();
+    loop {
+        let Some(sample) = input.next() else { break };
+        data.push(sample);
+        dbg!(sample);
+        if input.parameters_changed() {
+            break;
+        }
+        if data.len() > 32768.prev_multiple_of(channels) {
+            break;
+        }
+    }
+
+    if dbg!(data.is_empty()) {
+        return Arc::new(Span::End);
+    }
+
+    Arc::new(Span::Data(SpanData {
+        data,
+        channels,
+        rate,
+        next: Mutex::new(Arc::new(Span::Input(Mutex::new(Some(input))))),
+    }))
+}

src/source/mod.rs

@@ -179,7 +179,7 @@ pub trait Source: Iterator<Item = Sample> {
     where
         Self: Sized,
     {
-        buffered::buffered(self)
+        Buffered::new(self)
     }
 
     /// Mixes this source with another one.