diff --git a/src/agc.rs b/src/agc.rs
index 362469f..2cfda3b 100644
--- a/src/agc.rs
+++ b/src/agc.rs
@@ -1,9 +1,9 @@
-use num_complex::Complex;
-
 use crate::iq_reader::IqChunk;
+use crate::iq_reader::IqSample;
 
 // Automatic Gain Control
-pub struct Agc {
+pub struct Agc<I> {
+    inner: I,
     // Previous power estimate
     pub power_estimate: f32,
     // Previous gain
@@ -16,8 +16,14 @@ pub struct Agc {
     pub max_gain: f32,
 }
 
-impl Agc {
-    pub fn new(sample_rate: f32, target_power: f32, min_gain: f32, max_gain: f32) -> Self {
+impl<I> Agc<I> {
+    pub fn new(
+        inner: I,
+        sample_rate: f32,
+        target_power: f32,
+        min_gain: f32,
+        max_gain: f32,
+    ) -> Self {
         // Target attack time 5 ms
         let tau_attack = 0.005;
 
@@ -33,6 +39,7 @@ impl Agc {
         let current_gain = 1.0;
 
         Self {
+            inner,
             power_estimate,
             current_gain,
             target_power,
@@ -77,7 +84,24 @@ impl Agc {
 
             self.current_gain = final_gain;
 
-            *z = Complex::new(i * final_gain, q * final_gain);
+            *z = IqSample::new(i * final_gain, q * final_gain);
+        }
+    }
+}
+
+impl<I, E> Iterator for Agc<I>
+where
+    I: Iterator<Item = Result<IqChunk, E>>,
+{
+    type Item = Result<IqChunk, E>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner.next()? {
+            Ok(mut chunk) => {
+                self.process_chunk(&mut chunk);
+                Some(Ok(chunk))
+            }
+            Err(e) => Some(Err(e)),
         }
     }
 }
diff --git a/src/fir.rs b/src/fir.rs
index d1d2ef5..5ee9265 100644
--- a/src/fir.rs
+++ b/src/fir.rs
@@ -1,7 +1,8 @@
 // Finite Impulse response + Decimation
 use crate::{iq_reader::IqChunk, iq_reader::IqSample, utils::ring_buffer::RingBuffer};
 
-pub struct Fir<const N: usize> {
+pub struct Fir<I, const N: usize> {
+    inner: I,
     // Filter coefs
     pub taps: [f32; N],
 
@@ -15,9 +16,10 @@ pub struct Fir<const N: usize> {
     pub decimation_index: usize,
 }
 
-impl<const N: usize> Fir<N> {
-    pub fn new(taps: [f32; N], decimation_factor: usize) -> Self {
+impl<I, const N: usize> Fir<I, N> {
+    pub fn new(inner: I, taps: [f32; N], decimation_factor: usize) -> Self {
         Self {
+            inner,
             taps,
             history: RingBuffer::<IqSample>::new(N),
             decimation_factor,
@@ -39,3 +41,20 @@ impl<const N: usize> Fir<N> {
         // TODO: Decimation
     }
 }
+
+impl<I, E, const N: usize> Iterator for Fir<I, N>
+where
+    I: Iterator<Item = Result<IqChunk, E>>,
+{
+    type Item = Result<IqChunk, E>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.inner.next()? {
+            Ok(mut chunk) => {
+                self.process_chunk(&mut chunk);
+                Some(Ok(chunk))
+            }
+            Err(e) => Some(Err(e)),
+        }
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index af40885..2e393f8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,31 +1,25 @@
-use crate::agc::Agc;
-use crate::fir::Fir;
 use crate::iq_reader::FileSource;
+use crate::pipeline::DspPipelineExt;
 use std::error::Error;
 
 mod agc;
 mod fir;
 mod iq_reader;
+mod pipeline;
 mod utils;
 
 fn main() -> Result<(), Box<dyn Error>> {
     let source = FileSource::new("test.iq", 32769)?;
 
-    // TODO: make a clean pipline use like GNU radio with iterators (like chunk.agc(...).fir(...)...)
-
-    // 20 MSps
-    let mut agc = Agc::new(20_000_000.0, 0.1, 0.001, 100.0);
-
     // Fir coefs
     let taps = [1.0; 64];
-    // Fir
-    let mut fir = Fir::<64>::new(taps, 4);
 
-    // Apply Auto Gain Control
-    for chunk_r in source {
-        let mut chunk = chunk_r?;
-        agc.process_chunk(&mut chunk);
-        fir.process_chunk(&mut chunk);
+    let pipeline = source
+        .agc(20_000_000.0, 1.0, 0.001, 100.0)
+        .fir::<64>(taps, 4);
+
+    for chunk_r in pipeline {
+        let _chunk = chunk_r?;
     }
 
     Ok(())
diff --git a/src/pipeline.rs b/src/pipeline.rs
new file mode 100644
index 0000000..b762b07
--- /dev/null
+++ b/src/pipeline.rs
@@ -0,0 +1,15 @@
+use crate::agc::Agc;
+use crate::fir::Fir;
+use crate::iq_reader::IqChunk;
+
+pub trait DspPipelineExt<E>: Iterator<Item = Result<IqChunk, E>> + Sized {
+    fn agc(self, sample_rate: f32, target_power: f32, min_gain: f32, max_gain: f32) -> Agc<Self> {
+        Agc::new(self, sample_rate, target_power, min_gain, max_gain)
+    }
+
+    fn fir<const N: usize>(self, taps: [f32; N], decimation_factor: usize) -> Fir<Self, N> {
+        Fir::new(self, taps, decimation_factor)
+    }
+}
+
+impl<I, E> DspPipelineExt<E> for I where I: Iterator<Item = Result<IqChunk, E>> {}