Update version to 0.5.0 beta 2

Co-authored-by: Dennis Sheirer <dsheirer@github.com>

build.gradle

@@ -28,7 +28,7 @@ repositories {
     maven { url "https://jitpack.io" }
 }
 
-version = '0.5.0-beta1'
+version = '0.5.0-beta2'
 sourceCompatibility = '17'
 
 sourceSets {

src/main/java/io/github/dsheirer/controller/channel/ChannelProcessingManager.java

@@ -62,6 +62,8 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Channel processing manager handles all starting and stopping of channel decoding.  A processing chain is created
@@ -74,7 +76,9 @@ public class ChannelProcessingManager implements Listener<ChannelEvent>
     private final static Logger mLog = LoggerFactory.getLogger(ChannelProcessingManager.class);
     private static final String TUNER_UNAVAILABLE_DESCRIPTION = "TUNER UNAVAILABLE";
     private Map<Channel,ProcessingChain> mProcessingChains = new ConcurrentHashMap<>();
+    private Lock mLock = new ReentrantLock();
 
+    private ChannelSourceEventErrorListener mSourceErrorListener = new ChannelSourceEventErrorListener();
     private List<Listener<AudioSegment>> mAudioSegmentListeners = new CopyOnWriteArrayList<>();
     private List<Listener<IDecodeEvent>> mDecodeEventListeners = new CopyOnWriteArrayList<>();
     private Broadcaster<ChannelEvent> mChannelEventBroadcaster = new Broadcaster();
@@ -170,9 +174,8 @@ public class ChannelProcessingManager implements Listener<ChannelEvent>
      *
      * @param event that requests either enable/start or disable/stop a channel.
      */
-    @Deprecated //TODO: convert all usage to start(ChannelStartProcessingRequest request) method
     @Override
-    public synchronized void receive(ChannelEvent event)
+    public void receive(ChannelEvent event)
     {
         Channel channel = event.getChannel();
 
@@ -350,38 +353,9 @@ public class ChannelProcessingManager implements Listener<ChannelEvent>
             processingChain.addDecodeEventListener(listener);
         }
 
-        //Add a listener to detect source error state that indicates the channel should be shutdown
-        processingChain.addSourceEventListener(sourceEvent ->
-        {
-            if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE && sourceEvent.getSource() != null)
-            {
-                Channel toShutdown = null;
-
-                for(Map.Entry<Channel,ProcessingChain> entry: mProcessingChains.entrySet())
-                {
-                    if(entry.getValue().hasSource(sourceEvent.getSource()))
-                    {
-                        toShutdown = entry.getKey();
-                        break;
-                    }
-                }
-
-                if(toShutdown != null)
-                {
-                    mLog.warn("Channel source error detected - stopping channel [" + toShutdown.getName() + "]");
-
-                    try
-                    {
-                        stopProcessing(toShutdown);
-                    }
-                    catch(ChannelException ce)
-                    {
-                        mLog.error("Error stopping channel [" + channel.getName() + "] with source error - " +
-                            ce.getMessage());
-                    }
-                }
-            }
-        });
+        //Add a listener to detect source error state that indicates the channel should be shutdown.
+        //Note: processing chain will only add this once.
+        processingChain.addSourceEventListener(mSourceErrorListener);
 
         //Register this manager to receive channel events from traffic channel manager modules within
         //the processing chain
@@ -459,18 +433,90 @@ public class ChannelProcessingManager implements Listener<ChannelEvent>
                     processingChain.getChannelState().updateChannelStateIdentifiers(notification);
                 }
             }
-
         }
 
-        mProcessingChains.put(channel, processingChain);
-        processingChain.start();
+        if(addProcessingChain(channel, processingChain))
+        {
+            processingChain.start();
 
-        //This has to be done on the FX event thread when the playlist editor is constructed
-        Platform.runLater(() -> channel.setProcessing(true));
+            //This has to be done on the FX event thread when the playlist editor is constructed
+            Platform.runLater(() -> channel.setProcessing(true));
 
-        getChannelMetadataModel().add(new ChannelAndMetadata(channel, processingChain.getChannelState().getChannelMetadata()));
+            mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_START));
+        }
+        else
+        {
+            mLog.warn("Channel [" + channel.getName() + "] processing chain not added because it already exists");
+            processingChain.removeEventLoggingModules();
+            processingChain.removeRecordingModules();
+            processingChain.removeFrequencyChangeListener(channel);
+            channel.resetFrequencyCorrection();
+            mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP));
+            mChannelEventBroadcaster.removeListener(processingChain);
+            processingChain.getEventBus().unregister(ChannelProcessingManager.this);
+            processingChain.dispose();
+        }
+    }
 
-        mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_START));
+    /**
+     * Thread-safe add processing chain and add channel metadata to channel metadata model.
+     * @param channel for the processing chain
+     * @param processingChain to add
+     * @return true if processing chain was added or false if it was not added due to there already
+     * being a processing chain registered for that channel.
+     */
+    private boolean addProcessingChain(Channel channel, ProcessingChain processingChain)
+    {
+        boolean added = false;
+
+        mLock.lock();
+
+        try
+        {
+            if(!mProcessingChains.containsKey(channel))
+            {
+                added = true;
+                mProcessingChains.put(channel, processingChain);
+                getChannelMetadataModel().add(new ChannelAndMetadata(channel, processingChain.getChannelState().getChannelMetadata()));
+            }
+        }
+        finally
+        {
+            mLock.unlock();
+        }
+
+        return added;
+    }
+
+    /**
+     * Thread-safe remove processing chain.
+     * @param channel for identifying the processing chain
+     * @return the removed processing chain or null
+     */
+    private ProcessingChain removeProcessingChain(Channel channel)
+    {
+        ProcessingChain removed = null;
+
+        mLock.lock();
+
+        try
+        {
+            removed = mProcessingChains.remove(channel);
+
+            if(removed != null)
+            {
+                for(ChannelMetadata channelMetadata: removed.getChannelState().getChannelMetadata())
+                {
+                    getChannelMetadataModel().remove(channelMetadata);
+                }
+            }
+        }
+        finally
+        {
+            mLock.unlock();
+        }
+
+        return removed;
     }
 
     /**
@@ -483,30 +529,34 @@ public class ChannelProcessingManager implements Listener<ChannelEvent>
         //This has to be done on the FX event thread when the playlist editor is constructed
         Platform.runLater(() -> channel.setProcessing(false));
 
-        if(mProcessingChains.containsKey(channel))
+        ProcessingChain processingChain = removeProcessingChain(channel);
+
+        if(processingChain != null)
         {
-            ProcessingChain processingChain = mProcessingChains.remove(channel);
+            //Since the processing chain's source will block on stop(), throw it to the thread pool to run
+            ThreadPool.CACHED.submit(() -> {
+                try
+                {
+                    processingChain.stop();
+                    processingChain.removeEventLoggingModules();
+                    processingChain.removeRecordingModules();
 
-            for(ChannelMetadata channelMetadata: processingChain.getChannelState().getChannelMetadata())
-            {
-                getChannelMetadataModel().remove(channelMetadata);
-            }
+                    //Deregister channel from receive frequency correction events to show in the spectral display (hack!)
+                    processingChain.removeFrequencyChangeListener(channel);
+                    channel.resetFrequencyCorrection();
 
-            processingChain.stop();
+                    mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP));
+                    mChannelEventBroadcaster.removeListener(processingChain);
 
-            processingChain.removeEventLoggingModules();
-            processingChain.removeRecordingModules();
-
-            //Deregister channel from receive frequency correction events to show in the spectral display (hack!)
-            processingChain.removeFrequencyChangeListener(channel);
-            channel.resetFrequencyCorrection();
-
-            mChannelEventBroadcaster.broadcast(new ChannelEvent(channel, ChannelEvent.Event.NOTIFICATION_PROCESSING_STOP));
-            mChannelEventBroadcaster.removeListener(processingChain);
-
-            //Unregister for event bus requests and notifications
-            processingChain.getEventBus().unregister(ChannelProcessingManager.this);
-            processingChain.dispose();
+                    //Unregister for event bus requests and notifications
+                    processingChain.getEventBus().unregister(ChannelProcessingManager.this);
+                    processingChain.dispose();
+                }
+                catch(Exception e)
+                {
+                    mLog.error("Error during shutdown of processing chain for channel [" + channel.getName() + "}", e);
+                }
+            });
         }
         else
         {
@@ -653,4 +703,52 @@ public class ChannelProcessingManager implements Listener<ChannelEvent>
             }
         }
     }
+
+    /**
+     * Monitors all channels for an error in the source event that would require the
+     * channel's processing chain to be stopped
+     */
+    private class ChannelSourceEventErrorListener implements Listener<SourceEvent>
+    {
+        @Override public void receive(SourceEvent sourceEvent)
+        {
+            if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE && sourceEvent.getSource() != null)
+            {
+                Channel toShutdown = null;
+
+                mLock.lock();
+
+                try
+                {
+                    for(Map.Entry<Channel,ProcessingChain> entry: mProcessingChains.entrySet())
+                    {
+                        if(entry.getValue().hasSource(sourceEvent.getSource()))
+                        {
+                            toShutdown = entry.getKey();
+                            break;
+                        }
+                    }
+                }
+                finally
+                {
+                    mLock.unlock();
+                }
+
+                if(toShutdown != null)
+                {
+                    mLog.warn("Channel source error detected - stopping channel [" + toShutdown.getName() + "]");
+
+                    try
+                    {
+                        stopProcessing(toShutdown);
+                    }
+                    catch(ChannelException ce)
+                    {
+                        mLog.error("Error stopping channel [" + (toShutdown != null ? toShutdown.getName() : "unknown") +
+                                "] with source error - " + ce.getMessage());
+                    }
+                }
+            }
+        }
+    }
 }

src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java

@@ -23,25 +23,29 @@ import io.github.dsheirer.dsp.filter.design.FilterDesignException;
 import io.github.dsheirer.sample.Listener;
 import io.github.dsheirer.sample.complex.ComplexSamples;
 import io.github.dsheirer.source.SourceEvent;
+import io.github.dsheirer.source.tuner.channel.StreamProcessorWithHeartbeat;
 import io.github.dsheirer.source.tuner.channel.TunerChannel;
 import io.github.dsheirer.source.tuner.channel.TunerChannelSource;
 
 import java.util.List;
 
+/**
+ * Polyphase Channelizer's Tuner Channel Source implementation.  Wraps a ChannelOutputProcessor instance and
+ * provides tuner channel source functionality.  Supports dynamic swapout of the underlying channel output processor
+ * as the upstream polyphase channelizer reconfigures during tuned center frequency changes.
+ */
 public class PolyphaseChannelSource extends TunerChannelSource implements Listener<ComplexSamples>
 {
     private IPolyphaseChannelOutputProcessor mPolyphaseChannelOutputProcessor;
     private IPolyphaseChannelOutputProcessor mReplacementPolyphaseChannelOutputProcessor;
+    private StreamProcessorWithHeartbeat<ComplexSamples> mStreamHeartbeatProcessor;
     private long mReplacementFrequency;
     private double mChannelSampleRate;
     private long mIndexCenterFrequency;
     private long mChannelFrequencyCorrection;
-    private Listener<ComplexSamples> mComplexSamplesListener;
 
     /**
-     * Polyphase channelizer tuner channel source implementation.  Adapts the channel array output samples from the
-     * polyphase channelizer into a single channel, or a channel synthesized from two adjacent channels that is
-     * frequency translated and decimated to a single channel.
+     * Constructs an instance
      *
      * @param tunerChannel describing the desired channel frequency and bandwidth/minimum sample rate
      * @param outputProcessor - to process polyphase channelizer channel results into a channel stream
@@ -58,7 +62,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen
         mPolyphaseChannelOutputProcessor = outputProcessor;
         mPolyphaseChannelOutputProcessor.setListener(this);
         mChannelSampleRate = channelSampleRate;
-
+        mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS);
         setFrequency(centerFrequency);
     }
 
@@ -67,6 +71,8 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen
     {
         super.start();
 
+        mStreamHeartbeatProcessor.start();
+
         if(mPolyphaseChannelOutputProcessor != null)
         {
             mPolyphaseChannelOutputProcessor.start();
@@ -82,6 +88,8 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen
         {
             mPolyphaseChannelOutputProcessor.stop();
         }
+
+        mStreamHeartbeatProcessor.stop();
     }
 
     /**
@@ -90,19 +98,14 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen
     @Override
     public void setListener(final Listener<ComplexSamples> listener)
     {
-        mComplexSamplesListener = listener;
+        mStreamHeartbeatProcessor.setListener(listener);
         mPolyphaseChannelOutputProcessor.setListener(this);
     }
 
     @Override
     public void receive(ComplexSamples complexSamples)
     {
-        getHeartbeatManager().broadcast();
-
-        if(mComplexSamplesListener != null)
-        {
-            mComplexSamplesListener.receive(complexSamples);
-        }
+        mStreamHeartbeatProcessor.receive(complexSamples);
     }
 
     /**

src/main/java/io/github/dsheirer/source/tuner/channel/CICTunerChannelSource.java

@@ -1,247 +0,0 @@
-/*
- * *****************************************************************************
- * Copyright (C) 2014-2022 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 io.github.dsheirer.source.tuner.channel;
-
-import io.github.dsheirer.buffer.INativeBuffer;
-import io.github.dsheirer.buffer.NativeBufferPoisonPill;
-import io.github.dsheirer.dsp.filter.cic.PrimeCicDecimationFilter;
-import io.github.dsheirer.dsp.filter.design.FilterDesignException;
-import io.github.dsheirer.dsp.mixer.ComplexMixer;
-import io.github.dsheirer.dsp.mixer.ComplexMixerFactory;
-import io.github.dsheirer.sample.Listener;
-import io.github.dsheirer.sample.complex.ComplexSamples;
-import io.github.dsheirer.sample.complex.ComplexSamplesRepackager;
-import io.github.dsheirer.source.SourceEvent;
-import io.github.dsheirer.util.Dispatcher;
-
-import java.util.Iterator;
-import java.util.List;
-
-/**
- * Complex sample source that provides a frequency-translated and decimated sample buffer stream.  Uses a
- * CIC decimation filter that requires the decimation rate to be an integer multiple.  Sample buffer processing
- * occurs on a scheduled runnable thread.
- */
-public class CICTunerChannelSource<T extends INativeBuffer> extends TunerChannelSource implements Listener<T>
-{
-    //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;
-
-    //Samples repackager is set to 2048 (a power of 2) to align with channel processing chain requirements.
-    private final ComplexSamplesRepackager mSamplesRepackager = new ComplexSamplesRepackager(1024);
-    private Dispatcher<T> mBufferDispatcher;
-    private ComplexMixer mFrequencyCorrectionMixer;
-    private PrimeCicDecimationFilter mIDecimationFilter;
-    private PrimeCicDecimationFilter mQDecimationFilter;
-    private double mChannelSampleRate;
-    private long mChannelFrequencyCorrection = 0;
-    private long mTunerFrequency;
-    private Listener<ComplexSamples> mListener;
-
-    /**
-     * Constructs a frequency translating and CIC decimating channel source.
-     *
-     * @param producerSourceEventListener to receive sample stream start/stop requests
-     * @param tunerChannel that details the desired channel frequency and bandwidth
-     * @param sampleRate of the incoming sample stream
-     * @param channelSpecification for the requested channel.
-     * @throws FilterDesignException if a final cleanup filter cannot be designed using the remez filter
-     *                               designer and the filter parameters.
-     */
-    public CICTunerChannelSource(Listener<SourceEvent> producerSourceEventListener, TunerChannel tunerChannel,
-                 double sampleRate, ChannelSpecification channelSpecification) throws FilterDesignException
-    {
-        super(producerSourceEventListener, tunerChannel);
-
-        int decimation = (int)(sampleRate / channelSpecification.getMinimumSampleRate());
-
-        mIDecimationFilter = new PrimeCicDecimationFilter(sampleRate, decimation, channelSpecification.getPassFrequency(),
-                channelSpecification.getStopFrequency());
-        mQDecimationFilter = new PrimeCicDecimationFilter(sampleRate, decimation, channelSpecification.getPassFrequency(),
-                channelSpecification.getStopFrequency());
-
-        mBufferDispatcher = new Dispatcher(BUFFER_MAX_CAPACITY, "sdrtrunk cic channel " +
-                tunerChannel.getFrequency(), new NativeBufferPoisonPill());
-        mBufferDispatcher.setListener(new NativeBufferProcessor());
-
-        //Setup the frequency mixer to the current source frequency
-        mChannelSampleRate = sampleRate / (double)decimation;
-        mTunerFrequency = tunerChannel.getFrequency();
-        long frequencyOffset = mTunerFrequency - getTunerChannel().getFrequency();
-
-        mFrequencyCorrectionMixer = ComplexMixerFactory.getMixer(frequencyOffset, sampleRate);
-    }
-
-    @Override
-    public void start()
-    {
-        super.start();
-        mBufferDispatcher.start();
-    }
-
-    @Override
-    public void stop()
-    {
-        super.stop();
-        mBufferDispatcher.stop();
-    }
-
-    @Override
-    public void dispose()
-    {
-        //
-    }
-
-    /**
-     * Primary interface for receiving incoming complex sample buffers to be frequency translated and decimated.
-     */
-    @Override
-    public void receive(T complexSamples)
-    {
-        mBufferDispatcher.receive(complexSamples);
-    }
-
-    /**
-     * Sets/updates the center frequency for the sample streaming being sent from the producer.
-     *
-     * @param frequency in hertz
-     */
-    @Override
-    public void setFrequency(long frequency)
-    {
-        mTunerFrequency = frequency;
-        updateMixerFrequencyOffset();
-    }
-
-    /**
-     * Current frequency correction being applied to the channel sample stream
-     *
-     * @return correction in hertz
-     */
-    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.
-     */
-    public void setFrequencyCorrection(long correction)
-    {
-        mChannelFrequencyCorrection = correction;
-        updateMixerFrequencyOffset();
-        broadcastConsumerSourceEvent(SourceEvent.frequencyCorrectionChange(mChannelFrequencyCorrection));
-    }
-
-    /**
-     * 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);
-    }
-
-    /**
-     * Sets the sample rate of the incoming sample stream from the producer
-     *
-     * @param sampleRate in hertz
-     */
-    @Override
-    protected void setSampleRate(double sampleRate)
-    {
-        //Not implemented.  Sample rate changes are not permitted once sample stream starts
-    }
-
-    /**
-     * Sets the frequency correction for the outbound sample stream to the consumer
-     *
-     * @param correction in hertz
-     */
-    @Override
-    protected void setChannelFrequencyCorrection(long correction)
-    {
-        mChannelFrequencyCorrection = correction;
-        broadcastConsumerSourceEvent(SourceEvent.channelFrequencyCorrectionChange(correction));
-        updateMixerFrequencyOffset();
-    }
-
-    /**
-     * Frequency correction value currently being applied to the outbound sample stream
-     */
-    @Override
-    public long getChannelFrequencyCorrection()
-    {
-        return mChannelFrequencyCorrection;
-    }
-
-    /**
-     * Sets the listener to receive the complex buffer sample output from this channel
-     *
-     * @param listener to receive complex buffers
-     */
-    @Override
-    public void setListener(Listener<ComplexSamples> listener)
-    {
-        mListener = listener;
-    }
-
-    @Override
-    public double getSampleRate()
-    {
-        return mChannelSampleRate;
-    }
-
-    /**
-     * Processes native buffers received from the dispatcher and sends to the registered listener
-     */
-    public class NativeBufferProcessor implements Listener<T>
-    {
-        @Override
-        public void receive(T nativeBuffer)
-        {
-            getHeartbeatManager().broadcast();
-
-            if(mListener != null)
-            {
-                Iterator<ComplexSamples> iterator = nativeBuffer.iterator();
-
-                while(iterator.hasNext())
-                {
-                    ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next());
-                    float[] i = mIDecimationFilter.decimate(basebanded.i());
-                    float[] q = mQDecimationFilter.decimate(basebanded.q());
-
-                    List<ComplexSamples> repackaged = mSamplesRepackager.process(i, q);
-
-                    for(ComplexSamples samples: repackaged)
-                    {
-                        mListener.receive(samples);
-                    }
-                }
-            }
-        }
-    }
-}

src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java

@@ -54,6 +54,7 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     private long mChannelFrequencyCorrection = 0;
     private long mTunerFrequency;
     private Listener<ComplexSamples> mListener;
+    private StreamProcessorWithHeartbeat<ComplexSamples> mStreamHeartbeatProcessor;
 
     /**
      * Constructs a frequency translating and CIC decimating channel source.
@@ -70,6 +71,8 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     {
         super(producerSourceEventListener, tunerChannel);
 
+        mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS);
+
         int desiredDecimation = (int)(sampleRate / channelSpecification.getMinimumSampleRate());
         int decimation = DecimationFilterFactory.getDecimationRate(desiredDecimation);
 
@@ -91,6 +94,7 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     public void start()
     {
         super.start();
+        mStreamHeartbeatProcessor.start();
         mBufferDispatcher.start();
     }
 
@@ -99,6 +103,7 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     {
         super.stop();
         mBufferDispatcher.stop();
+        mStreamHeartbeatProcessor.stop();
     }
 
     @Override
@@ -200,13 +205,7 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     @Override
     public void setListener(Listener<ComplexSamples> listener)
     {
-        if(listener == null)
-        {
-            //Wait for the native buffer processor thread to die before we nullify the listener
-            stop();
-        }
-
-        mListener = listener;
+        mStreamHeartbeatProcessor.setListener(listener);
     }
 
     @Override
@@ -223,21 +222,14 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
         @Override
         public void receive(T nativeBuffer)
         {
-            getHeartbeatManager().broadcast();
+            Iterator<ComplexSamples> iterator = nativeBuffer.iterator();
 
-            Listener<ComplexSamples> listener = mListener;
-
-            if(listener != null)
+            while(iterator.hasNext())
             {
-                Iterator<ComplexSamples> iterator = nativeBuffer.iterator();
-
-                while(iterator.hasNext())
-                {
-                    ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next());
-                    float[] i = mIDecimationFilter.decimateReal(basebanded.i());
-                    float[] q = mQDecimationFilter.decimateReal(basebanded.q());
-                    listener.receive(new ComplexSamples(i, q));
-                }
+                ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next());
+                float[] i = mIDecimationFilter.decimateReal(basebanded.i());
+                float[] q = mQDecimationFilter.decimateReal(basebanded.q());
+                mStreamHeartbeatProcessor.receive(new ComplexSamples(i, q));
             }
         }
     }

src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java

@@ -42,7 +42,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou
     private final static Logger mLog = LoggerFactory.getLogger(PassThroughChannelSource.class);
     private TunerController mTunerController;
     private Dispatcher<INativeBuffer> mBufferDispatcher;
-    private Listener<ComplexSamples> mComplexSamplesListener;
+    private StreamProcessorWithHeartbeat<ComplexSamples> mStreamHeartbeatProcessor;
 
     /**
      * Constructs an instance
@@ -59,12 +59,14 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou
         mBufferDispatcher = new Dispatcher<>(500, "sdrtrunk pass-through channel " +
                 tunerChannel.getFrequency(), new NativeBufferPoisonPill());
         mBufferDispatcher.setListener(new BufferProcessor());
+        mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS);
     }
 
     @Override
     public void start()
     {
         super.start();
+        mStreamHeartbeatProcessor.stop();
         mBufferDispatcher.start();
     }
 
@@ -73,6 +75,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou
     {
         super.stop();
         mBufferDispatcher.stop();
+        mStreamHeartbeatProcessor.stop();
     }
 
     @Override
@@ -100,9 +103,9 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou
     }
 
     @Override
-    public void setListener(Listener<ComplexSamples> complexSamplesListener)
+    public void setListener(Listener<ComplexSamples> listener)
     {
-        mComplexSamplesListener = complexSamplesListener;
+        mStreamHeartbeatProcessor.setListener(listener);
     }
 
     @Override
@@ -122,16 +125,11 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou
         @Override
         public void receive(INativeBuffer nativeBuffer)
         {
-            getHeartbeatManager().broadcast();
+            Iterator<ComplexSamples> iterator = nativeBuffer.iterator();
 
-            if(mComplexSamplesListener != null)
+            while(iterator.hasNext())
             {
-                Iterator<ComplexSamples> iterator = nativeBuffer.iterator();
-
-                while(iterator.hasNext())
-                {
-                    mComplexSamplesListener.receive(iterator.next());
-                }
+                mStreamHeartbeatProcessor.receive(iterator.next());
             }
         }
     }

src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java

@@ -0,0 +1,135 @@
+/*
+ * *****************************************************************************
+ * Copyright (C) 2014-2022 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 io.github.dsheirer.source.tuner.channel;
+
+import io.github.dsheirer.sample.Listener;
+import io.github.dsheirer.source.heartbeat.HeartbeatManager;
+import io.github.dsheirer.util.ThreadPool;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Thread-safe implementation that coordinates the flow of data/buffer stream processing with a parallel, periodic
+ * heartbeat using an internal locking mechanism to coordinate the two activities.  This class is intended to be used
+ * by TunerChannelSource implementations to coordinate the flow of sample buffers produced on a dispatch thread with a
+ * scheduled timer driven heartbeat mechanism, to ensure the downstream listener receives either samples or a
+ * heartbeat, but not both at the same time.
+ *
+ * Samples are processed as they arrive and are immediately followed by a heartbeat.  A separate scheduled timer runs
+ * on a periodic schedule to optionally fire a heartbeat, independent of the sample stream rate.  The scheduled
+ * heartbeat is optional and will skip a heartbeat interval if this processor is currently dispatching a sample buffer.
+ * Sample stream processing will block until a currently running scheduled heartbeat timer event completes, when these
+ * two events occur at the same time.
+ */
+public class StreamProcessorWithHeartbeat<T> implements Listener<T>
+{
+    private ReentrantLock mLock = new ReentrantLock();
+    private HeartbeatManager mHeartbeatManager;
+    private long mHeartbeatInterval;
+    private Listener<T> mListener;
+    private ScheduledFuture<?> mHeartbeatTimerFuture;
+
+    /**
+     * Constructs an instance
+     * @param heartbeatManager that will be invoked by an internal scheduled timer
+     * @param heartbeatInterval in milliseconds
+     */
+    public StreamProcessorWithHeartbeat(HeartbeatManager heartbeatManager, long heartbeatInterval)
+    {
+        mHeartbeatManager = heartbeatManager;
+        mHeartbeatInterval = heartbeatInterval;
+    }
+
+    /**
+     * Starts the heartbeat timer
+     */
+    public void start()
+    {
+        if(mHeartbeatTimerFuture == null)
+        {
+            mHeartbeatTimerFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(() -> fireOptionalHeartbeat(),
+                    mHeartbeatInterval, mHeartbeatInterval, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /**
+     * Stops the heartbeat timer
+     */
+    public void stop()
+    {
+        if(mHeartbeatTimerFuture != null)
+        {
+            mHeartbeatTimerFuture.cancel(true);
+            mHeartbeatTimerFuture = null;
+        }
+    }
+
+    /**
+     * Optionally fires a heartbeat if we're not currently processing a native buffer
+     */
+    private void fireOptionalHeartbeat()
+    {
+        //Acquire lock only if it's not already locked, otherwise ignore heartbeat for this timer interval.
+        if(mLock.tryLock())
+        {
+            try
+            {
+                mHeartbeatManager.broadcast();
+            }
+            finally
+            {
+                mLock.unlock();
+            }
+        }
+    }
+
+    @Override
+    public void receive(T data)
+    {
+        mLock.lock();
+
+        try
+        {
+            mHeartbeatManager.broadcast();
+
+            Listener<T> listener = mListener;
+
+            if(listener != null)
+            {
+                listener.receive(data);
+            }
+        }
+        finally
+        {
+            mLock.unlock();
+        }
+    }
+
+    /**
+     * Registers the listener to receive data/buffers for the defined generic type that are distributed to this processor.
+     * @param listener to register or null to deregister an existing listener
+     */
+    public void setListener(Listener<T> listener)
+    {
+        mListener = listener;
+    }
+}

src/main/java/io/github/dsheirer/source/tuner/channel/TunerChannelSource.java

@@ -31,12 +31,11 @@ import org.slf4j.LoggerFactory;
 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;
+    protected static final long HEARTBEAT_INTERVAL_MS = 100;
     private SourceEventListenerToProcessorAdapter mConsumerSourceEventListenerAdapter;
     protected TunerChannel mTunerChannel;
     private Listener<SourceEvent> mProducerSourceEventListener;
     private Listener<SourceEvent> mConsumerSourceEventListener;
-//    private ScheduledIntervalProcessor mScheduledIntervalProcessor = new ScheduledIntervalProcessor();
 
     /**
      * Tuner Channel Source is a Digital Drop Channel (DDC) abstract class that defines the minimum functionality
@@ -245,95 +244,4 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
     {
         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 ScheduledIntervalProcessor 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()
-//        {
-//            try
-//            {
-//                if(!mStopped)
-//                {
-//                    try
-//                    {
-//                        getHeartbeatManager().broadcast();
-//                    }
-//                    catch(Throwable t)
-//                    {
-//                        mLog.error("Error while sending heartbeat", t);
-//                    }
-//                }
-//
-//                if(!mStopped)
-//                {
-//                    try
-//                    {
-//                        processSamples();
-//                    }
-//                    catch(Throwable t)
-//                    {
-//                        mLog.error("Error while processing samples", t);
-//                    }
-//                }
-//
-//                if(mStopped)
-//                {
-//                    if(mScheduledFuture != null)
-//                    {
-//                        //Set may-interrupt to false so that we can complete this iteration
-//                        mScheduledFuture.cancel(false);
-//                    }
-//
-//                    mScheduledFuture = null;
-//
-//                    try
-//                    {
-//                        getHeartbeatManager().broadcast();
-//                        performDisposal();
-//                    }
-//                    catch(Throwable t)
-//                    {
-//                        mLog.error("Error during final shutdown processing of samples", t);
-//                    }
-//                }
-//            }
-//            catch(Throwable t)
-//            {
-//                mLog.error("Error during heartbeat processing", t);
-//            }
-//        }
-//    }
 }