#39 WIP - updates tunerchannelsource as the parent class for the CIC

and polyphase sub class implementations.  Converts all source implementations
and tuners to treat sample rate as a double.

src/audio/broadcast/StreamManager.java

@@ -140,7 +140,7 @@ public class StreamManager implements Listener<AudioPacket>
                     {
                         AudioRecorder recorder = BroadcastFactory.getAudioRecorder(getTemporaryRecordingPath(),
                             mBroadcastFormat);
-                        recorder.start(ThreadPool.SCHEDULED);
+                        recorder.start();
                         recorder.receive(audioPacket);
                         mStreamRecorders.put(channelMetadataID, recorder);
                     }

src/dsp/filter/channelizer/ChannelizerViewer.java

@@ -71,7 +71,7 @@ public class ChannelizerViewer extends JFrame
     {
         mTestTuner = new TestTuner();
 
-        mChannelCount = mTestTuner.getTunerController().getSampleRate() / CHANNEL_BANDWIDTH;
+        mChannelCount = (int)(mTestTuner.getTunerController().getSampleRate() / CHANNEL_BANDWIDTH);
         mChannelsPerRow = channelsPerRow;
 
         mToneFrequency = mTestTuner.getTunerController().getFrequency() + 19200;
@@ -127,7 +127,7 @@ public class ChannelizerViewer extends JFrame
             mControlPanel.setLayout(new MigLayout("insets 0 0 0 0", "", ""));
 
             mControlPanel.add(new JLabel("Tone:"), "align left");
-            int maximumFrequencyOffset = mTestTuner.getTunerController().getSampleRate() / 2;
+            int maximumFrequencyOffset = (int)(mTestTuner.getTunerController().getSampleRate() / 2);
             long minimumFrequency = mTestTuner.getTunerController().getFrequency() - maximumFrequencyOffset;
             long maximumFrequency = mTestTuner.getTunerController().getFrequency() + maximumFrequencyOffset;
 
@@ -185,16 +185,16 @@ public class ChannelizerViewer extends JFrame
         {
             setLayout(new MigLayout("insets 0 0 0 0", "fill", "fill"));
 
-            long spectralBandwidth = mTestTuner.getTunerController().getSampleRate();
+            double spectralBandwidth = mTestTuner.getTunerController().getSampleRate();
-            long halfSpectralBandwidth = spectralBandwidth / 2;
+            double halfSpectralBandwidth = spectralBandwidth / 2.0;
 
             int channelToLog = -1;
 
             for(int x = 0; x < mChannelCount; x++)
             {
                 //place the channels left and right of 10.0 MHz
-                long frequency = mTestTuner.getTunerController().getFrequency() +
+                long frequency = (long)(mTestTuner.getTunerController().getFrequency() +
-                    ((x * CHANNEL_BANDWIDTH) - halfSpectralBandwidth) + (CHANNEL_BANDWIDTH / 2);
+                    ((x * CHANNEL_BANDWIDTH) - halfSpectralBandwidth) + (CHANNEL_BANDWIDTH / 2));
 
                 mLog.debug("Channel " + x + " Frequency: " + frequency);
 
@@ -219,7 +219,7 @@ public class ChannelizerViewer extends JFrame
         private ComplexDecibelConverter mComplexDecibelConverter = new ComplexDecibelConverter();
         private SpectrumPanel mSpectrumPanel;
 
-        public PrimarySpectrumPanel(SettingsManager settingsManager, int sampleRate)
+        public PrimarySpectrumPanel(SettingsManager settingsManager, double sampleRate)
         {
             setLayout(new MigLayout("insets 0 0 0 0", "[grow,fill]", "[grow,fill]"));
             mSpectrumPanel = new SpectrumPanel(settingsManager);
@@ -256,7 +256,7 @@ public class ChannelizerViewer extends JFrame
         private SpectrumPanel mSpectrumPanel;
         private long mFrequency;
 
-        public ChannelPanel(SettingsManager settingsManager, int sampleRate, long frequency, int bandwidth, boolean enableLogging)
+        public ChannelPanel(SettingsManager settingsManager, double sampleRate, long frequency, int bandwidth, boolean enableLogging)
         {
             setLayout(new MigLayout("insets 0 0 0 0", "[grow,fill]", "[grow,fill]"));
             mSpectrumPanel = new SpectrumPanel(settingsManager);

src/dsp/filter/channelizer/PolyphaseChannelSource.java

@@ -18,55 +18,41 @@
  ******************************************************************************/
 package dsp.filter.channelizer;
 
-import source.heartbeat.Heartbeat;
 import dsp.filter.channelizer.output.IPolyphaseChannelOutputProcessor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import sample.Listener;
 import sample.complex.ComplexBuffer;
 import sample.complex.ComplexToComplexBufferAssembler;
-import source.ComplexSource;
+import sample.real.IOverflowListener;
-import source.ISourceEventProcessor;
 import source.SourceEvent;
-import source.SourceException;
 import source.tuner.channel.TunerChannel;
+import source.tuner.channel.TunerChannelSource;
 
-import java.util.concurrent.ScheduledExecutorService;
+public class PolyphaseChannelSource extends TunerChannelSource
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-
-public class PolyphaseChannelSource extends ComplexSource implements ISourceEventProcessor
 {
     private final static Logger mLog = LoggerFactory.getLogger(PolyphaseChannelSource.class);
 
-    private TunerChannel mTunerChannel;
-    private int mChannelSampleRate;
-
-    private DownstreamSourceEventHandler mDownstreamSourceEventHandler = new DownstreamSourceEventHandler();
-    private UpstreamSourceEventHandler mUpstreamSourceEventHandler;
-
     private ComplexToComplexBufferAssembler mComplexBufferAssembler = new ComplexToComplexBufferAssembler(2500);
     private IPolyphaseChannelOutputProcessor mPolyphaseChannelOutputProcessor;
-    private ChannelResultsProcessor mChannelResultsProcessor;
+    private double mChannelSampleRate;
-    private ScheduledFuture<?> mChannelResultsProcessorFuture;
 
     /**
-     * Polyphase channelizer channel implementation.  Adapts the channel array output samples from the polyphase
+     * Polyphase channelizer tuner channel source implementation.  Adapts the channel array output samples from the
-     * channelizer into a single channel, or a channel synthesized from two adjacent channels that is frequency
+     * polyphase channelizer into a single channel, or a channel synthesized from two adjacent channels that is
-     * translated and decimated to a single channel.
+     * frequency translated and decimated to a single channel.
      *
-     * @param tunerChannel - requested output tuner channel frequency and bandwidth.
+     * @param tunerChannel describing the desired channel frequency and bandwidth/minimum sample rate
      * @param outputProcessor - to process polyphase channelizer channel results into a channel stream
-     * @param externalEventListener to process sample stream start/stop requests
+     * @param producerSourceEventListener to receive source event requests (e.g. start/stop sample stream)
+     * @param channelSampleRate for the downstream sample output
      */
     public PolyphaseChannelSource(TunerChannel tunerChannel, IPolyphaseChannelOutputProcessor outputProcessor,
-                                  Listener<SourceEvent> externalEventListener, int channelSampleRate)
+                                  Listener<SourceEvent> producerSourceEventListener, double channelSampleRate)
     {
-        mTunerChannel = tunerChannel;
+        super(producerSourceEventListener, tunerChannel);
         mPolyphaseChannelOutputProcessor = outputProcessor;
-        mUpstreamSourceEventHandler = new UpstreamSourceEventHandler(externalEventListener);;
         mChannelSampleRate = channelSampleRate;
-
     }
 
     /**
@@ -75,21 +61,8 @@ public class PolyphaseChannelSource extends ComplexSource implements ISourceEven
     @Override
     public void setListener(final Listener<ComplexBuffer> listener)
     {
-//        mComplexBufferAssembler.setListener(listener);
+        mComplexBufferAssembler.setListener(listener);
-
+   }
-        mComplexBufferAssembler.setListener(new Listener<ComplexBuffer>()
-        {
-            @Override
-            public void receive(ComplexBuffer complexBuffer)
-            {
-                if(mTunerChannel.getFrequency() == 99981250)
-                {
-//                    mLog.debug(Arrays.toString(complexBuffer.getSamples()));
-                }
-                listener.receive(complexBuffer);
-            }
-        });
-    }
 
     /**
      * Removes the listener from receiving complex sample buffers from this channel source
@@ -100,97 +73,6 @@ public class PolyphaseChannelSource extends ComplexSource implements ISourceEven
         mComplexBufferAssembler.setListener(null);
     }
 
-    /**
-     * Starts scheduled thread pool processing of the inbound polyphase channel results buffer to produce channelized
-     * complex sample buffer output for broadcast to a registered complex sample buffer listener.
-     *
-     * @param executor to use when scheduling a buffer processor
-     */
-    @Override
-    public void start(ScheduledExecutorService executor)
-    {
-        if(mUpstreamSourceEventHandler != null)
-        {
-            mUpstreamSourceEventHandler.broadcast(SourceEvent.startSampleStream(PolyphaseChannelSource.this));
-        }
-
-        mChannelResultsProcessor = new ChannelResultsProcessor();
-
-        if(mChannelResultsProcessorFuture != null)
-        {
-            mChannelResultsProcessorFuture.cancel(true);
-            mLog.error("An existing channel results processor scheduled future was cancelled");
-        }
-
-        //Schedule the results processor to run every 100 milliseconds
-        mChannelResultsProcessorFuture =
-            executor.scheduleAtFixedRate(mChannelResultsProcessor, 0, 100, TimeUnit.MILLISECONDS);
-    }
-
-    /**
-     * Stops scheduled thread pool processing of the inbound polyphase channel results buffer to produce channelized
-     * complex sample buffer output for broadcast to a registered complex sample buffer listener.
-     */
-    @Override
-    public void stop()
-    {
-        if(mUpstreamSourceEventHandler != null)
-        {
-            mUpstreamSourceEventHandler.broadcast(SourceEvent.stopSampleStream(PolyphaseChannelSource.this));
-        }
-
-        if(mChannelResultsProcessorFuture != null)
-        {
-            mChannelResultsProcessorFuture.cancel(true);
-            mChannelResultsProcessorFuture = null;
-        }
-
-        mChannelResultsProcessor = null;
-    }
-
-
-    /**
-     * Module-level ISourceEventListener interface implementation for processing downstream events
-     */
-    @Override
-    public Listener<SourceEvent> getSourceEventListener()
-    {
-        return mDownstreamSourceEventHandler;
-    }
-
-    /**
-     * Module-level ISourceEventProvider interface implementation, sets the listener to receive downstream module
-     * level source events from this source
-     */
-    @Override
-    public void setSourceEventListener(Listener<SourceEvent> listener)
-    {
-        mDownstreamSourceEventHandler.setDownstreamEventListener(listener);
-    }
-
-    /**
-     * Module-level ISourceEventProvider interface implementation, removes the listener from receiving downstream module
-     * level source events from this source
-     */
-    @Override
-    public void removeSourceEventListener()
-    {
-        mDownstreamSourceEventHandler.removeDownstreamEventListener();
-    }
-
-    /**
-     * ISourceEventProcessor implementation for processing source events from the external polyphase channel manager
-     * and/or tuner providing samples to the polphase channelizer implementation.
-     *
-     * @param sourceEvent containing a request or notification
-     * @throws SourceException if there are any errors while processing the source event
-     */
-    @Override
-    public void process(SourceEvent sourceEvent) throws SourceException
-    {
-        mUpstreamSourceEventHandler.receive(sourceEvent);
-    }
-
     /**
      * Channel output processor used by this channel source to convert polyphase channel results into a specific
      * channel complex buffer output stream.
@@ -214,6 +96,13 @@ public class PolyphaseChannelSource extends ComplexSource implements ISourceEven
             //Swap out the processor so that incoming samples can accumulate in the new channel output processor
             mPolyphaseChannelOutputProcessor = outputProcessor;
 
+            //Remove the buffer overflow listener and clear any previous buffer overflow state
+            existingProcessor.setSourceOverflowListener(null);
+            broadcastOverflowState(false);
+
+            //Register to receive buffer overflow notifications so we can push them up through the source chain
+            mPolyphaseChannelOutputProcessor.setSourceOverflowListener(this);
+
             //Fully process the residual channel results buffer of the previous channel output processor
             if(existingProcessor != null)
             {
@@ -222,14 +111,6 @@ public class PolyphaseChannelSource extends ComplexSource implements ISourceEven
         }
     }
 
-    /**
-     * Tuner channel that is being sourced by this polyphase channel source.
-     */
-    public TunerChannel getTunerChannel()
-    {
-        return mTunerChannel;
-    }
-
     /**
      * Primary method for receiving channel results output from a polyphase channelizer.  The results array will be
      * queued for processing to extract the target channel samples, process them for frequency correction and/or
@@ -243,186 +124,80 @@ public class PolyphaseChannelSource extends ComplexSource implements ISourceEven
     }
 
     /**
-     * Downstream sample rate
+     * Downstream channel sample rate
      *
      * @return sample rate in Hertz
-     * @throws SourceException never
      */
     @Override
-    public int getSampleRate() throws SourceException
+    public double getSampleRate()
     {
         return mChannelSampleRate;
     }
 
     /**
-     * Center tuned frequency for this channel.
+     * Sets the downstream channel sample rate
-     *
+     * @param sampleRate in hertz
-     * @return frequency in hertz.
-     * @throws SourceException never
      */
     @Override
-    public long getFrequency() throws SourceException
+    protected void setSampleRate(double sampleRate)
     {
-        return mTunerChannel.getFrequency();
+        mChannelSampleRate = sampleRate;
     }
 
-    @Override
-    public void setHeartbeatListener(Listener<Heartbeat> listener)
-    {
-        mLog.info("Polyphase channel source received request to set heartbeat listener");
-        //TODO: implement this
-    }
-
-    @Override
-    public void removeHeartbeatListener()
-    {
-        mLog.info("Polyphase channel source received request to remove heartbeat listener");
-        //TODO: implement this
-    }
-
-    @Override
-    public void reset()
-    {
-        mLog.info("Polyphase channel source received request to reset");
-    }
-
-
     @Override
     public void dispose()
     {
         mLog.info("Polyphase channel source received request to dispose");
     }
 
+    @Override
+    public long getFrequencyCorrection()
+    {
+        IPolyphaseChannelOutputProcessor processor = mPolyphaseChannelOutputProcessor;
+
+        if(processor != null)
+        {
+            return processor.getFrequencyCorrection();
+        }
+
+        return 0;
+    }
+
     /**
      * Adjusts the frequency correction value that is being applied to the channelized output stream by the
      * polyphase channel output processor.
+     *
      * @param value to apply for frequency correction in hertz
      */
-    private void setFrequencyCorrection(long value)
+    protected void setFrequencyCorrection(long value)
     {
         if(mPolyphaseChannelOutputProcessor != null)
         {
             mPolyphaseChannelOutputProcessor.setFrequencyCorrection(value);
-            mDownstreamSourceEventHandler.broadcast(SourceEvent.frequencyCorrectionChange(value));
+            broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(value));
         }
     }
 
     /**
-     * Runnable for processing the incoming polyphase channelizer results queue and distributing the channel
+     * Sets the center frequency for the incoming sample stream channel results
-     * results to the registered complex sample listener
+     * @param frequency in hertz
      */
-    public class ChannelResultsProcessor implements Runnable
+    @Override
+    protected void setFrequency(long frequency)
     {
-        @Override
+        if(mPolyphaseChannelOutputProcessor != null)
-        public void run()
         {
-            try
+            mPolyphaseChannelOutputProcessor.setFrequency(frequency);
-            {
-                //Lock on the output processor so that it can't be changed out in the middle of processing
-                synchronized(mPolyphaseChannelOutputProcessor)
-                {
-                    mPolyphaseChannelOutputProcessor.processChannelResults(mComplexBufferAssembler);
-                }
-            }
-            catch(Throwable throwable)
-            {
-                mLog.error("Error while processing polyphase channel samples", throwable);
-            }
         }
     }
 
-    public class UpstreamSourceEventHandler implements Listener<SourceEvent>
+    @Override
+    protected void processSamples()
     {
-        private Listener<SourceEvent> mExternalSourceEventListener;
+        //Lock on the output processor so that it can't be changed out in the middle of processing
-
+        synchronized(mPolyphaseChannelOutputProcessor)
-        /**
-         * Upstream source event processor for handling any source events generated by the polyphase channelizer that
-         * provides this polyphase channel source.
-         *
-         * @param externalSourceEventListener to receive requests for start/stop sample streams
-         */
-        public UpstreamSourceEventHandler(Listener<SourceEvent> externalSourceEventListener)
         {
-            mExternalSourceEventListener = externalSourceEventListener;
+            mPolyphaseChannelOutputProcessor.processChannelResults(mComplexBufferAssembler);
-        }
-
-        /**
-         * Broadcasts the request source event to the external source event listener.  This listener is normally the
-         * polyphase channel manager.
-         *
-         * @param sourceEvent to broadcast
-         */
-        public void broadcast(SourceEvent sourceEvent)
-        {
-            if(mExternalSourceEventListener != null)
-            {
-                mExternalSourceEventListener.receive(sourceEvent);
-            }
-        }
-
-        /**
-         * Process source events received from the upstream (ie tuner or channelizer) sample provider.
-         * @param sourceEvent to process
-         */
-        @Override
-        public void receive(SourceEvent sourceEvent)
-        {
-            switch(sourceEvent.getEvent())
-            {
-                //Reset frequency correction anytime the upstream frequency or sample rate changes so that the
-                //downstream modules can reset and recalculate any needed frequency correction
-                case NOTIFICATION_FREQUENCY_CHANGE:
-                case NOTIFICATION_SAMPLE_RATE_CHANGE:
-                    setFrequencyCorrection(0);
-                    break;
-            }
-        }
-    }
-
-    /**
-     * Downstream source event handler implements the ISourceEventProvider and ISourceEventProcessor interfaces on
-     * for this module to handle broadcasting source events to other downstream modules and receiving requests from
-     * any downstream modules.
-     */
-    public class DownstreamSourceEventHandler implements Listener<SourceEvent>
-    {
-        private Listener<SourceEvent> mDownstreamEventListener;
-
-        public void broadcast(SourceEvent sourceEvent)
-        {
-            if(mDownstreamEventListener != null)
-            {
-                mDownstreamEventListener.receive(sourceEvent);
-            }
-        }
-
-        /**
-         * Sets the downstream event listener to receive source events from this channel source module
-         * @param listener to receive events
-         */
-        public void setDownstreamEventListener(Listener<SourceEvent> listener)
-        {
-            mDownstreamEventListener = listener;
-        }
-
-        public void removeDownstreamEventListener()
-        {
-            mDownstreamEventListener = null;
-        }
-
-        /**
-         * Process source events received from the downstream channel modules.
-         * @param sourceEvent to process
-         */
-        @Override
-        public void receive(SourceEvent sourceEvent)
-        {
-            switch(sourceEvent.getEvent())
-            {
-                case REQUEST_CHANNEL_FREQUENCY_CORRECTION_CHANGE:
-                    setFrequencyCorrection(sourceEvent.getValue().intValue());
-                    break;
-            }
         }
     }
 }

src/dsp/filter/channelizer/output/ChannelOutputProcessor.java

@@ -25,6 +25,7 @@ import sample.OverflowableTransferQueue;
 import sample.complex.Complex;
 import sample.complex.ComplexSampleListener;
 import sample.real.IOverflowListener;
+import source.Source;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -50,13 +51,25 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP
      * @param sampleRate of the output channel.  This is used to match the oscillator's sample rate to the output
      * channel sample rate for frequency translation/correction.
      */
-    public ChannelOutputProcessor(int inputChannelCount, int sampleRate)
+    public ChannelOutputProcessor(int inputChannelCount, double sampleRate)
     {
         mInputChannelCount = inputChannelCount;
         mFrequencyCorrectionMixer = new Oscillator(0, sampleRate);
-        mMaxResultsToProcess = sampleRate / 10 * 2;  //process at 100 millis interval, twice the expected inflow rate
+        mMaxResultsToProcess = (int)(sampleRate / 10) * 2;  //process at 100 millis interval, twice the expected inflow rate
 
-        mChannelResultsQueue = new OverflowableTransferQueue<>(sampleRate * 3, (int)(sampleRate * 0.5));
+        mChannelResultsQueue = new OverflowableTransferQueue<>((int)(sampleRate * 3), (int)(sampleRate * 0.5));
+    }
+
+    @Override
+    public void setFrequency(long frequency)
+    {
+
+    }
+
+    @Override
+    public long getFrequencyCorrection()
+    {
+        return 0;
     }
 
     @Override
@@ -100,9 +113,9 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP
     /**
      * Sets the overflow listener to monitor the internal channelizer channel results queue overflow state
      */
-    public void setOverflowListener(IOverflowListener listener)
+    public void setSourceOverflowListener(Source source)
     {
-        mChannelResultsQueue.setOverflowListener(listener);
+        mChannelResultsQueue.setSourceOverflowListener(source);
     }
 
     /**

src/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java

@@ -19,6 +19,8 @@
 package dsp.filter.channelizer.output;
 
 import sample.complex.ComplexSampleListener;
+import sample.real.IOverflowListener;
+import source.Source;
 
 import java.util.List;
 
@@ -37,6 +39,12 @@ public interface IPolyphaseChannelOutputProcessor
      */
     void processChannelResults(ComplexSampleListener listener);
 
+    /**
+     * Sets the center frequency of the incoming channel results samples
+     * @param frequency in hertz
+     */
+    void setFrequency(long frequency);
+
     /**
      * Specifies the frequency correction (+/-) that should be applied to samples extracted from the polyphase
      * channelizer output results.
@@ -45,6 +53,11 @@ public interface IPolyphaseChannelOutputProcessor
      */
     void setFrequencyCorrection(long frequencyCorrection);
 
+    /**
+     * Current frequency correction value.
+     */
+    long getFrequencyCorrection();
+
     /**
      * Indicates the number of input channels processed by this output processor
      * @return
@@ -56,4 +69,10 @@ public interface IPolyphaseChannelOutputProcessor
      * @param indexes
      */
     void setPolyphaseChannelIndices(List<Integer> indexes);
+
+    /**
+     * Sets the listener to receive buffer overflow notifications
+     * @param source to receive overflow notifications
+     */
+    void setSourceOverflowListener(Source source);
 }

src/dsp/filter/channelizer/output/OneChannelOutputProcessor.java

@@ -38,7 +38,7 @@ public class OneChannelOutputProcessor extends ChannelOutputProcessor
      * @param sampleRate of the output sample stream.
      * @param channelIndexes containing a single channel index.
      */
-    public OneChannelOutputProcessor(int sampleRate, List<Integer> channelIndexes)
+    public OneChannelOutputProcessor(double sampleRate, List<Integer> channelIndexes)
     {
         super(1, sampleRate);
         setPolyphaseChannelIndices(channelIndexes);

src/dsp/mixer/Oscillator.java

@@ -41,11 +41,11 @@ public class Oscillator
      * @param frequency - positive or negative frequency in hertz
      * @param sampleRate - in hertz
      */
-    public Oscillator(long frequency, int sampleRate)
+    public Oscillator(long frequency, double sampleRate)
     {
         mEnabled = frequency != 0;
 
-        mSampleRate = (double)sampleRate;
+        mSampleRate = sampleRate;
         mFrequency = (double)frequency;
 
         update();
@@ -87,9 +87,9 @@ public class Oscillator
     /**
      * Sets or changes the sample rate of this oscillator
      */
-    public void setSampleRate(int sampleRate)
+    public void setSampleRate(double sampleRate)
     {
-        mSampleRate = (double)sampleRate;
+        mSampleRate = sampleRate;
         update();
     }
 

src/module/decode/am/AMDecoder.java

@@ -1,54 +1,53 @@
 /*******************************************************************************
- *     SDR Trunk 
+ * sdrtrunk
- *     Copyright (C) 2014,2015 Dennis Sheirer
+ * Copyright (C) 2014-2017 Dennis Sheirer
- * 
+ *
- *     This program is free software: you can redistribute it and/or modify
+ * This program is free software: you can redistribute it and/or modify
- *     it under the terms of the GNU General Public License as published by
+ * it under the terms of the GNU General Public License as published by
- *     the Free Software Foundation, either version 3 of the License, or
+ * the Free Software Foundation, either version 3 of the License, or
- *     (at your option) any later version.
+ * (at your option) any later version.
- * 
+ *
- *     This program is distributed in the hope that it will be useful,
+ * This program is distributed in the hope that it will be useful,
- *     but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
- *     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *     GNU General Public License for more details.
+ * GNU General Public License for more details.
- * 
+ *
- *     You should have received a copy of the GNU General Public License
+ * You should have received a copy of the GNU General Public License
- *     along with this program.  If not, see <http://www.gnu.org/licenses/>
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>
+ *
  ******************************************************************************/
 
 package module.decode.am;
 
-import java.util.concurrent.ScheduledExecutorService;
-
 import module.decode.DecoderType;
 import module.decode.PrimaryDecoder;
 import module.decode.config.DecodeConfiguration;
 
 public class AMDecoder extends PrimaryDecoder
 {
-	public AMDecoder( DecodeConfiguration config )
+    public AMDecoder(DecodeConfiguration config)
-	{
-		super( config );
-	}
-
-	@Override
-    public DecoderType getDecoderType()
     {
-	    return DecoderType.AM;
+        super(config);
     }
 
-	@Override
+    @Override
-	public void reset()
+    public DecoderType getDecoderType()
-	{
+    {
-	}
+        return DecoderType.AM;
+    }
 
-	@Override
+    @Override
-	public void start( ScheduledExecutorService executor )
+    public void reset()
-	{
+    {
-	}
+    }
 
-	@Override
+    @Override
-	public void stop()
+    public void start()
-	{
+    {
-	}
+    }
+
+    @Override
+    public void stop()
+    {
+    }
 }

src/module/decode/nbfm/NBFMDecoder.java

@@ -1,24 +1,23 @@
 /*******************************************************************************
- *     SDR Trunk 
+ * sdrtrunk
- *     Copyright (C) 2014,2015 Dennis Sheirer
+ * Copyright (C) 2014-2017 Dennis Sheirer
- * 
+ *
- *     This program is free software: you can redistribute it and/or modify
+ * This program is free software: you can redistribute it and/or modify
- *     it under the terms of the GNU General Public License as published by
+ * it under the terms of the GNU General Public License as published by
- *     the Free Software Foundation, either version 3 of the License, or
+ * the Free Software Foundation, either version 3 of the License, or
- *     (at your option) any later version.
+ * (at your option) any later version.
- * 
+ *
- *     This program is distributed in the hope that it will be useful,
+ * This program is distributed in the hope that it will be useful,
- *     but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
- *     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *     GNU General Public License for more details.
+ * GNU General Public License for more details.
- * 
+ *
- *     You should have received a copy of the GNU General Public License
+ * You should have received a copy of the GNU General Public License
- *     along with this program.  If not, see <http://www.gnu.org/licenses/>
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>
+ *
  ******************************************************************************/
 package module.decode.nbfm;
 
-import java.util.concurrent.ScheduledExecutorService;
-
 import module.decode.DecoderType;
 import module.decode.PrimaryDecoder;
 import module.decode.config.DecodeConfiguration;
@@ -42,7 +41,7 @@ public class NBFMDecoder extends PrimaryDecoder
 	}
 
 	@Override
-	public void start( ScheduledExecutorService executor )
+	public void start()
 	{
 	}
 

src/record/RecorderManager.java

@@ -145,7 +145,7 @@ public class RecorderManager implements Listener<AudioPacket>
                         {
                             recorder = new RealBufferWaveRecorder(AUDIO_SAMPLE_RATE, filePrefix);
 
-                            recorder.start(ThreadPool.SCHEDULED);
+                            recorder.start();
 
                             recorder.receive(audioPacket.getAudioBuffer());
                             mRecorders.put(identifier, recorder);

src/sample/OverflowableTransferQueue.java

@@ -21,6 +21,7 @@ package sample;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import sample.real.IOverflowListener;
+import source.Source;
 
 import java.util.Collection;
 import java.util.concurrent.LinkedTransferQueue;
@@ -33,6 +34,7 @@ public class OverflowableTransferQueue<E>
 
     public enum State {NORMAL, OVERFLOW};
     private IOverflowListener mOverflowListener;
+    private Source mSourceOverflowListener;
 
     private LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<E>();
     private AtomicInteger mCounter = new AtomicInteger();
@@ -99,13 +101,21 @@ public class OverflowableTransferQueue<E>
     }
 
     /**
-     * Sets a listener to receive overflow state change events
+     * Sets a listener to receive overflow state change events.
      */
     public void setOverflowListener(IOverflowListener listener)
     {
         mOverflowListener = listener;
     }
 
+    /**
+     * Sets the source to receive overflow state change events (in addition to an IOverflow listener)
+     */
+    public void setSourceOverflowListener(Source source)
+    {
+        mSourceOverflowListener = source;
+    }
+
 
     /**
      * Toggles the overflow state and broadcast state change to listener
@@ -118,6 +128,11 @@ public class OverflowableTransferQueue<E>
             {
                 mOverflowListener.sourceOverflow(overflow);
             }
+
+            if(mSourceOverflowListener != null)
+            {
+                mSourceOverflowListener.broadcastOverflowState(overflow);
+            }
         }
     }
 

src/source/Source.java

@@ -52,14 +52,14 @@ public abstract class Source extends Module implements ISourceEventListener, ISo
      *
      * @throws SourceException if there is an issue determining the sample rate for this source
      */
-    public abstract int getSampleRate() throws SourceException;
+    public abstract double getSampleRate();
 
     /**
      * Center frequency for this source in Hertz
      *
      * @throws SourceException if there is an issue in determining the center frequency for this source
      */
-    public abstract long getFrequency() throws SourceException;
+    public abstract long getFrequency();
 
     /**
      * Process any cleanup actions to prepare for garbage collection of this source
@@ -79,7 +79,7 @@ public abstract class Source extends Module implements ISourceEventListener, ISo
      *
      * @param overflow true if overflow, false if normal
      */
-    protected void broadcastOverflowState(boolean overflow)
+    public void broadcastOverflowState(boolean overflow)
     {
         if(mOverflowListener != null)
         {

src/source/SourceEvent.java

@@ -130,7 +130,7 @@ public class SourceEvent
      *
      * @param sampleRate in hertz
      */
-    public static SourceEvent sampleRateChange(int sampleRate)
+    public static SourceEvent sampleRateChange(double sampleRate)
     {
         return new SourceEvent(Event.NOTIFICATION_SAMPLE_RATE_CHANGE, sampleRate);
     }
@@ -140,7 +140,7 @@ public class SourceEvent
      *
      * @param sampleRate in hertz
      */
-    public static SourceEvent channelSampleRateChange(int sampleRate)
+    public static SourceEvent channelSampleRateChange(double sampleRate)
     {
         return new SourceEvent(Event.NOTIFICATION_CHANNEL_SAMPLE_RATE_CHANGE, sampleRate);
     }

src/source/SourceEventListenerToProcessorAdapter.java

@@ -0,0 +1,61 @@
+/*******************************************************************************
+ * sdrtrunk
+ * Copyright (C) 2014-2017 Dennis Sheirer
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>
+ *
+ ******************************************************************************/
+package source;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sample.Listener;
+import source.tuner.channel.TunerChannelSource;
+
+public class SourceEventListenerToProcessorAdapter implements Listener<SourceEvent>
+{
+    private final static Logger mLog = LoggerFactory.getLogger(TunerChannelSource.class);
+
+    private ISourceEventProcessor mSourceEventProcessor;
+
+    /**
+     * Adapter to implement ISourceEventListener/Listener<SourceEvent> and adapt to a source event processor
+     */
+    public SourceEventListenerToProcessorAdapter(ISourceEventProcessor sourceEventProcessor)
+    {
+        mSourceEventProcessor = sourceEventProcessor;
+    }
+
+    public void dispose()
+    {
+        mSourceEventProcessor = null;
+    }
+
+    @Override
+    public void receive(SourceEvent sourceEvent)
+    {
+        if(mSourceEventProcessor != null)
+        {
+            try
+            {
+                mSourceEventProcessor.process(sourceEvent);
+            }
+            catch(SourceException e)
+            {
+                mLog.error("Error while transferring source event to source event processor [" +
+                    mSourceEventProcessor.getClass() + "]", e);
+            }
+        }
+    }
+}

src/source/heartbeat/HeartbeatManager.java

@@ -26,6 +26,9 @@ public class HeartbeatManager
     private static final Heartbeat HEARTBEAT = new Heartbeat();
     private Broadcaster<Heartbeat> mHeartbeatBroadcaster = new Broadcaster<>();
 
+    /**
+     * Broadcasts heartbeat as commanded and handles listener registration details.
+     */
     public HeartbeatManager()
     {
     }
@@ -54,5 +57,4 @@ public class HeartbeatManager
     {
         mHeartbeatBroadcaster.removeListener(listener);
     }
-
 }

src/source/mixer/RealMixerSource.java

@@ -93,33 +93,13 @@ public class RealMixerSource extends RealSource
 		return null;
 	}
 
-    /**
-     * Registers the listener to receive heartbeats from this source.
-     *
-     * @param listener to receive heartbeats
-     */
-    @Override
-    public void setHeartbeatListener(Listener<Heartbeat> listener)
-    {
-        mHeartbeatListener = listener;
-    }
-
-    /**
-     * Removes the currently registered heartbeat listener
-     */
-    @Override
-    public void removeHeartbeatListener()
-    {
-        mHeartbeatListener = null;
-    }
-
     @Override
     public void reset()
     {
     }
 
     @Override
-    public void start(ScheduledExecutorService executor)
+    public void start()
     {
     }
 
@@ -174,11 +154,11 @@ public class RealMixerSource extends RealSource
         }
     }
 
-    public int getSampleRate()
+    public double getSampleRate()
     {
         if(mTargetDataLine != null)
         {
-            return (int)mTargetDataLine.getFormat().getSampleRate();
+            return mTargetDataLine.getFormat().getSampleRate();
         }
         else
         {
@@ -189,7 +169,7 @@ public class RealMixerSource extends RealSource
     /**
      * Returns the frequency of this source.  Default is 0.
      */
-    public long getFrequency() throws SourceException
+    public long getFrequency()
     {
         return mFrequency;
     }

src/source/tuner/ComplexChannelSource.java

@@ -1,169 +0,0 @@
-/*******************************************************************************
- * sdrtrunk
- * Copyright (C) 2014-2017 Dennis Sheirer
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>
- *
- ******************************************************************************/
-package source.tuner;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import sample.Listener;
-import sample.complex.ComplexBuffer;
-import source.ComplexSource;
-import source.ISourceEventProcessor;
-import source.SourceEvent;
-import source.SourceException;
-
-import java.util.concurrent.ScheduledExecutorService;
-
-/**
- * Complex source that provides a complex sample stream at the channel's required sample rate
- */
-public abstract class ComplexChannelSource extends ComplexSource implements ISourceEventProcessor
-{
-    private final static Logger mLog = LoggerFactory.getLogger(ComplexChannelSource.class);
-    private SourceEventProcessor mSourceEventProcessor = new SourceEventProcessor();
-    private Listener<SourceEvent> mSourceEventListener;
-    private Listener<ComplexBuffer> mComplexBufferListener;
-    private ISourceEventProcessor mUpstreamSourceEventProcessor;
-
-    public ComplexChannelSource(ISourceEventProcessor upstreamSourceEventProcessor)
-    {
-        mUpstreamSourceEventProcessor = upstreamSourceEventProcessor;
-    }
-
-    /**
-     * Sets the source event listener to receive source events from this source
-     */
-    @Override
-    public void setSourceEventListener(Listener<SourceEvent> listener)
-    {
-        mSourceEventListener = listener;
-    }
-
-    /**
-     * Removes the source event listener from receiving source events from this source
-     */
-    @Override
-    public void removeSourceEventListener()
-    {
-        mSourceEventListener = null;
-    }
-
-    /**
-     * Broadcasts the source event to the registered source event listener
-     */
-    protected void broadcast(SourceEvent sourceEvent)
-    {
-        if(mSourceEventListener != null)
-        {
-            mSourceEventListener.receive(sourceEvent);
-        }
-    }
-
-    /**
-     * Registers the listener to receive complex sample buffers from this channel source
-     */
-    @Override
-    public void setListener(Listener<ComplexBuffer> listener)
-    {
-        mComplexBufferListener = listener;
-    }
-
-    @Override
-    public void removeListener(Listener<ComplexBuffer> listener)
-    {
-        mComplexBufferListener = null;
-    }
-
-    /**
-     * Dispatches the processed complex buffer to the downstream consumer.
-     * @param complexBuffer containing complex samples
-     */
-    protected void broadcast(ComplexBuffer complexBuffer)
-    {
-        if(mComplexBufferListener != null)
-        {
-            mComplexBufferListener.receive(complexBuffer);
-        }
-    }
-
-    @Override
-    public void start(ScheduledExecutorService executor)
-    {
-        //TODO: start a runnable to process downstream buffer dispatching
-
-        if(mUpstreamSourceEventProcessor != null)
-        {
-            try
-            {
-                mUpstreamSourceEventProcessor.process(SourceEvent.startSampleStream(ComplexChannelSource.this));
-            }
-            catch(SourceException se)
-            {
-                mLog.error("Error starting complex channel source");
-            }
-        }
-    }
-
-    @Override
-    public void stop()
-    {
-        if(mUpstreamSourceEventProcessor != null)
-        {
-            try
-            {
-                mUpstreamSourceEventProcessor.process(SourceEvent.stopSampleStream(ComplexChannelSource.this));
-            }
-            catch(SourceException se)
-            {
-                mLog.error("Error starting complex channel source");
-            }
-        }
-
-        //TODO: cancel a runnable to process downstream buffer dispatching
-    }
-
-
-    /**
-     * Source Event processor interface for this source to receive source events
-     */
-    @Override
-    public Listener<SourceEvent> getSourceEventListener()
-    {
-        return mSourceEventProcessor;
-    }
-
-    /**
-     * Adapter for implementing ISourceEventProcessor and dispatching source events to subclasses via the
-     * ISourceEventProcessor.process() method.
-     */
-    public class SourceEventProcessor implements Listener<SourceEvent>
-    {
-        @Override
-        public void receive(SourceEvent sourceEvent)
-        {
-            try
-            {
-                process(sourceEvent);
-            }
-            catch(SourceException se)
-            {
-                mLog.error("Error processing source event", se);
-            }
-        }
-    }
-}

src/source/tuner/TunerController.java

@@ -21,12 +21,16 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import sample.Broadcaster;
 import sample.Listener;
+import sample.complex.Complex;
 import sample.complex.ComplexBuffer;
 import sample.complex.IComplexBufferProvider;
+import source.ISourceEventListener;
 import source.ISourceEventProcessor;
 import source.SourceEvent;
 import source.SourceEvent.Event;
+import source.SourceEventListenerToProcessorAdapter;
 import source.SourceException;
+import source.tuner.channel.CICTunerChannelSource;
 import source.tuner.channel.TunerChannel;
 import source.tuner.channel.TunerChannelSource;
 import source.tuner.configuration.TunerConfiguration;
@@ -37,7 +41,7 @@ import java.util.SortedSet;
 import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.RejectedExecutionException;
 
-public abstract class TunerController implements Tunable, ISourceEventProcessor,
+public abstract class TunerController implements Tunable, ISourceEventProcessor, ISourceEventListener,
     IComplexBufferProvider, Listener<ComplexBuffer>
 {
     private final static Logger mLog = LoggerFactory.getLogger(TunerController.class);
@@ -48,6 +52,7 @@ public abstract class TunerController implements Tunable, ISourceEventProcessor,
     protected FrequencyController mFrequencyController;
     private int mMiddleUnusable;
     private double mUsableBandwidthPercentage;
+    private Listener<SourceEvent> mSourceEventListener;
 
     /**
      * Abstract tuner controller class.  The tuner controller manages frequency bandwidth and currently tuned channels
@@ -65,6 +70,16 @@ public abstract class TunerController implements Tunable, ISourceEventProcessor,
         mFrequencyController = new FrequencyController(this, minimumFrequency, maximumFrequency, 0.0d);
         mMiddleUnusable = middleUnusable;
         mUsableBandwidthPercentage = usableBandwidth;
+        mSourceEventListener = new SourceEventListenerToProcessorAdapter(this);
+    }
+
+    /**
+     * Implements the ISourceEventListener interface to receive requests from sample consumers
+     */
+    @Override
+    public Listener<SourceEvent> getSourceEventListener()
+    {
+        return mSourceEventListener;
     }
 
     /**
@@ -76,12 +91,29 @@ public abstract class TunerController implements Tunable, ISourceEventProcessor,
      * Responds to requests to set the frequency
      */
     @Override
-	public void process(SourceEvent event ) throws SourceException
+	public void process(SourceEvent sourceEvent ) throws SourceException
 	{
-    	if( event.getEvent() == Event.REQUEST_FREQUENCY_CHANGE )
+	    switch(sourceEvent.getEvent())
-    	{
+        {
-			setFrequency( event.getValue().longValue() );
+            case REQUEST_FREQUENCY_CHANGE:
-    	}
+                setFrequency( sourceEvent.getValue().longValue() );
+                break;
+            case REQUEST_START_SAMPLE_STREAM:
+                if(sourceEvent.getSource() instanceof Listener)
+                {
+                    addComplexBufferListener((Listener<ComplexBuffer>)sourceEvent.getSource());
+                }
+                break;
+            case REQUEST_STOP_SAMPLE_STREAM:
+                if(sourceEvent.getSource() instanceof Listener)
+                {
+                    removeComplexBufferListener((Listener<ComplexBuffer>)sourceEvent.getSource());
+                }
+                break;
+            default:
+                mLog.error("Ignoring unrecognized source event: " + sourceEvent.getEvent().name() + " from [" +
+                    sourceEvent.getSource().getClass() + "]" );
+        }
 	}
 
     public int getBandwidth()
@@ -116,7 +148,7 @@ public abstract class TunerController implements Tunable, ISourceEventProcessor,
         return mFrequencyController.canTune(frequency);
     }
 
-    public int getSampleRate()
+    public double getSampleRate()
     {
         return mFrequencyController.getSampleRate();
     }
@@ -224,7 +256,7 @@ public abstract class TunerController implements Tunable, ISourceEventProcessor,
                     updateLOFrequency();
                 }
 
-                source = new TunerChannelSource(tuner, channel);
+                source = new CICTunerChannelSource(tuner, channel);
             }
             catch(SourceException se)
             {

src/source/tuner/TunerModel.java

@@ -220,7 +220,7 @@ public class TunerModel extends AbstractTableModel implements Listener<TunerEven
                 case TUNER_ID:
                     return tuner.getUniqueID();
                 case SAMPLE_RATE:
-                    int sampleRate = tuner.getTunerController().getSampleRate();
+                    double sampleRate = tuner.getTunerController().getSampleRate();
 
                     return mSampleRateFormat.format(sampleRate / 1E6D) + MHZ;
                 case FREQUENCY:

src/source/tuner/airspy/AirspyTunerController.java

@@ -375,7 +375,7 @@ public class AirspyTunerController extends USBTunerController
     }
 
     @Override
-    public int getCurrentSampleRate() throws SourceException
+    public double getCurrentSampleRate()
     {
         return mSampleRate;
     }

src/source/tuner/channel/CICTunerChannelSource.java

@@ -0,0 +1,225 @@
+/*******************************************************************************
+ * sdrtrunk
+ * Copyright (C) 2014-2017 Dennis Sheirer
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>
+ *
+ ******************************************************************************/
+package source.tuner.channel;
+
+import dsp.filter.FilterFactory;
+import dsp.filter.Window;
+import dsp.filter.cic.ComplexPrimeCICDecimate;
+import dsp.mixer.Oscillator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sample.Buffer;
+import sample.Listener;
+import sample.OverflowableTransferQueue;
+import sample.complex.Complex;
+import sample.complex.ComplexBuffer;
+import source.SourceEvent;
+import source.tuner.Tuner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class CICTunerChannelSource extends TunerChannelSource
+{
+    private final static Logger mLog = LoggerFactory.getLogger(CICTunerChannelSource.class);
+
+    //Maximum number of filled buffers for the blocking queue
+    private static final int BUFFER_MAX_CAPACITY = 300;
+
+    //Threshold for resetting buffer overflow condition
+    private static final int BUFFER_OVERFLOW_RESET_THRESHOLD = 100;
+
+    private static double CHANNEL_RATE = 48000.0;
+    private static int CHANNEL_PASS_FREQUENCY = 12000;
+
+    private OverflowableTransferQueue<ComplexBuffer> mBuffer;
+
+    private Oscillator mFrequencyCorrectionMixer;
+    private ComplexPrimeCICDecimate mDecimationFilter;
+    private Listener<ComplexBuffer> mListener;
+    private List<ComplexBuffer> mSampleBuffers = new ArrayList<ComplexBuffer>();
+    private long mTunerFrequency = 0;
+    private double mTunerSampleRate;
+    private long mChannelFrequencyCorrection = 0;
+
+    /**
+     * Implements a heterodyne/decimate Digital Drop Channel (DDC) to decimate the IQ output from a tuner down to a
+     * fixed 48 kHz IQ channel rate.
+     *
+     * Note: this class can only be used once (started and stopped) and a new tuner channel source must be requested
+     * from the tuner once this object has been stopped.  This is because channels are managed dynamically and center
+     * tuned frequency may have changed since this source was obtained and thus the tuner might no longer be able to
+     * source this channel once it has been stopped.
+     *
+     * @param tunerChannel specifying the center frequency for the DDC
+     */
+    public CICTunerChannelSource(Tuner tuner, TunerChannel tunerChannel)
+    {
+        super(tuner.getTunerController().getSourceEventListener(), tunerChannel);
+
+        mBuffer = new OverflowableTransferQueue<>(BUFFER_MAX_CAPACITY, BUFFER_OVERFLOW_RESET_THRESHOLD);
+        mBuffer.setSourceOverflowListener(this);
+
+        //Setup the frequency translator to the current source frequency
+        mTunerFrequency = tuner.getTunerController().getFrequency();
+        long frequencyOffset = mTunerFrequency - getTunerChannel().getFrequency();
+        mFrequencyCorrectionMixer = new Oscillator(frequencyOffset, tuner.getTunerController().getSampleRate());
+
+        //Setup the decimation filter chain
+        setSampleRate(tuner.getTunerController().getSampleRate());
+    }
+
+    /**
+     * Sets the center frequency for the incoming sample buffers
+     * @param frequency in hertz
+     */
+    @Override
+    protected void setFrequency(long frequency)
+    {
+        mTunerFrequency = frequency;
+
+        //Reset frequency correction so that consumer components can adjust
+        setFrequencyCorrection(0);
+
+        updateMixerFrequencyOffset();
+    }
+
+    /**
+     * Current frequency correction being applied to the channel sample stream
+     * @return correction in hertz
+     */
+    @Override
+    public long getFrequencyCorrection()
+    {
+        return mChannelFrequencyCorrection;
+    }
+
+    /**
+     * Changes the frequency correction value and broadcasts the change to the registered downstream listener.
+     * @param correction current frequency correction value.
+     */
+    protected void setFrequencyCorrection(long correction)
+    {
+        mChannelFrequencyCorrection = correction;
+
+        updateMixerFrequencyOffset();
+
+        broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mChannelFrequencyCorrection));
+    }
+
+    /**
+     * Primary input method for receiving complex sample buffers from the wideband source (ie tuner)
+     * @param buffer to receive and eventually process
+     */
+    public void receive(ComplexBuffer buffer)
+    {
+        mBuffer.offer(buffer);
+    }
+
+    @Override
+    public void setListener(Listener<ComplexBuffer> listener)
+    {
+		/* Save a pointer to the listener so that if we have to change the
+		 * decimation filter, we can re-add the listener */
+        mListener = listener;
+
+        mDecimationFilter.setListener(listener);
+    }
+
+    @Override
+    public void removeListener(Listener<ComplexBuffer> listener)
+    {
+        mDecimationFilter.removeListener();
+    }
+
+    /**
+     * Updates the sample rate to the requested value and notifies any downstream components of the change
+     * @param sampleRate to set
+     */
+    protected void setSampleRate(double sampleRate)
+    {
+        if(mTunerSampleRate != sampleRate)
+        {
+            mFrequencyCorrectionMixer.setSampleRate(sampleRate);
+
+            /* Get new decimation filter */
+            mDecimationFilter = FilterFactory.getDecimationFilter((int)sampleRate, (int)CHANNEL_RATE, 1,
+                CHANNEL_PASS_FREQUENCY, 60, Window.WindowType.HAMMING);
+
+            /* re-add the original output listener */
+            mDecimationFilter.setListener(mListener);
+
+            mTunerSampleRate = sampleRate;
+
+            broadcastConsumerSourceEvent(SourceEvent.channelSampleRateChange(getSampleRate()));
+        }
+    }
+
+    /**
+     * Calculates the local mixer frequency offset from the tuned frequency,
+     * channel's requested frequency, and channel frequency correction.
+     */
+    private void updateMixerFrequencyOffset()
+    {
+        long offset = mTunerFrequency - getTunerChannel().getFrequency() - mChannelFrequencyCorrection;
+        mFrequencyCorrectionMixer.setFrequency(offset);
+    }
+
+    public double getSampleRate()
+    {
+        return CHANNEL_RATE;
+    }
+
+    @Override
+    protected void processSamples()
+    {
+        mBuffer.drainTo(mSampleBuffers, 20);
+
+        for(Buffer buffer : mSampleBuffers)
+        {
+            float[] samples = buffer.getSamples();
+
+            /* We make a copy of the buffer so that we don't affect
+             * anyone else that is using the same buffer, like other
+             * channels or the spectral display */
+            float[] translated = new float[samples.length];
+
+            /* Perform frequency translation */
+            for(int x = 0; x < samples.length; x += 2)
+            {
+                mFrequencyCorrectionMixer.rotate();
+
+                translated[x] = Complex.multiplyInphase(
+                    samples[x], samples[x + 1], mFrequencyCorrectionMixer.inphase(), mFrequencyCorrectionMixer.quadrature());
+
+                translated[x + 1] = Complex.multiplyQuadrature(
+                    samples[x], samples[x + 1], mFrequencyCorrectionMixer.inphase(), mFrequencyCorrectionMixer.quadrature());
+            }
+
+            final ComplexPrimeCICDecimate filter = mDecimationFilter;
+
+            if(filter != null)
+            {
+                filter.receive(new ComplexBuffer(translated));
+            }
+        }
+
+        mSampleBuffers.clear();
+    }
+}

src/source/tuner/channel/DecimatingTunerChannelSource.java

@@ -1,541 +0,0 @@
-/*******************************************************************************
- * sdrtrunk
- * Copyright (C) 2014-2017 Dennis Sheirer
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>
- *
- ******************************************************************************/
-package source.tuner.channel;
-
-import dsp.filter.FilterFactory;
-import dsp.filter.Window;
-import dsp.filter.cic.ComplexPrimeCICDecimate;
-import dsp.mixer.Oscillator;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import sample.Buffer;
-import sample.Listener;
-import sample.OverflowableTransferQueue;
-import sample.complex.Complex;
-import sample.complex.ComplexBuffer;
-import sample.real.IOverflowListener;
-import source.ISourceEventListener;
-import source.ISourceEventProcessor;
-import source.ISourceEventProvider;
-import source.SourceEvent;
-import source.SourceException;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-public class DecimatingTunerChannelSource extends TunerChannelSource
-{
-    private final static Logger mLog = LoggerFactory.getLogger(DecimatingTunerChannelSource.class);
-
-    //Maximum number of filled buffers for the blocking queue
-    private static final int BUFFER_MAX_CAPACITY = 300;
-
-    //Threshold for resetting buffer overflow condition
-    private static final int BUFFER_OVERFLOW_RESET_THRESHOLD = 100;
-
-    private static int CHANNEL_RATE = 48000;
-    private static int CHANNEL_PASS_FREQUENCY = 12000;
-
-    private OverflowableTransferQueue<ComplexBuffer> mBuffer;
-
-    private TunerChannel mTunerChannel;
-    private Oscillator mMixer;
-    private ComplexPrimeCICDecimate mDecimationFilter;
-    private Listener<ComplexBuffer> mListener;
-    private ISourceEventProcessor mFrequencyChangeProcessor;
-    private DownstreamProcessor mDownstreamFrequencyEventProcessor = new DownstreamProcessor();
-    private ScheduledFuture<?> mTaskHandle;
-
-    private long mTunerFrequency = 0;
-    private int mTunerSampleRate;
-    private int mChannelFrequencyCorrection = 0;
-
-    private DecimationProcessor mDecimationProcessor = new DecimationProcessor();
-
-    private AtomicBoolean mRunning = new AtomicBoolean();
-    private boolean mExpended = false;
-
-    /**
-     * Provides a Digital Drop Channel (DDC) to decimate the IQ output from a
-     * tuner down to a 48 kHz IQ channel rate.
-     *
-     * Note: this class can only be used once (started and stopped) and a new
-     * tuner channel source must be requested from the tuner once this object
-     * has been stopped.  This is because channels are managed dynamically and
-     * center tuned frequency may have changed since this source was obtained
-     * and thus the tuner might no longer be able to source this channel once it
-     * has been stopped.
-     *
-     * @param tunerChannel specifying the center frequency for the DDC
-     * @throws RejectedExecutionException if the thread pool manager cannot
-     *                                    accept the decimation processing task
-     * @throws SourceException            if the tuner has an issue providing IQ samples
-     */
-    public DecimatingTunerChannelSource(Listener<SourceEvent> listener, TunerChannel tunerChannel, int inputSampleRate,
-                                        long frequency)
-    {
-        super(listener, tunerChannel);
-
-        mTunerChannel = tunerChannel;
-        mTunerFrequency = frequency;
-
-        mBuffer = new OverflowableTransferQueue<>(BUFFER_MAX_CAPACITY, BUFFER_OVERFLOW_RESET_THRESHOLD);
-
-	    /* Setup the frequency translator to the current source frequency */
-        long frequencyOffset = mTunerFrequency - mTunerChannel.getFrequency();
-
-        mMixer = new Oscillator(frequencyOffset, inputSampleRate);
-
-		/* Fire a sample rate change event to setup the decimation chain */
-		try
-        {
-            process(SourceEvent.sampleRateChange(inputSampleRate));
-        }
-        catch(SourceException se)
-        {
-            mLog.error("Error", se);
-        }
-
-    }
-
-    /**
-     * Changes the frequency correction value and broadcasts the change to the registered downstream listener.
-     * @param correction current frequency correction value.
-     */
-    private void setFrequencyCorrection(int correction)
-    {
-        mChannelFrequencyCorrection = correction;
-
-        updateMixerFrequencyOffset();
-
-        mDownstreamFrequencyEventProcessor.broadcast(
-            SourceEvent.channelFrequencyCorrectionChange(mChannelFrequencyCorrection));
-    }
-
-    /**
-     * Overrides the default source overflow listener management to delegate responsibility to the overflow buffer
-     */
-    @Override
-    public void setOverflowListener(IOverflowListener listener)
-    {
-        mBuffer.setOverflowListener(listener);
-    }
-
-    public void start(ScheduledExecutorService executor)
-    {
-        if(mExpended)
-        {
-            throw new IllegalStateException("Attempt to re-start an expended tuner channel source.  TunerChannelSource" +
-                " objects can only be used once. ");
-        }
-
-        if(mRunning.compareAndSet(false, true))
-        {
-            //Broadcast current frequency and sample rate settings so all downstream components are aware
-            mDownstreamFrequencyEventProcessor.broadcastCurrentFrequency();
-            mDownstreamFrequencyEventProcessor.broadcastCurrentSampleRate();
-
-            //Schedule the decimation task to run every 9 ms (111 iterations/second), an odd periodicity relative
-            //to the inbound periodicity of 20 ms, to attempt to avoid thread queue contention
-            mTaskHandle = executor.scheduleAtFixedRate(mDecimationProcessor, 0, 9, TimeUnit.MILLISECONDS);
-
-		    /* Finally, register to receive samples from the tuner */
-		    mSourceEventListener.receive(SourceEvent.startSampleStream(this));
-        }
-        else
-        {
-            mLog.warn("Attempt to start() an already running tuner channel source was ignored");
-        }
-    }
-
-    @Override
-    public void reset()
-    {
-    }
-
-    @Override
-    public void stop()
-    {
-        if(mRunning.compareAndSet(true, false))
-        {
-            mSourceEventListener.receive(SourceEvent.stopSampleStream(this));
-            mDecimationProcessor.shutdown();
-
-            if(mTaskHandle != null)
-            {
-                mTaskHandle.cancel(true);
-                mTaskHandle = null;
-            }
-
-            mBuffer.clear();
-
-            mExpended = true;
-        }
-        else
-        {
-            mLog.warn("Attempt to stop() an already stopped tuner channel source was ignored");
-        }
-    }
-
-
-    @Override
-    public void dispose()
-    {
-    }
-
-    public TunerChannel getTunerChannel()
-    {
-        return mTunerChannel;
-    }
-
-    @Override
-    public void receive(ComplexBuffer buffer)
-    {
-        if(mRunning.get())
-        {
-            mBuffer.offer(buffer);
-        }
-    }
-
-    public void setFrequencyChangeListener(ISourceEventProcessor processor)
-    {
-        mFrequencyChangeProcessor = processor;
-    }
-
-    @Override
-    public void setListener(Listener<ComplexBuffer> listener)
-    {
-		/* Save a pointer to the listener so that if we have to change the
-		 * decimation filter, we can re-add the listener */
-        mListener = listener;
-
-        mDecimationFilter.setListener(listener);
-    }
-
-    @Override
-    public void removeListener(Listener<ComplexBuffer> listener)
-    {
-        mDecimationFilter.removeListener();
-    }
-
-    /**
-     * Handler for frequency change events received from the tuner and channel
-     * frequency correction events received from the channel consumer/listener
-     */
-    @Override
-    public void process(SourceEvent event) throws SourceException
-    {
-        // Echo the event to the registered event listener
-        if(mFrequencyChangeProcessor != null)
-        {
-            mFrequencyChangeProcessor.process(event);
-        }
-
-        switch(event.getEvent())
-        {
-            case NOTIFICATION_FREQUENCY_CHANGE:
-                mTunerFrequency = event.getValue().longValue();
-                updateMixerFrequencyOffset();
-
-                //Reset frequency correction so that downstream components can recalculate the value
-                setFrequencyCorrection(0);
-                break;
-            case NOTIFICATION_SAMPLE_RATE_CHANGE:
-                int sampleRate = event.getValue().intValue();
-                setSampleRate(sampleRate);
-                break;
-            default:
-                break;
-        }
-    }
-
-    /**
-     * Updates the sample rate to the requested value and notifies any downstream components of the change
-     * @param sampleRate to set
-     */
-    private void setSampleRate(int sampleRate)
-    {
-        if(mTunerSampleRate != sampleRate)
-        {
-            mMixer.setSampleRate(sampleRate);
-
-            /* Get new decimation filter */
-            mDecimationFilter = FilterFactory.getDecimationFilter(sampleRate, CHANNEL_RATE, 1,
-                CHANNEL_PASS_FREQUENCY, 60, Window.WindowType.HAMMING);
-
-            /* re-add the original output listener */
-            mDecimationFilter.setListener(mListener);
-
-            mTunerSampleRate = sampleRate;
-
-            mDownstreamFrequencyEventProcessor.broadcastCurrentSampleRate();
-        }
-    }
-
-    /**
-     * Calculates the local mixer frequency offset from the tuned frequency,
-     * channel's requested frequency, and channel frequency correction.
-     */
-    private void updateMixerFrequencyOffset()
-    {
-        long offset = mTunerFrequency - mTunerChannel.getFrequency() - mChannelFrequencyCorrection;
-        mMixer.setFrequency(offset);
-    }
-
-    public int getSampleRate() throws SourceException
-    {
-        return CHANNEL_RATE;
-    }
-
-    public long getFrequency() throws SourceException
-    {
-        return mTunerChannel.getFrequency();
-    }
-
-    /**
-     * Implements ISourceEventProvider to enable this source to broadcast frequency change events to downstream
-     * listeners.
-     *
-     * @param listener to receive downstream events
-     */
-    @Override
-    public void setSourceEventListener(Listener<SourceEvent> listener)
-    {
-        mDownstreamFrequencyEventProcessor.setSourceEventListener(listener);
-    }
-
-    /**
-     * Implements ISourceEventProvider to remove the frequency change listener from receiving down-stream frequency
-     * change events.
-     */
-    @Override
-    public void removeSourceEventListener()
-    {
-        mDownstreamFrequencyEventProcessor.removeSourceEventListener();
-    }
-
-    /**
-     * Implements ISourceEventListener to receive frequency change events containing requests from downstream
-     * listeners to change frequency values.
-     * @return listener
-     */
-    @Override
-    public Listener<SourceEvent> getSourceEventListener()
-    {
-        return mDownstreamFrequencyEventProcessor.getSourceEventListener();
-    }
-
-    /**
-     * Managers frequency change requests and notifications from/to any downstream component.  Downstream
-     * components are those that receive samples from this tuner channel source.  These downstream components will be
-     * notified of any frequency or sample rate change events and will also be able to request frequency correction
-     * updates.
-     */
-    public class DownstreamProcessor implements ISourceEventListener, ISourceEventProvider,
-        Listener<SourceEvent>
-    {
-        //Listener to receive downstream events
-        private Listener<SourceEvent> mListener;
-
-        /**
-         * Broadcasts the frequency change event to the downstream frequency change listener
-         * @param event to broadcast
-         */
-        public void broadcast(SourceEvent event)
-        {
-            if(mListener != null)
-            {
-                mListener.receive(event);
-            }
-        }
-
-        /**
-         * Broadcasts the current frequency of this tuner channel source to the downstream listener
-         */
-        public void broadcastCurrentFrequency()
-        {
-            try
-            {
-                long frequency = getFrequency();
-                broadcast(SourceEvent.frequencyChange(frequency));
-            }
-            catch(SourceException se)
-            {
-                mLog.error("Error obtaining frequency from tuner to broadcast downstream");
-            }
-        }
-
-        /**
-         * Broadcasts the current decimated sample rate of this tuner channel source
-         */
-        public void broadcastCurrentSampleRate()
-        {
-            try
-            {
-                //Note: downstream sample rate is currently a fixed value -- it will change in the future
-                broadcast(SourceEvent.sampleRateChange(getSampleRate()));
-            }
-            catch(SourceException se)
-            {
-                mLog.error("Error obtaining sample rate from tuner to broadcast downstream");
-            }
-        }
-
-        /**
-         * Sets the downstream listener to receive frequency change events from this tuner channel source
-         * @param listener to receive events
-         */
-        @Override
-        public void setSourceEventListener(Listener<SourceEvent> listener)
-        {
-            mListener = listener;
-        }
-
-        /**
-         * Removes the downstream listener from receiving frequency change events.
-         */
-        @Override
-        public void removeSourceEventListener()
-        {
-            mListener = null;
-        }
-
-        /**
-         * Listener for receiving frequency change events from downstream components
-         */
-        @Override
-        public Listener<SourceEvent> getSourceEventListener()
-        {
-            return this;
-        }
-
-        /**
-         * Processes frequency change events from downstream components.
-         * @param event to process
-         */
-        @Override
-        public void receive(SourceEvent event)
-        {
-            switch(event.getEvent())
-            {
-                //Frequency correction requests are the only change requests supported from downstream components
-                case REQUEST_CHANNEL_FREQUENCY_CORRECTION_CHANGE:
-                    setFrequencyCorrection(event.getValue().intValue());
-                    break;
-            }
-        }
-    }
-
-    /**
-     * Decimates an inbound buffer of I/Q samples from the source down to the
-     * standard 48000 channel sample rate
-     */
-    public class DecimationProcessor implements Runnable
-    {
-        private boolean mProcessing = true;
-        private List<ComplexBuffer> mSampleBuffers = new ArrayList<ComplexBuffer>();
-
-        public void shutdown()
-        {
-            mProcessing = false;
-        }
-
-        @Override
-        public void run()
-        {
-			/* General exception handler so that any errors won't kill the
-			 * decimation thread and cause the input buffers to fill up and
-			 * run the program out of memory */
-            try
-            {
-                if(mProcessing)
-                {
-                    //Send a heartbeat every time this runs to allow downstream components to perform periodic
-                    //state monitoring functions on this thread
-                    getHeartbeatManager().broadcast();
-
-                    mBuffer.drainTo(mSampleBuffers, 20);
-
-                    for(Buffer buffer : mSampleBuffers)
-                    {
-							/* Check to see if we've been shutdown */
-                        if(!mProcessing)
-                        {
-                            mBuffer.clear();
-                            return;
-                        }
-                        else
-                        {
-                            float[] samples = buffer.getSamples();
-
-								/* We make a copy of the buffer so that we don't affect
-								 * anyone else that is using the same buffer, like other
-								 * channels or the spectral display */
-                            float[] translated = new float[samples.length];
-
-								/* Perform frequency translation */
-                            for(int x = 0; x < samples.length; x += 2)
-                            {
-                                mMixer.rotate();
-
-                                translated[x] = Complex.multiplyInphase(
-                                    samples[x], samples[x + 1], mMixer.inphase(), mMixer.quadrature());
-
-                                translated[x + 1] = Complex.multiplyQuadrature(
-                                    samples[x], samples[x + 1], mMixer.inphase(), mMixer.quadrature());
-                            }
-
-                            if(mProcessing)
-                            {
-                                final ComplexPrimeCICDecimate filter = mDecimationFilter;
-                                filter.receive(new ComplexBuffer(translated));
-                            }
-                        }
-                    }
-
-                    mSampleBuffers.clear();
-                }
-            }
-            catch(Exception e)
-            {
-				/* Only log the stack trace if we're still processing */
-                if(mProcessing)
-                {
-                    mLog.error("Error encountered during decimation process", e);
-                }
-            }
-            catch(Throwable throwable)
-            {
-                mLog.error("Code error encountered during decimation process - channel thread will probably die", throwable);
-            }
-
-			/* Check to see if we've been shutdown */
-            if(!mProcessing)
-            {
-                mBuffer.clear();
-                mSampleBuffers.clear();
-            }
-        }
-    }
-}

src/source/tuner/channel/TunerChannelSource.java

@@ -18,51 +18,85 @@
  ******************************************************************************/
 package source.tuner.channel;
 
-import dsp.filter.FilterFactory;
-import dsp.filter.Window.WindowType;
-import dsp.filter.cic.ComplexPrimeCICDecimate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import sample.Buffer;
 import sample.Listener;
-import sample.complex.Complex;
 import sample.complex.ComplexBuffer;
-import sample.real.IOverflowListener;
 import source.ComplexSource;
-import source.SourceException;
-import source.ISourceEventListener;
-import source.ISourceEventProvider;
-import source.SourceEvent;
 import source.ISourceEventProcessor;
+import source.SourceEvent;
+import source.SourceEventListenerToProcessorAdapter;
+import source.SourceException;
+import util.ThreadPool;
 
-import java.util.ArrayList;
+import java.util.concurrent.ScheduledFuture;
-import java.util.List;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-public abstract class TunerChannelSource extends ComplexSource implements ISourceEventProcessor, Listener<ComplexBuffer>
+public abstract class TunerChannelSource extends ComplexSource implements ISourceEventProcessor
 {
     private final static Logger mLog = LoggerFactory.getLogger(TunerChannelSource.class);
+    private static final long BUFFER_PROCESSOR_RUN_INTERVAL_MILLISECONDS = 50;
+    private SourceEventListenerToProcessorAdapter mConsumerSourceEventListenerAdapter;
     private TunerChannel mTunerChannel;
-    protected Listener<SourceEvent> mSourceEventListener;
+    private Listener<SourceEvent> mProducerSourceEventListener;
+    private Listener<SourceEvent> mConsumerSourceEventListener;
+    private BufferProcessor mBufferProcessor = new BufferProcessor();
 
     /**
-     * Requirements:
+     * Tuner Channel Source is a Digital Drop Channel (DDC) abstract class that defines the minimum functionality
-     * 1. Provide complex buffer samples
+     * requirements for a channel sample source.
-     * 2. Fetch complex buffer samples from subclass implementations (for distribution)
-     * 3. Provide heartbeat at each interval
-     * 4. Provide start/stop running
-     * 5. Provide support for downstream consumer source events (delegate to the sub class)
-     * 6. Support buffer overflow listeners
      *
-     * @param sourceEventListener to receive source dispose request
+     * @param producerSourceEventListener to receive source event requests (e.g. start/stop sample stream)
-     * @param tunerChannel
+     * @param tunerChannel describing the desired channel frequency and bandwidth/minimum sample rate
      */
-    public TunerChannelSource(Listener<SourceEvent> sourceEventListener, TunerChannel tunerChannel)
+    public TunerChannelSource(Listener<SourceEvent> producerSourceEventListener, TunerChannel tunerChannel)
     {
-        mSourceEventListener = sourceEventListener;
+        mProducerSourceEventListener = producerSourceEventListener;
+        mTunerChannel = tunerChannel;
+        mConsumerSourceEventListenerAdapter = new SourceEventListenerToProcessorAdapter(this);
     }
 
+    @Override
+    public long getFrequency()
+    {
+        return mTunerChannel.getFrequency();
+    }
+
+    /**
+     * Sets the center frequency for the sample streaming being sent from the producer.
+     * @param frequency in hertz
+     */
+    protected abstract void setFrequency(long frequency);
+
+    /**
+     * Sets the sample rate of the incoming sample stream from the producer
+     * @param sampleRate in hertz
+     */
+    protected abstract void setSampleRate(double sampleRate);
+
+    /**
+     * Sets the frequency correction for the outbound sample stream to the consumer
+     * @param correction in hertz
+     */
+    protected abstract void setFrequencyCorrection(long correction);
+
+    /**
+     * Frequency correction value currently being applied to the outbound sample stream
+     */
+    public abstract long getFrequencyCorrection();
+
+    /**
+     * Sets the listener to receive the complex buffer sample output from this channel
+     * @param complexBufferListener to receive complex buffers
+     */
+    public abstract void setListener(Listener<ComplexBuffer> complexBufferListener);
+
+    /**
+     * Commands sub-class to process queued samples and distribute them to the consumer.  This method will be invoked
+     * by an interval timer.
+     */
+    protected abstract void processSamples();
+
     /**
      * Tuner channel for this tuner channel source
      */
@@ -71,17 +105,27 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
         return mTunerChannel;
     }
 
+    @Override
+    public void dispose()
+    {
+        stop();
+    }
+
     /**
      * Starts this tuner channel source producing sample stream.
      */
     public void start()
     {
-        if(mSourceEventListener != null)
+        //Broadcast current frequency and sample rate so consumer can configure correctly
+        broadcastConsumerSourceEvent(SourceEvent.frequencyChange(getFrequency()));
+        broadcastConsumerSourceEvent(SourceEvent.sampleRateChange(getSampleRate()));
+
+        if(mProducerSourceEventListener != null)
         {
-            mSourceEventListener.receive(SourceEvent.startSampleStream(this));
+            mProducerSourceEventListener.receive(SourceEvent.startSampleStream(this));
         }
 
-        //TODO: broadcast center frequency and sample rate to downstream listeners.
+        mBufferProcessor.start();
     }
 
     /**
@@ -90,11 +134,13 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
      */
     public void stop()
     {
-        if(mSourceEventListener != null)
+        if(mProducerSourceEventListener != null)
         {
-            mSourceEventListener.receive(SourceEvent.stopSampleStream(this));
+            mProducerSourceEventListener.receive(SourceEvent.stopSampleStream(this));
-            mSourceEventListener.receive(SourceEvent.sourceDisposeRequest(this));
+            mProducerSourceEventListener.receive(SourceEvent.sourceDisposeRequest(this));
         }
+
+        mBufferProcessor.stop();
     }
 
     @Override
@@ -102,4 +148,159 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
     {
         //Reset is not valid for a tuner channel source - ignored
     }
+
+    /**
+     * This method is invoked after the buffer processor is completely shutdown so that this instance can perform any
+     * cleanup operations needed to dispose of this instance.
+     */
+    protected void performDisposal()
+    {
+        mProducerSourceEventListener = null;
+        mConsumerSourceEventListener = null;
+        mConsumerSourceEventListenerAdapter.dispose();
+    }
+
+    /**
+     * Processes source event notifications received from the producer/parent and source event requests received from
+     * the consumer.
+     *
+     * @param sourceEvent to process
+     * @throws SourceException if there is an error while processing the source event
+     */
+    @Override
+    public void process(SourceEvent sourceEvent) throws SourceException
+    {
+        switch(sourceEvent.getEvent())
+        {
+            //Notification events from the producer
+            case NOTIFICATION_FREQUENCY_CHANGE:
+                setFrequency(sourceEvent.getValue().longValue());
+                break;
+            case NOTIFICATION_SAMPLE_RATE_CHANGE:
+                setSampleRate(sourceEvent.getValue().doubleValue());
+                break;
+
+            //Request events from the consumer
+            case REQUEST_CHANNEL_FREQUENCY_CORRECTION_CHANGE:
+                setFrequencyCorrection(sourceEvent.getValue().longValue());
+                break;
+
+            default:
+                mLog.error("Ignoring unrecognized source event: " + sourceEvent.getEvent() + " from [" +
+                    sourceEvent.getSource().getClass() + "]");
+                break;
+        }
+    }
+
+    /**
+     * Broadcasts the source event to the incoming sample stream producer.
+     * @param sourceEvent to broadcast
+     */
+    protected void broadcastProducerSourceEvent(SourceEvent sourceEvent)
+    {
+        if(mProducerSourceEventListener != null)
+        {
+            mProducerSourceEventListener.receive(sourceEvent);
+        }
+    }
+
+    /**
+     * Broadcasts the source event to the outbound sample stream consumer.
+     * @param sourceEvent to broadcast
+     */
+    protected void broadcastConsumerSourceEvent(SourceEvent sourceEvent)
+    {
+        if(mConsumerSourceEventListener != null)
+        {
+            mConsumerSourceEventListener.receive(sourceEvent);
+        }
+    }
+
+    /**
+     * Implements the ISourceEventListener interface for receiving source events (ie requests) from a consumer.
+     *
+     * Note: source events are simply transferred to the process() method in this class.
+     */
+    @Override
+    public Listener<SourceEvent> getSourceEventListener()
+    {
+        return mConsumerSourceEventListenerAdapter;
+    }
+
+    /**
+     * Implements the ISourceEventProvider interface to set the consumer listener for source events.
+     * @param listener to receive consumer source events
+     */
+    @Override
+    public void setSourceEventListener(Listener<SourceEvent> listener)
+    {
+        mConsumerSourceEventListener = listener;
+    }
+
+    /**
+     * Implements the ISourceEventProvider interface to remove the consumer listener for source events.
+     */
+    @Override
+    public void removeSourceEventListener()
+    {
+        mConsumerSourceEventListener = null;
+    }
+
+    /**
+     * Processor to periodically invoke buffer sample processing on an interval timer.  At each interval this processor
+     * sends a heartbeat to the registered consumer and then commands the sub-class implementation to process any
+     * queued buffers and distribute complex buffer sample(s) to the registered consumer.
+     */
+    public class BufferProcessor implements Runnable
+    {
+        private ScheduledFuture<?> mScheduledFuture;
+        private boolean mStopped = false;
+
+        /**
+         * Commands this processor to do a shutdown at the end of this or the next iteration.  Once successfully
+         * shutdown, it will invoke the performDisposal() method to cleanup this instance.
+         */
+        public void stop()
+        {
+            mStopped = true;
+        }
+
+        /**
+         * Starts buffer processing on a fixed interval using a scheduled thread pool
+         */
+        public void start()
+        {
+            if(mScheduledFuture == null)
+            {
+                mScheduledFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(this, 0,
+                    BUFFER_PROCESSOR_RUN_INTERVAL_MILLISECONDS, TimeUnit.MILLISECONDS);
+            }
+        }
+
+        /**
+         * Implementation of the Runnable interface to periodically send a heartbeat and then process buffer samples.
+         */
+        @Override
+        public void run()
+        {
+            if(!mStopped)
+            {
+                getHeartbeatManager().broadcast();
+                processSamples();
+            }
+
+            if(mStopped)
+            {
+                if(mScheduledFuture != null)
+                {
+                    //Set may-interrupt to false so that we can complete this iteration
+                    mScheduledFuture.cancel(false);
+                }
+
+                mScheduledFuture = null;
+
+                performDisposal();
+            }
+        }
+    }
 }

src/source/tuner/fcd/FCDTunerController.java

@@ -174,7 +174,7 @@ public abstract class FCDTunerController extends TunerController
     /**
      * Sample rate of the tuner
      */
-    public abstract int getCurrentSampleRate() throws SourceException;
+    public abstract double getCurrentSampleRate();
 
     /**
      * Tuner class

src/source/tuner/fcd/proV1/FCD1TunerController.java

@@ -85,7 +85,7 @@ public class FCD1TunerController extends FCDTunerController
         }
     }
 
-    public int getCurrentSampleRate()
+    public double getCurrentSampleRate()
     {
         return SAMPLE_RATE;
     }

src/source/tuner/fcd/proplusV2/FCD2TunerController.java

@@ -64,7 +64,7 @@ public class FCD2TunerController extends FCDTunerController
         }
     }
 
-    public int getCurrentSampleRate()
+    public double getCurrentSampleRate()
     {
         return SAMPLE_RATE;
     }

src/source/tuner/frequency/FrequencyController.java

@@ -36,8 +36,8 @@ public class FrequencyController
     private long mTunedFrequency = 101100000;
     private long mMinimumFrequency;
     private long mMaximumFrequency;
-    private double mFrequencyCorrection = 0d;
+    private double mFrequencyCorrection = 0.0d;
-    private int mSampleRate = 0;
+    private double mSampleRate = 0.0d;
 
     private ArrayList<ISourceEventProcessor> mProcessors = new ArrayList<>();
 
@@ -54,7 +54,7 @@ public class FrequencyController
      */
     public int getBandwidth()
     {
-        return mSampleRate;
+        return (int)mSampleRate;
     }
 
     /**
@@ -70,7 +70,7 @@ public class FrequencyController
     /**
      * Get sample rate in hertz
      */
-    public int getSampleRate()
+    public double getSampleRate()
     {
         return mSampleRate;
     }
@@ -266,7 +266,7 @@ public class FrequencyController
         /**
          * Gets the current bandwidth setting of the device
          */
-        public int getCurrentSampleRate() throws SourceException;
+        public double getCurrentSampleRate() throws SourceException;
 
         /**
          * Indicates if this tunable can tune the frequency

src/source/tuner/hackrf/HackRFTunerController.java

@@ -337,7 +337,7 @@ public class HackRFTunerController extends USBTunerController
     }
 
     @Override
-    public int getCurrentSampleRate() throws SourceException
+    public double getCurrentSampleRate()
     {
         return mSampleRate.getRate();
     }
@@ -531,7 +531,7 @@ public class HackRFTunerController extends USBTunerController
         write(Request.SET_SAMPLE_RATE, 0, 0, buffer);
     }
 
-    public int getSampleRate()
+    public double getSampleRate()
     {
         return mSampleRate.getRate();
     }

src/source/tuner/rtl/RTL2832TunerController.java

@@ -932,7 +932,7 @@ public abstract class RTL2832TunerController extends USBTunerController
         return false;
     }
 
-    public int getCurrentSampleRate() throws SourceException
+    public double getCurrentSampleRate()
     {
         return mSampleRate.getRate();
     }

src/source/tuner/test/TestTunerController.java

@@ -144,7 +144,7 @@ public class TestTunerController extends TunerController
      * Current sample rate for this tuner controller
      */
     @Override
-    public int getCurrentSampleRate() throws SourceException
+    public double getCurrentSampleRate()
     {
         return mFrequencyController.getSampleRate();
     }

src/source/wave/ComplexWaveSource.java

@@ -75,31 +75,15 @@ public class ComplexWaveSource extends ComplexSource implements IControllableFil
         return null;
     }
 
-    /**
-     * Not implemented
-     */
-    @Override
-    public void setHeartbeatListener(Listener<Heartbeat> listener)
-    {
-    }
-
-    /**
-     * Not implemented
-     */
-    @Override
-    public void removeHeartbeatListener()
-    {
-    }
-
     @Override
     public void reset()
     {
         stop();
-        start(null);
+        start();
     }
 
     @Override
-    public void start(ScheduledExecutorService executor)
+    public void start()
     {
         try
         {
@@ -132,11 +116,11 @@ public class ComplexWaveSource extends ComplexSource implements IControllableFil
     }
 
     @Override
-    public int getSampleRate()
+    public double getSampleRate()
     {
         if(mInputStream != null)
         {
-            return (int)mInputStream.getFormat().getSampleRate();
+            return mInputStream.getFormat().getSampleRate();
         }
 
         return 0;

src/source/wave/RealWaveSource.java

@@ -73,33 +73,16 @@ public class RealWaveSource extends RealSource implements IControllableFileSourc
 		return null;
 	}
 
-    /**
-     * Not implemented
-     */
-    @Override
-    public void setHeartbeatListener(Listener<Heartbeat> listener)
-    {
-    }
-
-    /**
-     * Not implemented
-     */
-    @Override
-    public void removeHeartbeatListener()
-    {
-    }
-
-
     @Override
     public void reset()
     {
         stop();
-        start(null);
+        start();
     }
 
 
     @Override
-    public void start(ScheduledExecutorService executor)
+    public void start()
     {
         try
         {
@@ -133,11 +116,11 @@ public class RealWaveSource extends RealSource implements IControllableFileSourc
     }
 
     @Override
-    public int getSampleRate()
+    public double getSampleRate()
     {
         if(mInputStream != null)
         {
-            return (int)mInputStream.getFormat().getSampleRate();
+            return mInputStream.getFormat().getSampleRate();
         }
 
         return 0;

src/spectrum/ChannelSpectrumPanel.java

@@ -90,7 +90,7 @@ public class ChannelSpectrumPanel extends JPanel
     	mDFTConverter.addListener( mSpectrumPanel );
 
     	/* Set the DFTProcessor to the decimated 24kHz sample rate */
-    	mDFTProcessor.process(SourceEvent.sampleRateChange(24000));
+    	mDFTProcessor.process(SourceEvent.sampleRateChange(24000.0));
     	
     	initGui();
     }

src/spectrum/SpectralDisplayPanel.java

@@ -447,7 +447,7 @@ public class SpectralDisplayPanel extends JPanel implements Listener<ComplexBuff
             mTuner.getTunerController().removeListener(this);
 
             //Deregister the dft processor from receiving samples
-            mTuner.removeListener((Listener<ComplexBuffer>)mDFTProcessor);
+            mTuner.getTunerController().removeComplexBufferListener(mDFTProcessor);
             mTuner = null;
         }
     }