Updates build to version 0.3.0 Beta 13

249 Channel TEARDOWN/OVERFLOW state lockup

build/build.xml

@@ -7,7 +7,7 @@
   <property name="major" value="0"/>
   <property name="minor" value="3"/>
   <property name="patch" value="0"/>
-  <property name="release" value="-beta12"/>
+  <property name="release" value="-beta13"/>
   <property name="filename" value="sdrtrunk_${major}.${minor}.${patch}${release}"/>
     
   <!-- Build properties -->

src/channel/heartbeat/Heartbeat.java

@@ -0,0 +1,31 @@
+/*******************************************************************************
+ * 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 channel.heartbeat;
+
+public class Heartbeat
+{
+    /**
+     * Periodic pulse that is broadcast to subscribing modules within the demodulating chain so that
+     * monitoring of state and other attributes can occur on the primary demodulation and decoding
+     * thread.
+     */
+    public Heartbeat()
+    {
+    }
+}

src/channel/heartbeat/IHeartbeatListener.java

@@ -0,0 +1,26 @@
+/*******************************************************************************
+ * 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 channel.heartbeat;
+
+import sample.Listener;
+
+public interface IHeartbeatListener
+{
+    public Listener<Heartbeat> getHeartbeatListener();
+}

src/channel/heartbeat/IHeartbeatProvider.java

@@ -0,0 +1,28 @@
+/*******************************************************************************
+ * 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 channel.heartbeat;
+
+import sample.Listener;
+
+public interface IHeartbeatProvider
+{
+    public void setHeartbeatListener(Listener<Heartbeat> listener);
+
+    public void removeHeartbeatListener();
+}

src/channel/state/ChannelState.java

@@ -20,6 +20,8 @@ package channel.state;
 
 import audio.squelch.ISquelchStateProvider;
 import audio.squelch.SquelchState;
+import channel.heartbeat.Heartbeat;
+import channel.heartbeat.IHeartbeatListener;
 import channel.metadata.Attribute;
 import channel.metadata.AttributeChangeRequest;
 import channel.metadata.IAttributeChangeRequestListener;
@@ -39,16 +41,13 @@ import sample.real.IOverflowListener;
 import source.tuner.frequency.FrequencyChangeEvent;
 import source.tuner.frequency.IFrequencyChangeListener;
 
-import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
 
 import static source.tuner.frequency.FrequencyChangeEvent.Event.NOTIFICATION_FREQUENCY_CHANGE;
 
 public class ChannelState extends Module implements ICallEventProvider, IDecoderStateEventListener,
-    IDecoderStateEventProvider, IFrequencyChangeListener, ISquelchStateProvider, IAttributeChangeRequestListener,
+    IDecoderStateEventProvider, IFrequencyChangeListener, IHeartbeatListener, ISquelchStateProvider,
-    IOverflowListener
+    IAttributeChangeRequestListener, IOverflowListener
 {
     private final static Logger mLog = LoggerFactory.getLogger(ChannelState.class);
 
@@ -61,8 +60,7 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
     private Listener<DecoderStateEvent> mDecoderStateListener;
     private Listener<SquelchState> mSquelchStateListener;
     private DecoderStateEventReceiver mDecoderStateEventReceiver = new DecoderStateEventReceiver();
-    private StateMonitor mStateMonitor = new StateMonitor();
+    private HeartbeatReceiver mHeartbeatReceiver = new HeartbeatReceiver();
-    private ScheduledFuture<?> mStateMonitorFuture;
     private ChannelType mChannelType;
     private TrafficChannelManager mTrafficChannelEndListener;
     private CallEvent mTrafficChannelCallEvent;
@@ -79,30 +77,21 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
 
 
     /**
-     * Channel state tracks the overall state of all processing modules and decoders
+     * Channel state tracks the overall state of all processing modules and decoders configured for the channel and
-     * configured for the channel and provides squelch control and decoder state
+     * provides squelch control and decoder state reset events.
-     * reset events.
      *
-     * Uses a state enumeration that defines allowable channel state transitions in
+     * Uses a state enumeration that defines allowable channel state transitions in order to track a call or data decode
-     * order to track a call or data decode event from start to finish.  Uses a
+     * event from start to finish.  Uses a timer to monitor for inactivity and to provide a FADE period that indicates
-     * timer to monitor for inactivity and to provide a FADE period that indicates
+     * to the user that the activity has stopped while continuing to provide details about the call, before the state is
-     * to the user that the activity has stopped while continuing to provide details
+     * reset to IDLE.
-     * about the call, before the state is reset to IDLE.
-     *
-     * Since this class is multi-threaded between decoder events and the internal
-     * timer that monitors for inactivity, this class uses the mState object for
-     * thread synchronization.
      *
      * State Descriptions:
-     * IDLE		Normal state. No voice or data call activity
+     * IDLE:  Normal state. No voice or data call activity
-     * CALL/DATA/ENCRYPTED/CONTROL
+     * CALL/DATA/ENCRYPTED/CONTROL:  Decoding states.
-     * Decoding states.
+     * FADE:  The phase after a voice or data call when either an explicit call end has been received, or when no new
-     * FADE		The phase after a voice or data call when either an explicit
+     *     signalling updates have been received, and the fade timer has expired.  This phase allows for gui updates to
-     * call end has been received, or when no new signalling updates
+     *     signal to the user that the call is ended, while continuing to display the call details for the user
-     * have been received, and the fade timer has expired.  This phase
+     * TEARDOWN:  Indicates a traffic channel that will be torn down for reuse.
-     * allows for gui updates to signal to the user that the call is
-     * ended, while continuing to display the call details for the user
-     * TEARDOWN	Indicates a traffic channel that will be torn down for reuse.
      */
     public ChannelState(ChannelType channelType)
     {
@@ -136,8 +125,6 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
 
     @Override
     public void start(ScheduledExecutorService executor)
-    {
-        if(mStateMonitorFuture == null && mStateMonitor != null)
     {
         mMutableMetadata.receive(new AttributeChangeRequest<State>(Attribute.CHANNEL_STATE, mState));
 
@@ -145,16 +132,6 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
         {
             setState(State.CALL);
         }
-
-            try
-            {
-                mStateMonitorFuture = executor.scheduleAtFixedRate(mStateMonitor, 0, 20, TimeUnit.MILLISECONDS);
-            }
-            catch(RejectedExecutionException ree)
-            {
-                mLog.error("state monitor scheduled task rejected", ree);
-            }
-        }
     }
 
     @Override
@@ -162,16 +139,6 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
     {
         mMutableMetadata.resetAllAttributes();
 
-        if(mStateMonitorFuture != null)
-        {
-            boolean success = mStateMonitorFuture.cancel(true);
-
-            if(!success)
-            {
-                mLog.error("Couldn't stop monitoring scheduled future");
-            }
-        }
-
         mTrafficChannelEndListener = null;
 
         if(mTrafficChannelCallEvent != null)
@@ -182,8 +149,6 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
 
         mTrafficChannelCallEvent = null;
 
-        mStateMonitorFuture = null;
-
         mSquelchLocked = false;
     }
 
@@ -192,7 +157,6 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
         mCallEventListener = null;
         mDecoderStateListener = null;
         mSquelchStateListener = null;
-        mStateMonitor = null;
     }
 
     @Override
@@ -312,15 +276,12 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
     }
 
     /**
-     * Sets the channel state to the specified state, or updates the timeout
+     * Sets the channel state to the specified state, or updates the timeout values so that the state monitor will not
-     * values so that the state monitor will not change state.  Broadcasts a
+     * change state.  Broadcasts a squelch event when the state changes and the audio squelch state should change.  Also
-     * squelch event when the state changes and the audio squelch state should
+     * broadcasts changed attribute and decoder state events so that external processes can maintain sync with this
-     * change.  Also broadcasts changed attribute and decoder state events so
+     * channel state.
-     * that external processes can maintain sync with this channel state.
      */
     protected void setState(State state)
-    {
-        synchronized(mState)
     {
         if(state == mState)
         {
@@ -329,13 +290,12 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
                 updateFadeTimeout();
             }
         }
-            else if(state != mState && mState.canChangeTo(state))
+        else if(mState.canChangeTo(state))
         {
             switch(state)
             {
                 case CONTROL:
-                        //Don't allow traffic channels to be control channels,
+                    //Don't allow traffic channels to be control channels, otherwise they can't transition to teardown
-                        //otherwise they can't transition to call tear down
                     if(isStandardChannel())
                     {
                         broadcast(SquelchState.SQUELCH);
@@ -374,6 +334,14 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
                 mSourceOverflow ? State.OVERFLOW : mState));
         }
     }
+
+    /**
+     * Receiver inner class that implements the IHeartbeatListener interface to receive heartbeat messages.
+     */
+    @Override
+    public Listener<Heartbeat> getHeartbeatListener()
+    {
+        return mHeartbeatReceiver;
     }
 
     /**
@@ -625,30 +593,27 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
     }
 
     /**
-     * Monitors decoder state events to automatically transition the channel
+     * Processes periodic heartbeats received from the processing chain to perform state monitoring and cleanup
-     * state to IDLE (standard channel) or to TEARDOWN (traffic channel) when
+     * functions.
-     * decoding stops or the monitored channel returns to a no signal state.
      *
-     * Provides a FADE transition state to allow for momentary decoding dropouts
+     * Monitors decoder state events to automatically transition the channel state to IDLE (standard channel) or to
-     * and to allow the user access to call details for a fade period upon
+     * TEARDOWN (traffic channel) when decoding stops or the monitored channel returns to a no signal state.
-     * call end.
+     *
+     * Provides a FADE transition state to allow for momentary decoding dropouts and to allow the user access to call
+     * details for a fade period upon call end.
      */
-    public class StateMonitor implements Runnable
+    public class HeartbeatReceiver implements Listener<Heartbeat>
     {
         @Override
-        public void run()
+        public void receive(Heartbeat heartbeat)
         {
             try
             {
-                synchronized(mState)
+                if(State.CALL_STATES.contains(mState) && mFadeTimeout <= System.currentTimeMillis())
-                {
-                    if(State.CALL_STATES.contains(mState) &&
-                        mFadeTimeout <= System.currentTimeMillis())
                 {
                     processFadeState();
                 }
-                    else if(mState == State.FADE &&
+                else if(mState == State.FADE && mEndTimeout <= System.currentTimeMillis())
-                        mEndTimeout <= System.currentTimeMillis())
                 {
                     if(isTrafficChannel())
                     {
@@ -660,10 +625,9 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder
                     }
                 }
             }
-            }
+            catch(Throwable e)
-            catch(Exception e)
             {
-                mLog.error("Exception thrown while state monitor is running " +
+                mLog.error("An error occurred while state monitor was running " +
                     "- state [" + getState() +
                     "] current [" + System.currentTimeMillis() +
                     "] mResetTimeout [" + mEndTimeout +

src/module/ProcessingChain.java

@@ -24,16 +24,17 @@ import audio.IAudioPacketProvider;
 import audio.squelch.ISquelchStateListener;
 import audio.squelch.ISquelchStateProvider;
 import audio.squelch.SquelchState;
+import channel.heartbeat.Heartbeat;
+import channel.heartbeat.IHeartbeatListener;
+import channel.heartbeat.IHeartbeatProvider;
 import channel.metadata.AttributeChangeRequest;
 import channel.metadata.IAttributeChangeRequestListener;
 import channel.metadata.IAttributeChangeRequestProvider;
 import channel.state.ChannelState;
 import channel.state.DecoderState;
 import channel.state.DecoderStateEvent;
-import channel.state.DecoderStateEvent.Event;
 import channel.state.IDecoderStateEventListener;
 import channel.state.IDecoderStateEventProvider;
-import channel.state.State;
 import controller.channel.Channel.ChannelType;
 import controller.channel.ChannelEvent;
 import controller.channel.IChannelEventListener;
@@ -63,8 +64,6 @@ import sample.real.RealBuffer;
 import source.ComplexSource;
 import source.RealSource;
 import source.Source;
-import source.SourceException;
-import source.tuner.TunerChannelSource;
 import source.tuner.frequency.FrequencyChangeEvent;
 import source.tuner.frequency.IFrequencyChangeListener;
 import source.tuner.frequency.IFrequencyChangeProvider;
@@ -103,6 +102,7 @@ public class ProcessingChain implements IChannelEventListener
     private Broadcaster<ComplexBuffer> mComplexBufferBroadcaster = new Broadcaster<>();
     private Broadcaster<DecoderStateEvent> mDecoderStateEventBroadcaster = new Broadcaster<>();
     private Broadcaster<FrequencyChangeEvent> mFrequencyChangeEventBroadcaster = new Broadcaster<>();
+    private Broadcaster<Heartbeat> mHeartbeatBroadcaster = new Broadcaster<>();
     private Broadcaster<Message> mMessageBroadcaster = new Broadcaster<>();
     private Broadcaster<RealBuffer> mFilteredRealBufferBroadcaster = new Broadcaster<>();
     private Broadcaster<RealBuffer> mUnFilteredRealBufferBroadcaster = new Broadcaster<>();
@@ -329,6 +329,11 @@ public class ProcessingChain implements IChannelEventListener
             }
         }
 
+        if(module instanceof IHeartbeatListener)
+        {
+            mHeartbeatBroadcaster.addListener(((IHeartbeatListener)module).getHeartbeatListener());
+        }
+
         if(module instanceof IMessageListener)
         {
             mMessageBroadcaster.addListener(((IMessageListener) module).getMessageListener());
@@ -391,6 +396,11 @@ public class ProcessingChain implements IChannelEventListener
             mFrequencyChangeEventBroadcaster.removeListener(((IFrequencyChangeListener) module).getFrequencyChangeListener());
         }
 
+        if(module instanceof IHeartbeatListener)
+        {
+            mHeartbeatBroadcaster.removeListener(((IHeartbeatListener)module).getHeartbeatListener());
+        }
+
         if(module instanceof IMessageListener)
         {
             mMessageBroadcaster.removeListener(((IMessageListener) module).getMessageListener());
@@ -448,6 +458,11 @@ public class ProcessingChain implements IChannelEventListener
             ((IFrequencyChangeProvider) module).setFrequencyChangeListener(mFrequencyChangeEventBroadcaster);
         }
 
+        if(module instanceof IHeartbeatProvider)
+        {
+            ((IHeartbeatProvider)module).setHeartbeatListener(mHeartbeatBroadcaster);
+        }
+
         if(module instanceof IMessageProvider)
         {
             ((IMessageProvider) module).setMessageListener(mMessageBroadcaster);
@@ -505,6 +520,11 @@ public class ProcessingChain implements IChannelEventListener
             ((IFrequencyChangeProvider) module).setFrequencyChangeListener(null);
         }
 
+        if(module instanceof IHeartbeatProvider)
+        {
+            ((IHeartbeatProvider)module).removeHeartbeatListener();
+        }
+
         if(module instanceof IMessageProvider)
         {
             ((IMessageProvider) module).setMessageListener(null);

src/module/decode/p25/P25DecoderState.java

@@ -1713,6 +1713,8 @@ public class P25DecoderState extends DecoderState
                         .build());
                     break;
                 case SNDCP_RF_CONFIRMED_DATA:
+                    if(pduc instanceof SNDCPUserData)
+                    {
                         SNDCPUserData sud = (SNDCPUserData)pduc;
 
                         StringBuilder sbFrom = new StringBuilder();
@@ -1740,6 +1742,12 @@ public class P25DecoderState extends DecoderState
                             .from(sbFrom.toString())
                             .to(pduc.getLogicalLinkID())
                             .build());
+                    }
+                    else
+                    {
+                        mLog.error("Error - expected SNDCPUserData instance but class was: " + pduc.getClass() +
+                        " and PDU Type:" + pduc.getPDUType().name());
+                    }
                     break;
                 case SNDCP_RF_UNCONFIRMED_DATA:
                     break;

src/sample/OverflowableTransferQueue.java

@@ -124,10 +124,6 @@ public class OverflowableTransferQueue<E>
             mQueue.clear();
             mCounter.set(0);
             mOverflow.set(false);
-            if(mOverflowListener != null)
-            {
-                mOverflowListener.sourceOverflow(false);
-            }
         }
     }
 }

src/source/Source.java

@@ -17,6 +17,7 @@
  ******************************************************************************/
 package source;
 
+import channel.heartbeat.IHeartbeatProvider;
 import module.Module;
 import sample.SampleType;
 import sample.real.IOverflowListener;
@@ -26,7 +27,8 @@ import source.tuner.frequency.IFrequencyChangeProvider;
 /**
  * Abstract class to define the minimum functionality of a sample data provider.
  */
-public abstract class Source extends Module implements IFrequencyChangeListener, IFrequencyChangeProvider
+public abstract class Source extends Module implements IFrequencyChangeListener, IFrequencyChangeProvider,
+    IHeartbeatProvider
 {
     protected SampleType mSampleType;
     protected IOverflowListener mOverflowListener;

src/source/mixer/RealMixerSource.java

@@ -17,27 +17,26 @@
  ******************************************************************************/
 package source.mixer;
 
-import java.util.Iterator;
+import channel.heartbeat.Heartbeat;
-import java.util.concurrent.CopyOnWriteArrayList;
+import dsp.gain.AutomaticGainControl;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import javax.sound.sampled.AudioFormat;
-import javax.sound.sampled.AudioSystem;
-import javax.sound.sampled.LineUnavailableException;
-import javax.sound.sampled.TargetDataLine;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import sample.Listener;
 import sample.adapter.ISampleAdapter;
 import sample.real.RealBuffer;
 import source.RealSource;
 import source.SourceException;
-import dsp.gain.AutomaticGainControl;
 import source.tuner.frequency.FrequencyChangeEvent;
 
+import javax.sound.sampled.AudioFormat;
+import javax.sound.sampled.AudioSystem;
+import javax.sound.sampled.LineUnavailableException;
+import javax.sound.sampled.TargetDataLine;
+import java.util.Iterator;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 public class RealMixerSource extends RealSource
 {
     private final static Logger mLog = LoggerFactory.getLogger(RealMixerSource.class);
@@ -51,8 +50,10 @@ public class RealMixerSource extends RealSource
     private int mBytesPerFrame = 0;
     private ISampleAdapter mSampleAdapter;
     private AutomaticGainControl mAGC = new AutomaticGainControl();
+    private Listener<Heartbeat> mHeartbeatListener;
+    private static final Heartbeat HEARTBEAT = new Heartbeat();
 
-	CopyOnWriteArrayList<Listener<RealBuffer>> mSampleListeners = new CopyOnWriteArrayList<Listener<RealBuffer>>();
+    private CopyOnWriteArrayList<Listener<RealBuffer>> mSampleListeners = new CopyOnWriteArrayList<Listener<RealBuffer>>();
 
     /**
      * Real Mixer Source - constructs a reader on the mixer/sound card target
@@ -61,11 +62,7 @@ public class RealMixerSource extends RealSource
      * Reads buffers sized to 10% of the sample rate specified in audio format.
      *
      * @param targetDataLine - mixer or sound card to be used
-	 * 
      * @param format - audio format
-	 * 
-	 * @param name - token name to use for this source
-	 * 
      * @param sampleAdapter - adapter to convert byte array data read from the
      * mixer into float array data.  Can optionally invert the channel data if
      * the left/right stereo channels are inverted.
@@ -98,6 +95,26 @@ 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()
     {
@@ -203,6 +220,12 @@ public class RealMixerSource extends RealSource
         {
             if(mRunning.compareAndSet(false, true))
             {
+                //Send a heartbeat every time we run
+                if(mHeartbeatListener != null)
+                {
+                    mHeartbeatListener.receive(HEARTBEAT);
+                }
+
                 if(mTargetDataLine == null)
                 {
                     mRunning.set(false);

src/source/tuner/TunerChannelSource.java

@@ -18,6 +18,8 @@
  ******************************************************************************/
 package source.tuner;
 
+import channel.heartbeat.Heartbeat;
+import channel.heartbeat.IHeartbeatProvider;
 import dsp.filter.FilterFactory;
 import dsp.filter.Window.WindowType;
 import dsp.filter.cic.ComplexPrimeCICDecimate;
@@ -58,6 +60,7 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang
 
     private static int CHANNEL_RATE = 48000;
     private static int CHANNEL_PASS_FREQUENCY = 12000;
+    private static final Heartbeat HEARTBEAT = new Heartbeat();
 
     private OverflowableTransferQueue<ComplexBuffer> mBuffer;
 
@@ -66,6 +69,7 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang
     private Oscillator mMixer;
     private ComplexPrimeCICDecimate mDecimationFilter;
     private Listener<ComplexBuffer> mListener;
+    private Listener<Heartbeat> mHeartbeatListener;
     private IFrequencyChangeProcessor mFrequencyChangeProcessor;
     private DownstreamProcessor mDownstreamFrequencyEventProcessor = new DownstreamProcessor();
     private ScheduledFuture<?> mTaskHandle;
@@ -352,6 +356,25 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang
         return mDownstreamFrequencyEventProcessor.getFrequencyChangeListener();
     }
 
+    /**
+     * 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;
+    }
+
     /**
      * 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
@@ -477,6 +500,13 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang
             {
                 if(mProcessing)
                 {
+                    //Send a heartbeat every time this runs to allow downstream components to perform periodic
+                    //state monitoring functions on this thread
+                    if(mHeartbeatListener != null)
+                    {
+                        mHeartbeatListener.receive(HEARTBEAT);
+                    }
+
                     mBuffer.drainTo(mSampleBuffers, 20);
 
                     for(Buffer buffer : mSampleBuffers)

src/source/tuner/usb/USBTransferProcessor.java

@@ -54,7 +54,7 @@ public class USBTransferProcessor implements TransferCallback
     private static final int TRANSFER_BUFFER_POOL_SIZE = 16;
 
     //Maximum number of filled buffers for the blocking queue
-    private static final int FILLED_BUFFER_MAX_CAPACITY = 300;
+    private static final int FILLED_BUFFER_MAX_CAPACITY = 500;
 
     //Threshold for resetting buffer overflow condition
     private static final int FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD = 100;
@@ -200,7 +200,7 @@ public class USBTransferProcessor implements TransferCallback
 
 
             //Unregister from LibUSB processor so that it auto-stops LibUSB processing
-            TunerManager.LIBUSB_TRANSFER_PROCESSOR.registerTransferProcessor(this);
+            TunerManager.LIBUSB_TRANSFER_PROCESSOR.unregisterTransferProcessor(this);
 
             //Directly invoke the timeout handler to ensure that our cancelled transfer buffers are flushed.
             int result = LibUsb.handleEventsTimeoutCompleted(null, USB_TIMEOUT_MS,
@@ -414,11 +414,13 @@ public class USBTransferProcessor implements TransferCallback
 
         @Override
         public void run()
+        {
+            mFilledBuffers.drainTo(mBuffersToDispatch, 50);
+
+            while(!mBuffersToDispatch.isEmpty())
             {
                 try
                 {
-                mFilledBuffers.drainTo(mBuffersToDispatch, 15);
-
                     for(byte[] buffer : mBuffersToDispatch)
                     {
                         float[] complexSamples = mSampleAdapter.convert(buffer);
@@ -432,6 +434,8 @@ public class USBTransferProcessor implements TransferCallback
                 }
 
                 mBuffersToDispatch.clear();
+                mFilledBuffers.drainTo(mBuffersToDispatch, 50);
+            }
         }
     }
 }

src/source/wave/ComplexWaveSource.java

@@ -18,19 +18,9 @@
  ******************************************************************************/
 package source.wave;
 
-import java.io.File;
+import channel.heartbeat.Heartbeat;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.concurrent.ScheduledExecutorService;
-
-import javax.sound.sampled.AudioFormat;
-import javax.sound.sampled.AudioInputStream;
-import javax.sound.sampled.AudioSystem;
-import javax.sound.sampled.UnsupportedAudioFileException;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import sample.ConversionUtils;
 import sample.Listener;
 import sample.complex.ComplexBuffer;
@@ -39,6 +29,15 @@ import source.IControllableFileSource;
 import source.IFrameLocationListener;
 import source.tuner.frequency.FrequencyChangeEvent;
 
+import javax.sound.sampled.AudioFormat;
+import javax.sound.sampled.AudioInputStream;
+import javax.sound.sampled.AudioSystem;
+import javax.sound.sampled.UnsupportedAudioFileException;
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.ScheduledExecutorService;
+
 public class ComplexWaveSource extends ComplexSource implements IControllableFileSource
 {
     private final static Logger mLog =
@@ -76,6 +75,22 @@ 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()
     {

src/source/wave/RealWaveSource.java

@@ -17,19 +17,9 @@
  ******************************************************************************/
 package source.wave;
 
-import java.io.File;
+import channel.heartbeat.Heartbeat;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.concurrent.ScheduledExecutorService;
-
-import javax.sound.sampled.AudioFormat;
-import javax.sound.sampled.AudioInputStream;
-import javax.sound.sampled.AudioSystem;
-import javax.sound.sampled.UnsupportedAudioFileException;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import sample.ConversionUtils;
 import sample.Listener;
 import sample.real.RealBuffer;
@@ -38,11 +28,18 @@ import source.IFrameLocationListener;
 import source.RealSource;
 import source.tuner.frequency.FrequencyChangeEvent;
 
-public class RealWaveSource extends RealSource 
+import javax.sound.sampled.AudioFormat;
-	implements IControllableFileSource, AutoCloseable
+import javax.sound.sampled.AudioInputStream;
+import javax.sound.sampled.AudioSystem;
+import javax.sound.sampled.UnsupportedAudioFileException;
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.ScheduledExecutorService;
+
+public class RealWaveSource extends RealSource implements IControllableFileSource, AutoCloseable
 {
-	private final static Logger mLog = 
+    private final static Logger mLog = LoggerFactory.getLogger(RealWaveSource.class);
-			LoggerFactory.getLogger( RealWaveSource.class );
 
     private IFrameLocationListener mFrameLocationListener;
     private int mBytesPerFrame;
@@ -76,6 +73,22 @@ public class RealWaveSource extends RealSource
         return null;
     }
 
+    /**
+     * Not implemented
+     */
+    @Override
+    public void setHeartbeatListener(Listener<Heartbeat> listener)
+    {
+    }
+
+    /**
+     * Not implemented
+     */
+    @Override
+    public void removeHeartbeatListener()
+    {
+    }
+
 
     @Override
     public void reset()