Added the error reporting and blocking

src/amdemod.rs

@@ -1,11 +1,13 @@
 use num::traits::Float;
 use std::iter::FromIterator;
 
+use super::error::DspError;
+
 pub struct SquaringAMDemodulator<'a, NumType>
 where
     NumType: Float,
 {
-    iterator: Box<dyn Iterator<Item = NumType> + 'a>,
+    iterator: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
 }
 
 impl<'a, NumType> SquaringAMDemodulator<'a, NumType>
@@ -14,7 +16,7 @@ where
 {
     pub fn from<I>(iterator1: I) -> SquaringAMDemodulator<'a, NumType>
     where
-        I: Iterator<Item = NumType> + 'a,
+        I: Iterator<Item = Result<NumType, DspError>> + 'a,
     {
         SquaringAMDemodulator {
             iterator: Box::new(iterator1),
@@ -26,11 +28,12 @@ impl<'a, NumType> Iterator for SquaringAMDemodulator<'a, NumType>
 where
     NumType: Float,
 {
-    type Item = NumType;
+    type Item = Result<NumType, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         match self.iterator.next() {
-            Some(x) => Some((x * x).sqrt()),
+            Some(Ok(x)) => Some(Ok((x * x).sqrt())),
+            Some(Err(e)) => Some(Err(e)),
             None => None,
         }
     }
@@ -44,10 +47,9 @@ mod tests {
     fn test_amdemod() {
         let test_data = vec![-1_f32, 2_f32, -3_f32, 4_f32, -5_f32];
 
-        let demod = SquaringAMDemodulator::from(test_data.into_iter());
+        let demod = SquaringAMDemodulator::from(test_data.into_iter().map(|x| Ok(x)));
 
-        let result_data = Vec::from_iter(demod);
+        let result_data: Result<Vec<f32>, DspError> = demod.collect();
-        assert_eq!(result_data.len(), 5);
+        assert_eq!(result_data, Ok(vec![1_f32, 2_f32, 3_f32, 4_f32, 5_f32]));
-        assert_eq!(result_data, [1_f32, 2_f32, 3_f32, 4_f32, 5_f32]);
     }
 }

src/error.rs

@@ -0,0 +1,7 @@
+use std::fmt::Debug;
+
+#[derive(Clone, PartialEq, Debug)]
+pub enum DspError {
+    WouldBlock,
+    Error(String),
+}

src/firfilter.rs

@@ -1,6 +1,8 @@
 use num::traits::Num;
 use std::iter::FromIterator;
 
+use super::error::DspError;
+
 pub struct FIRFilter<'a, NumType>
 where
     NumType: Num + Copy,
@@ -8,7 +10,7 @@ where
     coeffs: &'a [NumType],
     state: Vec<NumType>,
     pos: usize,
-    iterator: Box<dyn Iterator<Item = NumType> + 'a>,
+    iterator: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
 }
 
 impl<'a, NumType> FIRFilter<'a, NumType>
@@ -17,7 +19,7 @@ where
 {
     pub fn from<I>(iterator: I, coeffs: &'a [NumType]) -> FIRFilter<'a, NumType>
     where
-        I: Iterator<Item = NumType> + 'a,
+        I: Iterator<Item = Result<NumType, DspError>> + 'a,
     {
         let mut state = Vec::new();
         for _ in 0..coeffs.len() {
@@ -37,11 +39,12 @@ impl<'a, NumType> Iterator for FIRFilter<'a, NumType>
 where
     NumType: Num + Copy,
 {
-    type Item = NumType;
+    type Item = Result<NumType, DspError>;
 
-    fn next(&mut self) -> Option<NumType> {
+    fn next(&mut self) -> Option<Self::Item> {
         let cur = match self.iterator.next() {
-            Some(x) => x,
+            Some(Ok(x)) => x,
+            Some(Err(e)) => return Some(Err(e)),
             None => return None,
         };
 
@@ -51,10 +54,10 @@ where
         let mut result = NumType::zero();
         for i in 0..self.coeffs.len() {
             let pos = (self.pos + self.coeffs.len() - i) % self.coeffs.len();
-            result = result + (self.state[pos] * self.coeffs[i].clone());
+            result = result + (self.state[pos] * self.coeffs[i]);
         }
 
-        Some(result)
+        Some(Ok(result))
     }
 }
 
@@ -66,13 +69,14 @@ mod tests {
     fn fir_avg_test() {
         let test_data = vec![0_f32, 1_f32, 2_f32, 3_f32, 4_f32, 5_f32, 6_f32, 7_f32];
 
-        let upsampler = FIRFilter::from(test_data.into_iter(), &[0.5_f32, 0.5_f32]);
+        let upsampler = FIRFilter::from(test_data.into_iter().map(|x| Ok(x)), &[0.5_f32, 0.5_f32]);
 
-        let result_data = Vec::from_iter(upsampler);
+        let result_data: Result<Vec<f32>, DspError> = upsampler.collect();
-        assert_eq!(result_data.len(), 8);
         assert_eq!(
             result_data,
-            [0.0_f32, 0.5_f32, 1.5_f32, 2.5_f32, 3.5_f32, 4.5_f32, 5.5_f32, 6.5_f32]
+            Ok(vec![
+                0.0_f32, 0.5_f32, 1.5_f32, 2.5_f32, 3.5_f32, 4.5_f32, 5.5_f32, 6.5_f32
+            ]),
         );
     }
 }

src/lib.rs

@@ -1,5 +1,6 @@
 mod amdemod;
+mod error;
 mod firfilter;
-mod resamplers;
 mod multiply;
+mod resamplers;
 mod sources;

src/multiply.rs

@@ -1,23 +1,29 @@
 use num::traits::Num;
 use std::iter::FromIterator;
 
+use super::error::DspError;
+
 pub struct Multiply<'a, NumType>
 where
     NumType: Num + Copy,
 {
-    iterator_a: Box<dyn Iterator<Item = NumType> + 'a>,
+    buffer: Option<NumType>,
-    iterator_b: Box<dyn Iterator<Item = NumType> + 'a>,
+    iterator_a: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
+    iterator_b: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
 }
 
 impl<'a, NumType> Multiply<'a, NumType>
 where
     NumType: Num + Copy,
 {
-    pub fn from<I>(iterator_a: I, iterator_b: I) -> Multiply<'a, NumType>
+    // Use two different types, because map will make its closure part of the type
+    pub fn from<IA, IB>(iterator_a: IA, iterator_b: IB) -> Multiply<'a, NumType>
     where
-        I: Iterator<Item = NumType> + 'a,
+        IA: Iterator<Item = Result<NumType, DspError>> + 'a,
+        IB: Iterator<Item = Result<NumType, DspError>> + 'a,
     {
         Multiply {
+            buffer: None,
             iterator_a: Box::new(iterator_a),
             iterator_b: Box::new(iterator_b),
         }
@@ -28,20 +34,32 @@ impl<'a, NumType> Iterator for Multiply<'a, NumType>
 where
     NumType: Num + Copy,
 {
-    type Item = NumType;
+    type Item = Result<NumType, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        let value_a = match self.iterator_a.next() {
+        let value_a = match self.buffer {
-            Some(a) => a,
+            None => match self.iterator_a.next() {
-            None => return None,
+                Some(Ok(a)) => a,
+                Some(Err(e)) => return Some(Err(e)),
+                None => return None,
+            },
+            Some(a) => {
+                self.buffer = None;
+                a
+            }
         };
 
         let value_b = match self.iterator_b.next() {
-            Some(b) => b,
+            Some(Ok(b)) => b,
+            Some(Err(e)) => {
+                // Can't compute anything yet, buffer value_a until iterator_b stops blocking.
+                self.buffer = Some(value_a);
+                return Some(Err(e));
+            }
             None => return None,
         };
 
-        Some(value_a * value_b)
+        Some(Ok(value_a * value_b))
     }
 }
 
@@ -50,7 +68,7 @@ where
     NumType: Num + Copy,
 {
     factor: NumType,
-    iterator: Box<dyn Iterator<Item = NumType> + 'a>,
+    iterator: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
 }
 
 impl<'a, NumType> MultiplyConst<'a, NumType>
@@ -59,7 +77,7 @@ where
 {
     pub fn from<I>(iterator: I, factor: NumType) -> MultiplyConst<'a, NumType>
     where
-        I: Iterator<Item = NumType> + 'a,
+        I: Iterator<Item = Result<NumType, DspError>> + 'a,
     {
         MultiplyConst {
             factor: factor,
@@ -72,15 +90,16 @@ impl<'a, NumType> Iterator for MultiplyConst<'a, NumType>
 where
     NumType: Num + Copy,
 {
-    type Item = NumType;
+    type Item = Result<NumType, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         let value = match self.iterator.next() {
-            Some(x) => x,
+            Some(Ok(x)) => x,
+            Some(Err(e)) => return Some(Err(e)),
             None => return None,
         };
 
-        return Some(value * self.factor);
+        Some(Ok(value * self.factor))
     }
 }
 
@@ -93,21 +112,45 @@ mod tests {
         let test_a = vec![1_f32, 2_f32, 3_f32];
         let test_b = vec![4_f32, 5_f32, 6_f32, 7_f32];
 
-        let result = Multiply::from(test_a.into_iter(), test_b.into_iter());
+        let iter_a = test_a.into_iter().map(|a| Ok(a));
+        let iter_b = test_b.into_iter().map(|b| Ok(b));
 
-        let result_data = Vec::from_iter(result);
+        let result = Multiply::from(iter_a, iter_b);
-        assert_eq!(result_data.len(), 3);
+
-        assert_eq!(result_data, [4_f32, 10_f32, 18_f32]);
+        let result_data: Result<Vec<f32>, DspError> = result.collect();
+        assert_eq!(result_data, Ok(vec![4_f32, 10_f32, 18_f32]));
+    }
+
+    #[test]
+    fn multiply_blocking_test() {
+        let test_a = vec![1_f32, 2_f32, 3_f32];
+        let test_b = vec![
+            Ok(4_f32),
+            Err(DspError::WouldBlock),
+            Ok(5_f32),
+            Ok(6_f32),
+            Ok(7_f32),
+        ];
+
+        let iter_a = test_a.into_iter().map(|a| Ok(a));
+        let iter_b = test_b.into_iter();
+
+        let mut mult = Multiply::from(iter_a, iter_b);
+
+        assert_eq!(mult.next(), Some(Ok(4_f32)));
+        assert_eq!(mult.next(), Some(Err(DspError::WouldBlock)));
+        assert_eq!(mult.next(), Some(Ok(10_f32)));
+        assert_eq!(mult.next(), Some(Ok(18_f32)));
+        assert_eq!(mult.next(), None);
     }
 
     #[test]
     fn multiplyconst_test() {
         let test_a = vec![1_f32, 2_f32, 3_f32];
 
-        let result = MultiplyConst::from(test_a.into_iter(), 5_f32);
+        let result = MultiplyConst::from(test_a.into_iter().map(|x| Ok(x)), 5_f32);
 
-        let result_data = Vec::from_iter(result);
+        let result_data: Result<Vec<f32>, DspError> = result.collect();
-        assert_eq!(result_data.len(), 3);
+        assert_eq!(result_data, Ok(vec![5_f32, 10_f32, 15_f32]));
-        assert_eq!(result_data, [5_f32, 10_f32, 15_f32]);
     }
 }

src/resamplers.rs

@@ -1,6 +1,8 @@
 use num::traits::{Num, NumCast};
 use std::iter::FromIterator;
 
+use super::error::DspError;
+
 pub struct Upsampler<'a, NumType>
 where
     NumType: Num + Copy,
@@ -8,7 +10,7 @@ where
     factor: u16,
     state: u16,
     sample: Option<NumType>,
-    iterator: Box<dyn Iterator<Item = NumType> + 'a>,
+    iterator: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
 }
 
 impl<'a, NumType> Upsampler<'a, NumType>
@@ -17,7 +19,7 @@ where
 {
     pub fn from<I>(iterator: I, factor: u16) -> Upsampler<'a, NumType>
     where
-        I: Iterator<Item = NumType> + 'a,
+        I: Iterator<Item = Result<NumType, DspError>> + 'a,
     {
         Upsampler {
             factor: factor,
@@ -32,15 +34,22 @@ impl<'a, NumType> Iterator for Upsampler<'a, NumType>
 where
     NumType: Num + Copy,
 {
-    type Item = NumType;
+    type Item = Result<NumType, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         if self.state == 0 {
-            self.sample = self.iterator.next();
+            self.sample = match self.iterator.next() {
+                None => None,
+                Some(Ok(x)) => Some(x),
+                Some(Err(e)) => return Some(Err(e)),
+            };
         }
         self.state = (self.state + 1) % self.factor;
 
-        return self.sample;
+        match self.sample {
+            None => None,
+            Some(x) => Some(Ok(x)),
+        }
     }
 }
 
@@ -49,7 +58,7 @@ where
     NumType: Num + Copy + NumCast,
 {
     factor: u16,
-    iterator: Box<dyn Iterator<Item = NumType> + 'a>,
+    iterator: Box<dyn Iterator<Item = Result<NumType, DspError>> + 'a>,
 }
 
 impl<'a, NumType> Downsampler<'a, NumType>
@@ -58,7 +67,7 @@ where
 {
     pub fn from<I>(iterator: I, factor: u16) -> Downsampler<'a, NumType>
     where
-        I: Iterator<Item = NumType> + 'a,
+        I: Iterator<Item = Result<NumType, DspError>> + 'a,
     {
         Downsampler {
             factor: factor,
@@ -71,19 +80,20 @@ impl<'a, NumType> Iterator for Downsampler<'a, NumType>
 where
     NumType: Num + Copy + NumCast,
 {
-    type Item = NumType;
+    type Item = Result<NumType, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         let mut result = NumType::zero();
         for _ in 0..self.factor {
             match self.iterator.next() {
-                Some(x) => result = result + x,
+                Some(Ok(x)) => result = result + x,
+                Some(Err(e)) => return Some(Err(e)),
                 None => return None,
             }
         }
         result = result / NumType::from(self.factor).unwrap();
 
-        return Some(result);
+        Some(Ok(result))
     }
 }
 
@@ -95,13 +105,14 @@ mod tests {
     fn upsampler_test() {
         let test_data = vec![1_f32, 2_f32, 3_f32, 4_f32, 5_f32];
 
-        let upsampler = Upsampler::from(test_data.into_iter(), 2);
+        let upsampler = Upsampler::from(test_data.into_iter().map(|x| Ok(x)), 2);
 
-        let result_data = Vec::from_iter(upsampler);
+        let result_data: Result<Vec<f32>, DspError> = upsampler.collect();
-        assert_eq!(result_data.len(), 10);
         assert_eq!(
             result_data,
-            [1_f32, 1_f32, 2_f32, 2_f32, 3_f32, 3_f32, 4_f32, 4_f32, 5_f32, 5_f32]
+            Ok(vec![
+                1_f32, 1_f32, 2_f32, 2_f32, 3_f32, 3_f32, 4_f32, 4_f32, 5_f32, 5_f32
+            ]),
         );
     }
 
@@ -111,10 +122,9 @@ mod tests {
             1_f32, 1_f32, 2_f32, 2_f32, 3_f32, 3_f32, 4_f32, 4_f32, 5_f32, 5_f32,
         ];
 
-        let downsampler = Downsampler::from(test_data.into_iter(), 2);
+        let downsampler = Downsampler::from(test_data.into_iter().map(|x| Ok(x)), 2);
 
-        let result_data = Vec::from_iter(downsampler);
+        let result_data: Result<Vec<f32>, DspError> = downsampler.collect();
-        assert_eq!(result_data.len(), 5);
+        assert_eq!(result_data, Ok(vec![1_f32, 2_f32, 3_f32, 4_f32, 5_f32]));
-        assert_eq!(result_data, [1_f32, 2_f32, 3_f32, 4_f32, 5_f32]);
     }
 }

src/sources.rs

@@ -1,5 +1,7 @@
 use std::f32::consts::PI;
 
+use super::error::DspError;
+
 pub struct SinSource {
     sampling_rate: f32,
     frequency: f32,
@@ -17,7 +19,7 @@ impl SinSource {
 }
 
 impl Iterator for SinSource {
-    type Item = f32;
+    type Item = Result<f32, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         self.phase_acc += (1.0 / self.sampling_rate) * self.frequency * 2.0 * PI;
@@ -25,7 +27,7 @@ impl Iterator for SinSource {
             self.phase_acc -= 2.0 * PI;
         }
 
-        Some(self.phase_acc.sin())
+        Some(Ok(self.phase_acc.sin()))
     }
 }
 
@@ -46,7 +48,7 @@ impl CosSource {
 }
 
 impl Iterator for CosSource {
-    type Item = f32;
+    type Item = Result<f32, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         self.phase_acc += (1.0 / self.sampling_rate) * self.frequency * 2.0 * PI;
@@ -54,20 +56,20 @@ impl Iterator for CosSource {
             self.phase_acc -= 2.0 * PI;
         }
 
-        Some(self.phase_acc.cos())
+        Some(Ok(self.phase_acc.cos()))
     }
 }
 
 pub struct VFOSource<'a> {
     sampling_rate: f32,
     phase_acc: f32,
-    frequency_iterator: Box<dyn Iterator<Item = f32> + 'a>,
+    frequency_iterator: Box<dyn Iterator<Item = Result<f32, DspError>> + 'a>,
 }
 
 impl<'a> VFOSource<'a> {
     pub fn new<I>(frequency_iterator: I, sampling_rate: f32) -> VFOSource<'a>
     where
-        I: Iterator<Item = f32> + 'a,
+        I: Iterator<Item = Result<f32, DspError>> + 'a,
     {
         VFOSource {
             sampling_rate: sampling_rate,
@@ -78,11 +80,13 @@ impl<'a> VFOSource<'a> {
 }
 
 impl<'a> Iterator for VFOSource<'a> {
-    type Item = f32;
+    type Item = Result<f32, DspError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         let current_frequency = match self.frequency_iterator.next() {
-            Some(x) => x,
+            Some(Ok(x)) => x,
+            // Propagate any upstream errors
+            Some(Err(e)) => return Some(Err(e)),
             None => return None,
         };
 
@@ -91,6 +95,6 @@ impl<'a> Iterator for VFOSource<'a> {
             self.phase_acc -= 2.0 * PI;
         }
 
-        Some(self.phase_acc.sin())
+        Some(Ok(self.phase_acc.sin()))
     }
 }