Merge pull request #142 from DSheirer/126-tuner-buffer-processor-out-of-memory

126 - Tuner processing refactor and significant improvement in processor and memory utilization.  R…

src/sample/OverflowableTransferQueue.java

@@ -0,0 +1,127 @@
+/*******************************************************************************
+ * 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 sample;
+
+import java.util.Collection;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class OverflowableTransferQueue<E>
+{
+    public enum State {NORMAL, OVERFLOW};
+    private Listener<State> mStateListener;
+
+    private LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<E>();
+    private AtomicInteger mCounter = new AtomicInteger();
+    private int mMaximumSize;
+    private int mResetThreshold;
+    private boolean mOverflow = false;
+
+    /**
+     * Concurrent transfer queue that couples a higher-throughput linked transfer queue with an atomic integer for
+     * monitoring queue size.  When the queue size exceeds maximum size (overflow), all inbound elements are ignored
+     * until the queue size is reduced to or below the reset threshold.
+     *
+     * @param maximumSize of the queue.  Overflow state will occur once queue size exceeds this value.
+     * @param resetThreshold for resetting overflow state to normal, once queue size is at or below this value.
+     */
+    public OverflowableTransferQueue(int maximumSize, int resetThreshold)
+    {
+        mMaximumSize = maximumSize;
+        mResetThreshold = resetThreshold;
+    }
+
+    /**
+     * Adds the element to the queue if able to do so without exceeding maximum queue size.  Otherwise, ignores
+     * the element.
+     */
+    public void offer(E e)
+    {
+        if(!mOverflow)
+        {
+            if(mCounter.get() > mMaximumSize)
+            {
+                setOverflow(true);
+            }
+            else
+            {
+                mQueue.offer(e);
+                mCounter.incrementAndGet();
+            }
+        }
+    }
+
+    /**
+     * Retrieves elements from the queue into the collection up to the maximum number of elements specified
+     */
+    public int drainTo(Collection<? super E> collection, int maxElements)
+    {
+        int drainCount = mQueue.drainTo(collection, maxElements);
+
+        if(drainCount > 0)
+        {
+            mCounter.addAndGet(-drainCount);
+        }
+
+        return drainCount;
+    }
+
+    /**
+     * Sets a listener to receive state change events
+     */
+    public void setStateListener(Listener<State> listener)
+    {
+        mStateListener = listener;
+    }
+
+
+    /**
+     * Current state of the queue, normal or overflow
+     */
+    public State getState()
+    {
+        return mOverflow ? State.OVERFLOW : State.NORMAL;
+    }
+
+    /**
+     * Toggles the overflow state
+     */
+    private void setOverflow(boolean overflow)
+    {
+        mOverflow = overflow;
+
+        if(mStateListener != null)
+        {
+            mStateListener.receive(getState());
+        }
+    }
+
+    /**
+     * Clears all elements from the queue and resets the internal counter to 0
+     */
+    public void clear()
+    {
+        synchronized(mQueue)
+        {
+            mQueue.clear();
+            mCounter.set(0);
+            mOverflow = false;
+        }
+    }
+}

src/source/tuner/LibUSBTransferProcessor.java

@@ -0,0 +1,125 @@
+/*******************************************************************************
+ * 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 org.usb4java.LibUsb;
+import source.tuner.usb.USBTransferProcessor;
+import util.ThreadPool;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Runnable for executing LibUsb processing of bulk transfer buffers
+ */
+public class LibUSBTransferProcessor
+{
+    private final static Logger mLog = LoggerFactory.getLogger(LibUSBTransferProcessor.class);
+
+    private static final long USB_TIMEOUT_MS = 2000l; //milliseconds
+
+    private List<USBTransferProcessor> mRegisteredProcessors = new CopyOnWriteArrayList<>();
+    private ByteBuffer mStatusBuffer = ByteBuffer.allocateDirect(4);
+    private AtomicBoolean mRunning = new AtomicBoolean();
+    private Processor mProcessor = new Processor();
+    private ScheduledFuture mProcessorFuture;
+
+    /**
+     * Registers the transfer processor so that LibUSB timeout processing will auto-start.
+     * @param processor to register
+     */
+    public void registerTransferProcessor(USBTransferProcessor processor)
+    {
+        if(!mRegisteredProcessors.contains(processor))
+        {
+            mRegisteredProcessors.add(processor);
+        }
+
+        start();
+    }
+
+    /**
+     * Unregisters the transfer processor so that LibUSB timeout processing will auto-stop once all processors have
+     * been unregistered.
+     *
+     * @param processor to unregister
+     */
+    public void unregisterTransferProcessor(USBTransferProcessor processor)
+    {
+        mRegisteredProcessors.remove(processor);
+
+        if(mRegisteredProcessors.isEmpty())
+        {
+            stop();
+        }
+    }
+
+    /**
+     * Starts processing of LibUSB timeout events
+     */
+    private void start()
+    {
+        if(mRunning.compareAndSet(false, true))
+        {
+            //Set periodicity to an odd multiple to avoid contention with transfer buffer receivers
+            mProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(mProcessor, 0L, 9L, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /**
+     * Stops processing of LibUSB timeout events
+     */
+    private void stop()
+    {
+        if(mRunning.compareAndSet(true, false))
+        {
+            if(mProcessorFuture != null)
+            {
+                mProcessorFuture.cancel(true);
+                mProcessorFuture = null;
+            }
+        }
+    }
+
+    /**
+     * Runnable to invoke LibUSB timeout handler.  All downstream completed transfer processing will occur on this
+     * thread instance.
+     */
+    class Processor implements Runnable
+    {
+        @Override
+        public void run()
+        {
+            int result = LibUsb.handleEventsTimeoutCompleted(null, USB_TIMEOUT_MS, mStatusBuffer.asIntBuffer());
+
+            if(result != LibUsb.SUCCESS)
+            {
+                mLog.error("Error processing timeout events for LibUSB - error code:" + result);
+            }
+
+            mStatusBuffer.rewind();
+        }
+    }
+}

src/source/tuner/TunerChannelSource.java

@@ -1,6 +1,6 @@
 /*******************************************************************************
- *     SDR Trunk 
+ * sdrtrunk
- *     Copyright (C) 2014-2016 Dennis Sheirer
+ * 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
@@ -14,6 +14,7 @@
  *
  * 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;
 
@@ -25,6 +26,7 @@ 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.ComplexSource;
@@ -35,24 +37,27 @@ import source.tuner.frequency.IFrequencyChangeProcessor;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.LinkedTransferQueue;
 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 TunerChannelSource extends ComplexSource
+public class TunerChannelSource extends ComplexSource implements IFrequencyChangeProcessor, Listener<ComplexBuffer>
-    implements IFrequencyChangeProcessor, Listener<ComplexBuffer>
 {
-    private final static Logger mLog =
+    private final static Logger mLog = LoggerFactory.getLogger(TunerChannelSource.class);
-        LoggerFactory.getLogger(TunerChannelSource.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 = 200;
 
     private static int CHANNEL_RATE = 48000;
     private static int CHANNEL_PASS_FREQUENCY = 12000;
 
-    private LinkedTransferQueue<ComplexBuffer> mBuffer =
+    private OverflowableTransferQueue<ComplexBuffer> mBuffer;
-        new LinkedTransferQueue<ComplexBuffer>();
+
     private Tuner mTuner;
     private TunerChannel mTunerChannel;
     private Oscillator mMixer;
@@ -69,6 +74,7 @@ public class TunerChannelSource extends ComplexSource
 
     private AtomicBoolean mRunning = new AtomicBoolean();
     private boolean mExpended = false;
+    private int mCount;
 
     /**
      * Provides a Digital Drop Channel (DDC) to decimate the IQ output from a
@@ -87,26 +93,37 @@ public class TunerChannelSource extends ComplexSource
      *                                    accept the decimation processing task
      * @throws SourceException            if the tuner has an issue providing IQ samples
      */
-    public TunerChannelSource(Tuner tuner, TunerChannel tunerChannel)
+    public TunerChannelSource(Tuner tuner, TunerChannel tunerChannel) throws RejectedExecutionException, SourceException
-        throws RejectedExecutionException, SourceException
     {
         mTuner = tuner;
-        mTuner.getTunerController().addListener((IFrequencyChangeProcessor) this);
-
         mTunerChannel = tunerChannel;
-
+        mTuner.getTunerController().addListener((IFrequencyChangeProcessor) this);
         mTunerFrequency = mTuner.getTunerController().getFrequency();
 
+        mBuffer = new OverflowableTransferQueue<>(BUFFER_MAX_CAPACITY, BUFFER_OVERFLOW_RESET_THRESHOLD);
+        mBuffer.setStateListener(new Listener<OverflowableTransferQueue.State>()
+        {
+            @Override
+            public void receive(OverflowableTransferQueue.State state)
+            {
+                if(state == OverflowableTransferQueue.State.NORMAL)
+                {
+                    mLog.debug("Channel [" + mTunerFrequency + "] buffer overflow - temporary pause until processing catches up");
+                }
+                else
+                {
+                    mLog.debug("Channel [" + mTunerFrequency + "] buffer overflow cleared - resuming normal processing");
+                }
+            }
+        });
+
 	    /* Setup the frequency translator to the current source frequency */
         long frequencyOffset = mTunerFrequency - mTunerChannel.getFrequency();
 
-        mMixer = new Oscillator(frequencyOffset,
+        mMixer = new Oscillator(frequencyOffset, mTuner.getTunerController().getSampleRate());
-            mTuner.getTunerController().getSampleRate());
 
 		/* Fire a sample rate change event to setup the decimation chain */
-        frequencyChanged(new FrequencyChangeEvent(
+        frequencyChanged(new FrequencyChangeEvent(Event.NOTIFICATION_SAMPLE_RATE_CHANGE, mTuner.getTunerController().getSampleRate()));
-            Event.NOTIFICATION_SAMPLE_RATE_CHANGE,
-            mTuner.getTunerController().getSampleRate()));
     }
 
 
@@ -114,23 +131,22 @@ public class TunerChannelSource extends ComplexSource
     {
         if(mExpended)
         {
-            throw new IllegalStateException("Attempt to re-start an expended "
+            throw new IllegalStateException("Attempt to re-start an expended tuner channel source.  TunerChannelSource" +
-                + "tuner channel source.  TunerChannelSource objects can only "
+                " objects can only be used once. ");
-                + "be used once. ");
         }
 
         if(mRunning.compareAndSet(false, true))
         {
-			/* Schedule the decimation task to run every 10 ms (100 iterations/second) */
+            //Schedule the decimation task to run every 9 ms (111 iterations/second), an odd periodicity relative
-            mTaskHandle = executor.scheduleAtFixedRate(mDecimationProcessor,
+            //to the inbound periodicity of 20 ms, to attempt to avoid thread queue contention
-                0, 10, TimeUnit.MILLISECONDS);
+            mTaskHandle = executor.scheduleAtFixedRate(mDecimationProcessor, 0, 9, TimeUnit.MILLISECONDS);
 
 		    /* Finally, register to receive samples from the tuner */
             mTuner.addListener((Listener<ComplexBuffer>) this);
         }
         else
         {
-            mLog.warn("Attempt to start() and already running tuner channel source was ignored");
+            mLog.warn("Attempt to start() an already running tuner channel source was ignored");
         }
     }
 
@@ -171,14 +187,6 @@ public class TunerChannelSource extends ComplexSource
         {
 	    	/* Tell the tuner to release/unregister our resources */
             mTuner.getTunerController().removeListener(this);
-//            mTuner = null;
-//            mTunerChannel = null;
-//            mBuffer = null;
-//            mFrequencyChangeProcessor = null;
-//            mListener = null;
-//            mMixer = null;
-//            mDecimationFilter.dispose();
-//            mDecimationFilter = null;
         }
     }
 
@@ -197,7 +205,7 @@ public class TunerChannelSource extends ComplexSource
     {
         if(mRunning.get())
         {
-            mBuffer.add(buffer);
+            mBuffer.offer(buffer);
         }
     }
 
@@ -248,10 +256,8 @@ public class TunerChannelSource extends ComplexSource
 
                 if(mFrequencyChangeProcessor != null)
                 {
-                    mFrequencyChangeProcessor.frequencyChanged(
+                    mFrequencyChangeProcessor.frequencyChanged(new FrequencyChangeEvent(
-                        new FrequencyChangeEvent(
+                        Event.NOTIFICATION_CHANNEL_FREQUENCY_CORRECTION_CHANGE, mChannelFrequencyCorrection));
-                            Event.NOTIFICATION_CHANNEL_FREQUENCY_CORRECTION_CHANGE,
-                            mChannelFrequencyCorrection));
                 }
                 break;
             case NOTIFICATION_SAMPLE_RATE_CHANGE:
@@ -262,13 +268,8 @@ public class TunerChannelSource extends ComplexSource
                     mMixer.setSampleRate(sampleRate);
 
 					/* Get new decimation filter */
-                    mDecimationFilter = FilterFactory
+                    mDecimationFilter = FilterFactory.getDecimationFilter(sampleRate, CHANNEL_RATE, 1,
-                        .getDecimationFilter(sampleRate,
+                        CHANNEL_PASS_FREQUENCY, 60, WindowType.HAMMING);
-                            CHANNEL_RATE,
-                            1,   //Order
-                            CHANNEL_PASS_FREQUENCY,
-                            60, //dB attenuation
-                            WindowType.HAMMING);
 					
 					/* re-add the original output listener */
                     mDecimationFilter.setListener(mListener);
@@ -328,8 +329,7 @@ public class TunerChannelSource extends ComplexSource
             {
                 if(mProcessing)
                 {
-                    {
+                    mBuffer.drainTo(mSampleBuffers, 20);
-                        mBuffer.drainTo(mSampleBuffers, 8);
 
                     for(Buffer buffer : mSampleBuffers)
                     {
@@ -371,7 +371,6 @@ public class TunerChannelSource extends ComplexSource
                     mSampleBuffers.clear();
                 }
             }
-            }
             catch(Exception e)
             {
 				/* Only log the stack trace if we're still processing */

src/source/tuner/TunerManager.java

@@ -42,12 +42,24 @@ import java.util.Collection;
 
 public class TunerManager
 {
-    private final static Logger mLog =
+    private final static Logger mLog = LoggerFactory.getLogger(TunerManager.class);
-        LoggerFactory.getLogger(TunerManager.class);
 
     private MixerManager mMixerManager;
     private TunerModel mTunerModel;
 
+
+    /**
+     * Application-wide LibUSB timeout processor for transfer buffers.  All classes that need to use USB transfer
+     * buffers can register with this processor and the processor will auto-start and auto-stop while USB transfer
+     * processors are registered.
+     */
+    public static final LibUSBTransferProcessor LIBUSB_TRANSFER_PROCESSOR;
+
+    static
+    {
+        LIBUSB_TRANSFER_PROCESSOR = new LibUSBTransferProcessor();
+    }
+
     public TunerManager(MixerManager mixerManager, TunerModel tunerModel)
     {
         mMixerManager = mixerManager;

src/source/tuner/airspy/AirspySampleAdapter.java

@@ -1,8 +1,6 @@
-package source.tuner.airspy;
-
 /*******************************************************************************
- *     SDR Trunk 
+ * sdrtrunk
- *     Copyright (C) 2015 Dennis Sheirer
+ * 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
@@ -16,16 +14,23 @@ package source.tuner.airspy;
  *
  * 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.airspy;
 
+import dsp.filter.dc.DCRemovalFilter_RB;
+import dsp.filter.hilbert.HilbertTransform;
 import sample.adapter.ISampleAdapter;
 
 public class AirspySampleAdapter implements ISampleAdapter
 {
     private static final float SCALE_SIGNED_12_BIT_TO_FLOAT = 1.0f / 2048.0f;
 
+    private DCRemovalFilter_RB mDCFilter = new DCRemovalFilter_RB(0.01f);
+    private HilbertTransform mHilbertTransform = new HilbertTransform();
     private boolean mSamplePacking = false;
 
+
     /**
      * Adapter to translate byte buffers received from the airspy tuner into
      * float buffers for processing.
@@ -48,14 +53,20 @@ public class AirspySampleAdapter implements ISampleAdapter
     @Override
     public float[] convert(byte[] samples)
     {
+        float[] realSamples;
+
         if(mSamplePacking)
         {
-			return convertPacked( samples );
+            realSamples = convertPacked(samples);
         }
         else
         {
-			return convertUnpacked( samples );
+            realSamples = convertUnpacked(samples);
         }
+
+        mDCFilter.filter(realSamples);
+
+        return mHilbertTransform.filter(realSamples);
     }
 
     /**
@@ -84,7 +95,7 @@ public class AirspySampleAdapter implements ISampleAdapter
      * Converts every 3 bytes containing a pair of 12-bit unsigned values into
      * a pair of float values in the range -1 to 1;
      *
-	 * @param data - byte array of unsigned 12-bit values
+     * @param data1 - byte array of unsigned 12-bit values
      * @return converted float values
      */
     private float[] convertPacked(byte[] data1)

src/source/tuner/airspy/AirspyTunerController.java

@@ -17,6 +17,7 @@ import sample.complex.ComplexBuffer;
 import source.SourceException;
 import source.tuner.TunerController;
 import source.tuner.configuration.TunerConfiguration;
+import source.tuner.usb.USBTransferProcessor;
 import util.ThreadPool;
 
 import javax.usb.UsbException;
@@ -93,21 +94,11 @@ public class AirspyTunerController extends TunerController
     public static final AirspySampleRate DEFAULT_SAMPLE_RATE =
         new AirspySampleRate(0, 10000000, "10.00 MHz");
 
-    public static final long USB_TIMEOUT_MS = 2000l; //milliseconds
+    private static final long USB_TIMEOUT_MS = 2000l; //milliseconds
-    public static final byte USB_ENDPOINT = (byte) 0x81;
+    private static final byte USB_INTERFACE = (byte) 0x0;
-    public static final byte USB_INTERFACE = (byte) 0x0;
+    private static final int USB_TRANSFER_BUFFER_SIZE = 262144;
-    public static final int TRANSFER_BUFFER_POOL_SIZE = 16;
+    private static final byte USB_REQUEST_IN = (byte) (LibUsb.ENDPOINT_IN | LibUsb.REQUEST_TYPE_VENDOR | LibUsb.RECIPIENT_DEVICE);
-
+    private static final byte USB_REQUEST_OUT = (byte) (LibUsb.ENDPOINT_OUT | LibUsb.REQUEST_TYPE_VENDOR | LibUsb.RECIPIENT_DEVICE);
-    public static final byte USB_REQUEST_IN =
-        (byte) (LibUsb.ENDPOINT_IN |
-            LibUsb.REQUEST_TYPE_VENDOR |
-            LibUsb.RECIPIENT_DEVICE);
-
-    public static final byte USB_REQUEST_OUT =
-        (byte) (LibUsb.ENDPOINT_OUT |
-            LibUsb.REQUEST_TYPE_VENDOR |
-            LibUsb.RECIPIENT_DEVICE);
-
     public static final DecimalFormat MHZ_FORMATTER = new DecimalFormat("#.00 MHz");
 
     private final static Logger mLog = LoggerFactory.getLogger(AirspyTunerController.class);
@@ -115,17 +106,11 @@ public class AirspyTunerController extends TunerController
     private Device mDevice;
     private DeviceHandle mDeviceHandle;
 
-    private LinkedTransferQueue<byte[]> mFilledBuffers = new LinkedTransferQueue<byte[]>();
-    private Broadcaster<ComplexBuffer> mComplexBufferBroadcaster = new Broadcaster<>();
-    private int mBufferSize = 262144;
-    private BufferProcessor mBufferProcessor = new BufferProcessor();
     private AirspySampleAdapter mSampleAdapter = new AirspySampleAdapter();
-    private DCRemovalFilter_RB mDCFilter = new DCRemovalFilter_RB(0.01f);
-    private HilbertTransform mHilbertTransform = new HilbertTransform();
-
     private AirspyDeviceInformation mDeviceInfo;
     private List<AirspySampleRate> mSampleRates = new ArrayList<>();
     private int mSampleRate = 0;
+    private USBTransferProcessor mUSBTransferProcessor;
 
     public AirspyTunerController(Device device) throws SourceException
     {
@@ -140,7 +125,7 @@ public class AirspyTunerController extends TunerController
 
         int result = LibUsb.open(mDevice, mDeviceHandle);
 
-        if(result != 0)
+        if(result != LibUsb.SUCCESS)
         {
             if(result == LibUsb.ERROR_ACCESS)
             {
@@ -201,6 +186,11 @@ public class AirspyTunerController extends TunerController
         {
             mLog.error("Setting sample rate is not supported by firmware", e);
         }
+
+        String deviceName = "Airspy " + getDeviceInfo().getSerialNumber();
+
+        mUSBTransferProcessor = new USBTransferProcessor(deviceName, mDeviceHandle, mSampleAdapter,
+            USB_TRANSFER_BUFFER_SIZE);
     }
 
     /**
@@ -1205,250 +1195,32 @@ public class AirspyTunerController extends TunerController
     }
 
     /**
-     * Adds a sample listener.  If the buffer processing thread is
+     * Adds the IQ buffer listener and automatically starts buffer transfer processing, if not already started.
-     * not currently running, starts it running in a new thread.
      */
     public void addListener(Listener<ComplexBuffer> listener)
     {
-        synchronized(mComplexBufferBroadcaster)
+        if(mUSBTransferProcessor != null)
         {
-            mComplexBufferBroadcaster.addListener(listener);
+            mUSBTransferProcessor.addListener(listener);
-
-            if(mBufferProcessor == null || !mBufferProcessor.isRunning())
-            {
-                mBufferProcessor = new BufferProcessor();
-
-                Thread thread = new Thread(mBufferProcessor);
-                thread.setDaemon(true);
-                thread.setName("Airspy Buffer Processor");
-
-                thread.start();
-            }
-        }
-    }
-
-    /**
-     * Removes the sample listener.  If this is the last registered listener,
-     * shuts down the buffer processing thread.
-     */
-    public void removeListener(Listener<ComplexBuffer> listener)
-    {
-        synchronized(mComplexBufferBroadcaster)
-        {
-            mComplexBufferBroadcaster.removeListener(listener);
-
-            if(!mComplexBufferBroadcaster.hasListeners())
-            {
-                mBufferProcessor.stop();
-            }
-        }
-    }
-
-    /**
-     * Buffer processing thread.  Fetches samples from the Airspy Tuner and
-     * dispatches them to all registered listeners
-     */
-    public class BufferProcessor implements Runnable, TransferCallback
-    {
-        private ScheduledFuture<?> mSampleDispatcherTask;
-        private LinkedTransferQueue<Transfer> mAvailableTransfers;
-        private LinkedTransferQueue<Transfer> mTransfersInProgress = new LinkedTransferQueue<>();
-        private AtomicBoolean mRunning = new AtomicBoolean();
-        private ByteBuffer mLibUsbHandlerStatus;
-        private boolean mCancel = false;
-
-        @Override
-        public void run()
-        {
-            if(mRunning.compareAndSet(false, true))
-            {
-                prepareTransfers();
-
-                mSampleDispatcherTask = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferDispatcher(),
-                    20, 20, TimeUnit.MILLISECONDS);
-
-                mLibUsbHandlerStatus = ByteBuffer.allocateDirect(4);
-
-                List<Transfer> transfers = new ArrayList<>();
-
-                mCancel = false;
-
-                while(mRunning.get())
-                {
-                    mAvailableTransfers.drainTo(transfers);
-
-                    for(Transfer transfer : transfers)
-                    {
-                        int result = LibUsb.submitTransfer(transfer);
-
-                        if(result == LibUsb.SUCCESS)
-                        {
-                            mTransfersInProgress.add(transfer);
         }
         else
         {
-                            mLog.error("error submitting transfer [" + LibUsb.errorName(result) + "]");
+            mLog.error("Couldn't add IQ buffer listener to Airspy tuner - processor is null");
-                        }
-                    }
-
-                    int result = LibUsb.handleEventsTimeoutCompleted(
-                        null, USB_TIMEOUT_MS, mLibUsbHandlerStatus.asIntBuffer());
-
-                    if(result != LibUsb.SUCCESS)
-                    {
-                        mLog.error("error handling events for libusb");
-                    }
-
-                    transfers.clear();
-
-                    mLibUsbHandlerStatus.rewind();
-                }
-
-                if(mCancel)
-                {
-                    for(Transfer transfer : mTransfersInProgress)
-                    {
-                        LibUsb.cancelTransfer(transfer);
-                    }
-
-                    int result = LibUsb.handleEventsTimeoutCompleted(null,
-                        USB_TIMEOUT_MS, mLibUsbHandlerStatus.asIntBuffer());
-
-                    if(result != LibUsb.SUCCESS)
-                    {
-                        mLog.error("error handling events for libusb during cancel");
-                    }
-
-                    mLibUsbHandlerStatus.rewind();
-                }
-            }
-        }
-
-        /**
-         * Stops the sample fetching thread
-         */
-        public void stop()
-        {
-            mCancel = true;
-
-            if(mRunning.compareAndSet(true, false))
-            {
-                if(mSampleDispatcherTask != null)
-                {
-                    mSampleDispatcherTask.cancel(true);
-                    mFilledBuffers.clear();
-                }
-            }
-        }
-
-        /**
-         * Indicates if this thread is running
-         */
-        public boolean isRunning()
-        {
-            return mRunning.get();
-        }
-
-        @Override
-        public void processTransfer(Transfer transfer)
-        {
-            mTransfersInProgress.remove(transfer);
-
-            switch(transfer.status())
-            {
-                case LibUsb.TRANSFER_COMPLETED:
-                case LibUsb.TRANSFER_STALL:
-                    if(transfer.actualLength() > 0)
-                    {
-                        ByteBuffer buffer = transfer.buffer();
-
-                        byte[] data = new byte[transfer.actualLength()];
-
-                        buffer.get(data);
-
-                        buffer.rewind();
-
-                        if(isRunning())
-                        {
-                            mFilledBuffers.add(data);
-                        }
-                    }
-                    break;
-                case LibUsb.TRANSFER_CANCELLED:
-                    break;
-                default:
-					/* unexpected error */
-                    mLog.error("transfer error [" +
-                        getTransferStatus(transfer.status()) +
-                        "] transferred actual: " + transfer.actualLength());
-            }
-
-            mAvailableTransfers.add(transfer);
-        }
-
-        /**
-         * Prepares (allocates) a set of transfer buffers for use in
-         * transferring data from the tuner via the bulk interface
-         */
-        private void prepareTransfers() throws LibUsbException
-        {
-            if(mAvailableTransfers == null)
-            {
-                mAvailableTransfers = new LinkedTransferQueue<>();
-
-                for(int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++)
-                {
-                    Transfer transfer = LibUsb.allocTransfer();
-
-                    if(transfer == null)
-                    {
-                        throw new LibUsbException("couldn't allocate transfer",
-                            LibUsb.ERROR_NO_MEM);
-                    }
-
-                    final ByteBuffer buffer =
-                        ByteBuffer.allocateDirect(mBufferSize);
-
-                    LibUsb.fillBulkTransfer(transfer, mDeviceHandle, USB_ENDPOINT,
-                        buffer, BufferProcessor.this, "Buffer", USB_TIMEOUT_MS);
-
-                    mAvailableTransfers.add(transfer);
-                }
-            }
         }
     }
 
     /**
-     * Fetches byte[] chunks from the raw sample buffer.  Converts each byte
+     * Removes the IQ buffer listener and stops buffer transfer processing if there are no more listeners.
-     * array and broadcasts the array to all registered listeners
      */
-    public class BufferDispatcher implements Runnable
+    public void removeListener(Listener<ComplexBuffer> listener)
     {
-        @Override
+        if(mUSBTransferProcessor != null)
-        public void run()
         {
-            try
+            mUSBTransferProcessor.removeListener(listener);
+        }
+        else
         {
-                ArrayList<byte[]> buffers = new ArrayList<>();
+            mLog.error("Couldn't remove IQ buffer listener from Airspy tuner - processor is null");
-
-                mFilledBuffers.drainTo(buffers);
-
-                for(byte[] buffer : buffers)
-                {
-                    float[] realSamples = mSampleAdapter.convert(buffer);
-
-                    realSamples = mDCFilter.filter(realSamples);
-
-                    float[] quadratureSamples = mHilbertTransform.filter(realSamples);
-
-                    mComplexBufferBroadcaster.broadcast(
-                        new ComplexBuffer(quadratureSamples));
-                }
-            }
-            catch(Exception e)
-            {
-                mLog.error("error during Airspy buffer dispatching", e);
-            }
         }
     }
 }

src/source/tuner/fcd/FCDTunerController.java

@@ -1,6 +1,6 @@
 /*******************************************************************************
- *     SDR Trunk 
+ * sdrtrunk
- *     Copyright (C) 2014 Dennis Sheirer
+ * 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
@@ -14,16 +14,10 @@
  *
  * 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.fcd;
 
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.nio.IntBuffer;
-
-import javax.usb.UsbClaimException;
-import javax.usb.UsbException;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.usb4java.Device;
@@ -31,7 +25,6 @@ import org.usb4java.DeviceDescriptor;
 import org.usb4java.DeviceHandle;
 import org.usb4java.LibUsb;
 import org.usb4java.LibUsbException;
-
 import source.SourceException;
 import source.tuner.TunerClass;
 import source.tuner.TunerController;
@@ -39,10 +32,15 @@ import source.tuner.TunerType;
 import source.tuner.configuration.TunerConfiguration;
 import source.tuner.fcd.proV1.FCD1TunerController.Block;
 
+import javax.usb.UsbClaimException;
+import javax.usb.UsbException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.IntBuffer;
+
 public abstract class FCDTunerController extends TunerController
 {
-	private final static Logger mLog = 
+    private final static Logger mLog = LoggerFactory.getLogger(FCDTunerController.class);
-			LoggerFactory.getLogger( FCDTunerController.class );
 
     public static final double USABLE_BANDWIDTH_PERCENT = 1.0;
     public static final int DC_SPIKE_AVOID_BUFFER = 5000;
@@ -72,10 +70,7 @@ public abstract class FCDTunerController extends TunerController
                               int minTunableFrequency,
                               int maxTunableFrequency)
     {
-		super( minTunableFrequency, 
+        super(minTunableFrequency, maxTunableFrequency, DC_SPIKE_AVOID_BUFFER, USABLE_BANDWIDTH_PERCENT);
-			   maxTunableFrequency, 
-			   DC_SPIKE_AVOID_BUFFER, 
-			   USABLE_BANDWIDTH_PERCENT );
         mDevice = device;
         mDeviceDescriptor = descriptor;
 
@@ -108,8 +103,7 @@ public abstract class FCDTunerController extends TunerController
         {
             mDeviceHandle = null;
 
-			throw new SourceException( "libusb couldn't open funcube usb "
+            throw new SourceException("libusb couldn't open funcube usb device [" + LibUsb.errorName(result) + "]");
-					+ "device [" + LibUsb.errorName( result ) + "]" );
         }
 
         claimInterface();
@@ -190,13 +184,9 @@ public abstract class FCDTunerController extends TunerController
         {
             StringBuilder sb = new StringBuilder();
 
-			sb.append( String.format( "%04X", 
+            sb.append(String.format("%04X", (int)(mDeviceDescriptor.idVendor() & 0xFFFF)));
-					(int)( mDeviceDescriptor.idVendor() & 0xFFFF ) ) );
-
             sb.append(":");
-			
+            sb.append(String.format("%04X", (int)(mDeviceDescriptor.idProduct() & 0xFFFF)));
-			sb.append( String.format( "%04X", 
-					(int)( mDeviceDescriptor.idProduct() & 0xFFFF ) ) );
 
             return sb.toString();
         }
@@ -315,13 +305,11 @@ public abstract class FCDTunerController extends TunerController
 
             if(result == 1)
             {
-				result = LibUsb.detachKernelDriver( mDeviceHandle, 
+                result = LibUsb.detachKernelDriver(mDeviceHandle, FCD_INTERFACE);
-						FCD_INTERFACE );
 
                 if(result != LibUsb.SUCCESS)
                 {
-					mLog.error( "failed attempt to detach kernel driver [" + 
+                    mLog.error("failed attempt to detach kernel driver [" + LibUsb.errorName(result) + "]");
-					LibUsb.errorName( result ) + "]" );
                 }
             }
 
@@ -329,21 +317,19 @@ public abstract class FCDTunerController extends TunerController
 
             if(result != LibUsb.SUCCESS)
             {
-				throw new SourceException( "couldn't claim usb interface [" + 
+                throw new SourceException("couldn't claim usb interface [" + LibUsb.errorName(result) + "]");
-					LibUsb.errorName( result ) + "]" );
             }
         }
         else
         {
-			throw new SourceException( "couldn't claim usb hid interface - no "
+            throw new SourceException("couldn't claim usb hid interface - no device handle");
-					+ "device handle" );
         }
     }
 
     /**
      * Performs an interrupt write to the OUT endpoint.
-	 * @param buffer - direct allocated buffer.  Must be 64 bytes in length.
      *
+     * @param buffer - direct allocated buffer.  Must be 64 bytes in length.
      * @throws LibUsbException on error
      */
     private void write(ByteBuffer buffer) throws LibUsbException
@@ -352,33 +338,22 @@ public abstract class FCDTunerController extends TunerController
         {
             IntBuffer transferred = IntBuffer.allocate(1);
 
-			int result = LibUsb.interruptTransfer( mDeviceHandle,
+            int result = LibUsb.interruptTransfer(mDeviceHandle, FCD_ENDPOINT_OUT, buffer, transferred, 500l);
-												   FCD_ENDPOINT_OUT,
-												   buffer, 
-												   transferred,
-												   500l );
 
             if(result != LibUsb.SUCCESS)
             {
-				throw new LibUsbException( "error writing byte buffer", 
+                throw new LibUsbException("error writing byte buffer", result);
-								result );
             }
-//			else if( transferred.get() != buffer.capacity() )
-//			{
-//				throw new LibUsbException( "transferred bytes [" + 
-//						transferred.get( 0 ) + "] is not what was expected [" + 
-//						buffer.capacity() + "]", result );
-//			}
         }
         else
         {
-			throw new LibUsbException( "device handle is null", 
+            throw new LibUsbException("device handle is null", LibUsb.ERROR_NO_DEVICE);
-							LibUsb.ERROR_NO_DEVICE );
         }
     }
 
     /**
      * Performs an interrupt write to the OUT endpoint for the FCD command.
+     *
      * @param command - no-argument command to write
      * @throws LibUsbException - on error
      */
@@ -418,6 +393,7 @@ public abstract class FCDTunerController extends TunerController
 
     /**
      * Performs an interrupt write to the OUT endpoint for the FCD command.
+     *
      * @param command - command to write
      * @param argument - value to write with the command
      * @throws LibUsbException - on error
@@ -437,6 +413,7 @@ public abstract class FCDTunerController extends TunerController
 
     /**
      * Performs an interrupt read against the endpoint
+     *
      * @return buffer read from FCD
      * @throws LibUsbException on error
      */
@@ -449,29 +426,22 @@ public abstract class FCDTunerController extends TunerController
             IntBuffer transferred = IntBuffer.allocate(1);
 
 
-			int result = LibUsb.interruptTransfer( mDeviceHandle,
+            int result = LibUsb.interruptTransfer(mDeviceHandle, FCD_ENDPOINT_IN, buffer, transferred, 500l);
-												   FCD_ENDPOINT_IN,
-												   buffer, 
-												   transferred,
-												   500l );
 
             if(result != LibUsb.SUCCESS)
             {
-				throw new LibUsbException( "error reading byte buffer", 
+                throw new LibUsbException("error reading byte buffer", result);
-								result );
             }
             else if(transferred.get(0) != buffer.capacity())
             {
-				throw new LibUsbException( "received bytes [" + 
+                throw new LibUsbException("received bytes [" + transferred.get(0) +
-						transferred.get( 0 ) + "] didn't match expected "
+                    "] didn't match expected length [" + buffer.capacity() + "]", result);
-						+ "length [" + buffer.capacity() + "]", result );
             }
 
             return buffer;
         }
 
-		throw new LibUsbException( "device handle is null", 
+        throw new LibUsbException("device handle is null", LibUsb.ERROR_NO_DEVICE);
-				LibUsb.ERROR_NO_DEVICE );
     }
 
     /**

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

@@ -1,6 +1,6 @@
 /*******************************************************************************
- *     SDR Trunk 
+ * sdrtrunk
- *     Copyright (C) 2014-2016 Dennis Sheirer
+ * 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
@@ -14,17 +14,14 @@
  *
  * 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.fcd.proplusV2;
 
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.usb4java.Device;
 import org.usb4java.DeviceDescriptor;
-
 import source.SourceException;
 import source.tuner.MixerTunerType;
 import source.tuner.TunerClass;
@@ -33,6 +30,9 @@ import source.tuner.configuration.TunerConfiguration;
 import source.tuner.fcd.FCDCommand;
 import source.tuner.fcd.FCDTunerController;
 
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
 public class FCD2TunerController extends FCDTunerController
 {
     private final static Logger mLog = LoggerFactory.getLogger(FCD2TunerController.class);

src/source/tuner/hackrf/HackRFTunerController.java

@@ -1,6 +1,6 @@
 /*******************************************************************************
  *     SDR Trunk 
- *     Copyright (C) 2014 Dennis Sheirer
+ *     Copyright (C) 2014-2017 Dennis Sheirer
  *
  *     Ported from libhackrf at:
  *     https://github.com/mossmann/hackrf/tree/master/host/libhackrf
@@ -61,8 +61,6 @@ import org.usb4java.DeviceDescriptor;
 import org.usb4java.DeviceHandle;
 import org.usb4java.LibUsb;
 import org.usb4java.LibUsbException;
-import org.usb4java.Transfer;
-import org.usb4java.TransferCallback;
 import sample.Broadcaster;
 import sample.Listener;
 import sample.adapter.ByteSampleAdapter;
@@ -70,38 +68,25 @@ import sample.complex.ComplexBuffer;
 import source.SourceException;
 import source.tuner.TunerController;
 import source.tuner.configuration.TunerConfiguration;
-import util.ThreadPool;
+import source.tuner.usb.USBTransferProcessor;
 
 import javax.usb.UsbException;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.List;
 import java.util.concurrent.LinkedTransferQueue;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
 
 public class HackRFTunerController extends TunerController
 {
-    private final static Logger mLog =
+    private final static Logger mLog = LoggerFactory.getLogger(HackRFTunerController.class);
-        LoggerFactory.getLogger(HackRFTunerController.class);
 
     public final static long USB_TIMEOUT_US = 1000000l; //uSeconds
     public static final byte USB_ENDPOINT = (byte)0x81;
     public static final byte USB_INTERFACE = (byte)0x0;
+    public static final int USB_TRANSFER_BUFFER_SIZE = 262144;
 
-    public static final byte REQUEST_TYPE_IN =
+    public static final byte REQUEST_TYPE_IN = (byte)(LibUsb.ENDPOINT_IN | LibUsb.REQUEST_TYPE_VENDOR | LibUsb.RECIPIENT_DEVICE);
-        (byte) (LibUsb.ENDPOINT_IN |
+    public static final byte REQUEST_TYPE_OUT = (byte)(LibUsb.ENDPOINT_OUT | LibUsb.REQUEST_TYPE_VENDOR | LibUsb.RECIPIENT_DEVICE);
-            LibUsb.REQUEST_TYPE_VENDOR |
-            LibUsb.RECIPIENT_DEVICE);
-
-    public static final byte REQUEST_TYPE_OUT =
-        (byte) (LibUsb.ENDPOINT_OUT |
-            LibUsb.REQUEST_TYPE_VENDOR |
-            LibUsb.RECIPIENT_DEVICE);
 
     public static final long MIN_FREQUENCY = 10000000l;
     public static final long MAX_FREQUENCY = 6000000000l;
@@ -109,18 +94,10 @@ public class HackRFTunerController extends TunerController
     public static final double USABLE_BANDWIDTH_PERCENT = 0.95;
     public static final int DC_SPIKE_AVOID_BUFFER = 5000;
 
-    public final static int TRANSFER_BUFFER_POOL_SIZE = 16;
-
-    private LinkedTransferQueue<byte[]> mFilledBuffers =
-        new LinkedTransferQueue<byte[]>();
-
-    private Broadcaster<ComplexBuffer> mComplexBufferBroadcaster = new Broadcaster<>();
-
     private ByteSampleAdapter mSampleAdapter = new ByteSampleAdapter();
-    private BufferProcessor mBufferProcessor = new BufferProcessor();
+    private USBTransferProcessor mUSBTransferProcessor;
 
     private HackRFSampleRate mSampleRate = HackRFSampleRate.RATE2_016MHZ;
-    public int mBufferSize = 262144;
     private boolean mAmplifierEnabled = false;
 
     private Device mDevice;
@@ -167,6 +144,20 @@ public class HackRFTunerController extends TunerController
             throw new SourceException("HackRF Tuner Controller - couldn't "
                 + "claim USB interface or get endpoint or pipe", e);
         }
+
+        String name;
+
+        try
+        {
+            name = "HackRF " + getSerial().getSerialNumber();
+        }
+        catch(UsbException ue)
+        {
+            //Do nothing, we couldn't determine the serial number
+            name = "HackRF - Unidentified Serial";
+        }
+
+        mUSBTransferProcessor = new USBTransferProcessor(name, mDeviceHandle, mSampleAdapter, USB_TRANSFER_BUFFER_SIZE);
     }
 
     /**
@@ -890,21 +881,7 @@ public class HackRFTunerController extends TunerController
      */
     public void addListener(Listener<ComplexBuffer> listener)
     {
-        synchronized(mComplexBufferBroadcaster)
+        mUSBTransferProcessor.addListener(listener);
-        {
-            mComplexBufferBroadcaster.addListener(listener);
-
-            if(mBufferProcessor == null || !mBufferProcessor.isRunning())
-            {
-                mBufferProcessor = new BufferProcessor();
-
-                Thread thread = new Thread(mBufferProcessor);
-                thread.setDaemon(true);
-                thread.setName("HackRF Sample Processor");
-
-                thread.start();
-            }
-        }
     }
 
     /**
@@ -913,241 +890,7 @@ public class HackRFTunerController extends TunerController
      */
     public void removeListener(Listener<ComplexBuffer> listener)
     {
-        synchronized(mComplexBufferBroadcaster)
+        mUSBTransferProcessor.removeListener(listener);
-        {
-            mComplexBufferBroadcaster.removeListener(listener);
-
-            if(!mComplexBufferBroadcaster.hasListeners())
-            {
-                mBufferProcessor.stop();
-            }
-        }
     }
 
-    /**
-     * Buffer processing thread.  Fetches samples from the HackRF Tuner and
-     * dispatches them to all registered listeners
-     */
-    public class BufferProcessor implements Runnable, TransferCallback
-    {
-        private ScheduledFuture<?> mSampleDispatcherTask;
-        private AtomicBoolean mRunning = new AtomicBoolean();
-        private LinkedTransferQueue<Transfer> mAvailableTransfers;
-        private LinkedTransferQueue<Transfer> mTransfersInProgress = new LinkedTransferQueue<>();
-        private ByteBuffer mLibUsbHandlerStatus;
-        private boolean mCancel = false;
-
-        @Override
-        public void run()
-        {
-            if(mRunning.compareAndSet(false, true))
-            {
-                prepareTransfers();
-
-                mSampleDispatcherTask = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferDispatcher(),
-                    20, 20, TimeUnit.MILLISECONDS);
-
-                mLibUsbHandlerStatus = ByteBuffer.allocateDirect(4);
-
-                List<Transfer> transfers = new ArrayList<>();
-
-                mCancel = false;
-
-                while(mRunning.get())
-                {
-                    mAvailableTransfers.drainTo(transfers);
-
-                    for(Transfer transfer : transfers)
-                    {
-                        int result = LibUsb.submitTransfer(transfer);
-
-                        if(result == LibUsb.SUCCESS)
-                        {
-                            mTransfersInProgress.add(transfer);
-                        }
-                        else
-                        {
-                            mLog.error("error submitting transfer [" +
-                                LibUsb.errorName(result) + "]");
-                        }
-                    }
-
-                    int result = LibUsb.handleEventsTimeoutCompleted(
-                        null, USB_TIMEOUT_US, mLibUsbHandlerStatus.asIntBuffer());
-
-                    if(result != LibUsb.SUCCESS)
-                    {
-                        mLog.error("error handling events for libusb");
-                    }
-
-                    transfers.clear();
-
-                    mLibUsbHandlerStatus.rewind();
-                }
-
-                if(mCancel)
-                {
-                    for(Transfer transfer : mTransfersInProgress)
-                    {
-                        LibUsb.cancelTransfer(transfer);
-                    }
-
-                    int result = LibUsb.handleEventsTimeoutCompleted(null,
-                        USB_TIMEOUT_US, mLibUsbHandlerStatus.asIntBuffer());
-
-                    if(result != LibUsb.SUCCESS)
-                    {
-                        mLog.error("error cancelling transfers in progress");
-                    }
-
-                    mLibUsbHandlerStatus.rewind();
-                }
-            }
-        }
-
-        /**
-         * Stops the sample fetching thread
-         */
-        public void stop()
-        {
-            mCancel = true;
-
-            if(mRunning.compareAndSet(true, false))
-            {
-                if(mSampleDispatcherTask != null)
-                {
-                    mSampleDispatcherTask.cancel(true);
-                    mFilledBuffers.clear();
-                }
-            }
-        }
-
-        /**
-         * Indicates if this thread is running
-         */
-        public boolean isRunning()
-        {
-            return mRunning.get();
-        }
-
-        @Override
-        public void processTransfer(Transfer transfer)
-        {
-            mTransfersInProgress.remove(transfer);
-
-            switch(transfer.status())
-            {
-                case LibUsb.TRANSFER_COMPLETED:
-                case LibUsb.TRANSFER_STALL:
-                    if(transfer.actualLength() > 0)
-                    {
-                        ByteBuffer buffer = transfer.buffer();
-
-                        byte[] data = new byte[transfer.actualLength()];
-
-                        buffer.get(data);
-
-                        buffer.rewind();
-
-                        if(isRunning())
-                        {
-                            mFilledBuffers.add(data);
-                        }
-                    }
-                    break;
-                case LibUsb.TRANSFER_CANCELLED:
-                    break;
-                default:
-                    /* unexpected error */
-                    mLog.error("transfer error [" +
-                        getTransferStatus(transfer.status()) +
-                        "] transferred actual: " + transfer.actualLength());
-            }
-
-            mAvailableTransfers.add(transfer);
-        }
-
-        /**
-         * Prepares (allocates) a set of transfer buffers for use in
-         * transferring data from the tuner via the bulk interface
-         */
-        private void prepareTransfers() throws LibUsbException
-        {
-            if(mAvailableTransfers == null)
-            {
-                mAvailableTransfers = new LinkedTransferQueue<>();
-
-                for(int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++)
-                {
-                    Transfer transfer = LibUsb.allocTransfer();
-
-                    if(transfer == null)
-                    {
-                        throw new LibUsbException("couldn't allocate transfer",
-                            LibUsb.ERROR_NO_MEM);
-                    }
-
-                    final ByteBuffer buffer =
-                        ByteBuffer.allocateDirect(mBufferSize);
-
-                    LibUsb.fillBulkTransfer(transfer, mDeviceHandle, USB_ENDPOINT,
-                        buffer, BufferProcessor.this, "Buffer", USB_TIMEOUT_US);
-
-                    mAvailableTransfers.add(transfer);
-                }
-            }
-        }
-    }
-
-    /**
-     * Fetches byte[] chunks from the raw sample buffer.  Converts each byte
-     * array and broadcasts the array to all registered listeners
-     */
-    public class BufferDispatcher implements Runnable
-    {
-        @Override
-        public void run()
-        {
-            try
-            {
-                ArrayList<byte[]> buffers = new ArrayList<byte[]>();
-
-                mFilledBuffers.drainTo(buffers);
-
-                for(byte[] buffer : buffers)
-                {
-                    float[] samples = mSampleAdapter.convert(buffer);
-
-                    mComplexBufferBroadcaster.broadcast(new ComplexBuffer(samples));
-                }
-            }
-            catch(Exception e)
-            {
-                mLog.error("error duruing HackRF buffer dispatching", e);
-            }
-        }
-    }
-
-    public static String getTransferStatus(int status)
-    {
-        switch(status)
-        {
-            case 0:
-                return "TRANSFER COMPLETED (0)";
-            case 1:
-                return "TRANSFER ERROR (1)";
-            case 2:
-                return "TRANSFER TIMED OUT (2)";
-            case 3:
-                return "TRANSFER CANCELLED (3)";
-            case 4:
-                return "TRANSFER STALL (4)";
-            case 5:
-                return "TRANSFER NO DEVICE (5)";
-            case 6:
-                return "TRANSFER OVERFLOW (6)";
-            default:
-                return "UNKNOWN TRANSFER STATUS (" + status + ")";
-        }
-    }
 }

src/source/tuner/rtl/RTL2832TunerController.java

@@ -1,6 +1,6 @@
 /*******************************************************************************
  *     SDR Trunk 
- *     Copyright (C) 2014 Dennis Sheirer
+ *     Copyright (C) 2014-2017 Dennis Sheirer
  *
  *     Java version based on librtlsdr
  *     Copyright (C) 2012-2013 by Steve Markgraf <steve@steve-m.de>
@@ -21,7 +21,6 @@
  ******************************************************************************/
 package source.tuner.rtl;
 
-import buffer.FloatAveragingBuffer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.usb4java.Device;
@@ -29,54 +28,38 @@ import org.usb4java.DeviceDescriptor;
 import org.usb4java.DeviceHandle;
 import org.usb4java.LibUsb;
 import org.usb4java.LibUsbException;
-import org.usb4java.Transfer;
-import org.usb4java.TransferCallback;
 import sample.Broadcaster;
 import sample.Listener;
 import sample.adapter.ByteSampleAdapter;
+import sample.adapter.ISampleAdapter;
 import sample.complex.ComplexBuffer;
 import source.SourceException;
 import source.tuner.TunerController;
+import source.tuner.TunerManager;
 import source.tuner.TunerType;
-import util.ThreadPool;
+import source.tuner.usb.USBTransferProcessor;
 
 import javax.usb.UsbDisconnectedException;
 import javax.usb.UsbException;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.charset.Charset;
-import java.text.DecimalFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.LinkedTransferQueue;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 
 public abstract class RTL2832TunerController extends TunerController
 {
     private final static Logger mLog = LoggerFactory.getLogger(RTL2832TunerController.class);
 
-    public final static int INT_NULL_VALUE = -1;
-    public final static long LONG_NULL_VALUE = -1l;
-    public final static double DOUBLE_NULL_VALUE = -1.0D;
     public final static int TWO_TO_22_POWER = 4194304;
 
+    public final static int USB_TRANSFER_BUFFER_SIZE = 131072;
     public final static byte USB_INTERFACE = (byte) 0x0;
-    public final static byte USB_BULK_ENDPOINT = (byte) 0x81;
+    public final static byte CONTROL_ENDPOINT_IN = (byte) (LibUsb.ENDPOINT_IN | LibUsb.REQUEST_TYPE_VENDOR);
-    public final static byte CONTROL_ENDPOINT_IN =
+    public final static byte CONTROL_ENDPOINT_OUT = (byte) (LibUsb.ENDPOINT_OUT | LibUsb.REQUEST_TYPE_VENDOR);
-        (byte) (LibUsb.ENDPOINT_IN | LibUsb.REQUEST_TYPE_VENDOR);
-    public final static byte CONTROL_ENDPOINT_OUT =
-        (byte) (LibUsb.ENDPOINT_OUT | LibUsb.REQUEST_TYPE_VENDOR);
 
     public final static long TIMEOUT_US = 1000000l; //uSeconds
-
     public final static byte REQUEST_ZERO = (byte) 0;
-
-    public final static int TRANSFER_BUFFER_POOL_SIZE = 16;
-
     public final static byte EEPROM_ADDRESS = (byte) 0xA0;
 
     public final static byte[] sFIR_COEFFICIENTS =
@@ -87,8 +70,7 @@ public abstract class RTL2832TunerController extends TunerController
             (byte) 0x71, (byte) 0x74, (byte) 0x19, (byte) 0x41, (byte) 0xA5
         };
 
-    public static final SampleRate DEFAULT_SAMPLE_RATE =
+    public static final SampleRate DEFAULT_SAMPLE_RATE = SampleRate.RATE_2_400MHZ;
-        SampleRate.RATE_0_960MHZ;
 
     protected Device mDevice;
     protected DeviceDescriptor mDeviceDescriptor;
@@ -96,24 +78,13 @@ public abstract class RTL2832TunerController extends TunerController
 
     private SampleRate mSampleRate = DEFAULT_SAMPLE_RATE;
 
-    private BufferProcessor mBufferProcessor;
-
     private ByteSampleAdapter mSampleAdapter = new ByteSampleAdapter();
 
     private Broadcaster<ComplexBuffer> mComplexBufferBroadcaster = new Broadcaster<>();
 
-    private LinkedTransferQueue<byte[]> mFilledBuffers =
-        new LinkedTransferQueue<byte[]>();
-
-    private SampleRateMonitor mSampleRateMonitor;
-    private AtomicInteger mSampleCounter = new AtomicInteger();
-    private static final DecimalFormat mDecimalFormatter =
-        new DecimalFormat("###,###,###.0");
-    private static final DecimalFormat mPercentFormatter =
-        new DecimalFormat("###.00");
-
     protected int mOscillatorFrequency = 28800000; //28.8 MHz
-    public int mBufferSize = 131072;
+
+    private USBTransferProcessor mUSBTransferProcessor;
 
     protected Descriptor mDescriptor;
 
@@ -190,6 +161,11 @@ public abstract class RTL2832TunerController extends TunerController
                 + "descriptor byte array " +
                 (eeprom == null ? "[null]" : Arrays.toString(eeprom)), e);
         }
+
+        String deviceName = getTunerType().getLabel() + " " + getUniqueID();
+
+        mUSBTransferProcessor = new RTL2832USBTransferProcessor(deviceName, mDeviceHandle, mSampleAdapter,
+            USB_TRANSFER_BUFFER_SIZE);
     }
 
     /**
@@ -476,14 +452,13 @@ public abstract class RTL2832TunerController extends TunerController
     {
         try
         {
-            if(mBufferProcessor.isRunning())
+            if(mUSBTransferProcessor != null)
             {
-                mBufferProcessor.stop();
+                mUSBTransferProcessor.removeAllListeners();
+                TunerManager.LIBUSB_TRANSFER_PROCESSOR.unregisterTransferProcessor(mUSBTransferProcessor);
             }
 
             LibUsb.releaseInterface(mDeviceHandle, USB_INTERFACE);
-
-            LibUsb.exit(null);
         }
         catch(Exception e)
         {
@@ -1014,11 +989,6 @@ public abstract class RTL2832TunerController extends TunerController
         mSampleRate = sampleRate;
 
         mFrequencyController.setSampleRate(sampleRate.getRate());
-
-        if(mSampleRateMonitor != null)
-        {
-            mSampleRateMonitor.setSampleRate(mSampleRate.getRate());
-        }
     }
 
     public void setSampleRateFrequencyCorrection(int ppm) throws SourceException
@@ -1318,331 +1288,6 @@ public abstract class RTL2832TunerController extends TunerController
         }
     }
 
-    /**
-     * Adds a sample listener.  If the buffer processing thread is
-     * not currently running, starts it running in a new thread.
-     */
-    public void addListener(Listener<ComplexBuffer> listener)
-    {
-        synchronized(mComplexBufferBroadcaster)
-        {
-            mComplexBufferBroadcaster.addListener(listener);
-
-            if(mBufferProcessor == null)
-            {
-                mBufferProcessor = new BufferProcessor();
-            }
-
-            if(!mBufferProcessor.isRunning())
-            {
-                Thread thread = new Thread(mBufferProcessor);
-                thread.setDaemon(true);
-                thread.setName("RTL2832 Sample Processor");
-
-                thread.start();
-            }
-        }
-    }
-
-    /**
-     * Removes the sample listener.  If this is the last registered listener,
-     * shuts down the buffer processing thread.
-     */
-    public void removeListener(Listener<ComplexBuffer> listener)
-    {
-        synchronized(mComplexBufferBroadcaster)
-        {
-            mComplexBufferBroadcaster.removeListener(listener);
-
-            if(!mComplexBufferBroadcaster.hasListeners())
-            {
-                mBufferProcessor.stop();
-            }
-        }
-    }
-
-    /**
-     * Buffer processing thread.  Fetches samples from the RTL2832 Tuner and
-     * dispatches them to all registered listeners
-     */
-    public class BufferProcessor implements Runnable, TransferCallback
-    {
-        private ScheduledFuture<?> mSampleDispatcherTask;
-        private LinkedTransferQueue<Transfer> mAvailableTransfers;
-        private LinkedTransferQueue<Transfer> mTransfersInProgress = new LinkedTransferQueue<>();
-        private AtomicBoolean mRunning = new AtomicBoolean();
-        private ByteBuffer mLibUsbHandlerStatus;
-        private boolean mCancel = false;
-
-        public BufferProcessor()
-        {
-        }
-
-        @Override
-        public void run()
-        {
-            if(mRunning.compareAndSet(false, true))
-            {
-                try
-                {
-                    setSampleRate(mSampleRate);
-
-                    resetUSBBuffer();
-                }
-                catch(SourceException e)
-                {
-                    mLog.error("couldn't start buffer processor", e);
-
-                    mRunning.set(false);
-                }
-
-                prepareTransfers();
-
-                mSampleDispatcherTask = ThreadPool.SCHEDULED.scheduleAtFixedRate(new BufferDispatcher(),
-                    20, 20, TimeUnit.MILLISECONDS);
-
-                mLibUsbHandlerStatus = ByteBuffer.allocateDirect(4);
-
-                List<Transfer> transfers = new ArrayList<>();
-
-                mCancel = false;
-
-                while(mRunning.get())
-                {
-                    mAvailableTransfers.drainTo(transfers);
-
-                    for(Transfer transfer : transfers)
-                    {
-                        int result = LibUsb.submitTransfer(transfer);
-
-                        if(result == LibUsb.SUCCESS)
-                        {
-                            mTransfersInProgress.add(transfer);
-                        }
-                        else
-                        {
-                            mLog.error("Error submitting transfer [" +
-                                LibUsb.errorName(result) + "]");
-                        }
-                    }
-
-                    int result = LibUsb.handleEventsTimeoutCompleted(
-                        null, TIMEOUT_US, mLibUsbHandlerStatus.asIntBuffer());
-
-                    if(result != LibUsb.SUCCESS)
-                    {
-                        mLog.error("error handling events for libusb");
-                    }
-
-                    transfers.clear();
-
-                    mLibUsbHandlerStatus.rewind();
-                }
-
-                if(mCancel)
-                {
-                    for(Transfer transfer : mTransfersInProgress)
-                    {
-                        LibUsb.cancelTransfer(transfer);
-                    }
-
-                    int result = LibUsb.handleEventsTimeoutCompleted(null,
-                        TIMEOUT_US, mLibUsbHandlerStatus.asIntBuffer());
-
-                    if(result != LibUsb.SUCCESS)
-                    {
-                        mLog.error("error handling events for libusb during cancel");
-                    }
-
-                    mLibUsbHandlerStatus.rewind();
-                }
-            }
-        }
-
-        /**
-         * Stops the sample fetching thread
-         */
-        public void stop()
-        {
-            mCancel = true;
-
-            if(mRunning.compareAndSet(true, false))
-            {
-                if(mSampleDispatcherTask != null)
-                {
-                    mSampleDispatcherTask.cancel(true);
-                    mSampleDispatcherTask = null;
-                    mFilledBuffers.clear();
-                }
-            }
-        }
-
-        /**
-         * Indicates if this thread is running
-         */
-        public boolean isRunning()
-        {
-            return mRunning.get();
-        }
-
-        @Override
-        public void processTransfer(Transfer transfer)
-        {
-            mTransfersInProgress.remove(transfer);
-
-            switch(transfer.status())
-            {
-                case LibUsb.TRANSFER_COMPLETED:
-                case LibUsb.TRANSFER_STALL:
-                    if(transfer.actualLength() > 0)
-                    {
-                        ByteBuffer buffer = transfer.buffer();
-
-                        byte[] data = new byte[transfer.actualLength()];
-
-                        buffer.get(data);
-
-                        buffer.rewind();
-
-                        if(isRunning())
-                        {
-                            mFilledBuffers.add(data);
-                        }
-                    }
-                    break;
-                case LibUsb.TRANSFER_CANCELLED:
-                    break;
-                default:
-					/* unexpected error */
-                    mLog.error("transfer error [" +
-                        getTransferStatus(transfer.status()) +
-                        "] transferred actual: " + transfer.actualLength());
-            }
-
-            mAvailableTransfers.add(transfer);
-        }
-
-        /**
-         * Prepares (allocates) a set of transfer buffers for use in
-         * transferring data from the tuner via the bulk interface
-         */
-        private void prepareTransfers() throws LibUsbException
-        {
-            if(mAvailableTransfers == null)
-            {
-                mAvailableTransfers = new LinkedTransferQueue<>();
-
-                for(int x = 0; x < TRANSFER_BUFFER_POOL_SIZE; x++)
-                {
-                    Transfer transfer = LibUsb.allocTransfer();
-
-                    if(transfer == null)
-                    {
-                        throw new LibUsbException("couldn't allocate transfer",
-                            LibUsb.ERROR_NO_MEM);
-                    }
-
-                    final ByteBuffer buffer =
-                        ByteBuffer.allocateDirect(mBufferSize);
-
-                    LibUsb.fillBulkTransfer(transfer, mDeviceHandle, USB_BULK_ENDPOINT,
-                        buffer, BufferProcessor.this, "Buffer", TIMEOUT_US);
-
-                    mAvailableTransfers.add(transfer);
-                }
-            }
-        }
-    }
-
-    /**
-     * Fetches byte[] chunks from the raw sample buffer.  Converts each byte
-     * array and broadcasts the array to all registered listeners
-     */
-    public class BufferDispatcher implements Runnable
-    {
-        @Override
-        public void run()
-        {
-            try
-            {
-                ArrayList<byte[]> buffers = new ArrayList<byte[]>();
-
-                mFilledBuffers.drainTo(buffers);
-
-                for(byte[] buffer : buffers)
-                {
-                    float[] samples = mSampleAdapter.convert(buffer);
-
-                    mComplexBufferBroadcaster.broadcast(new ComplexBuffer(samples));
-                }
-            }
-            catch(Exception e)
-            {
-                mLog.error("error during rtl2832 buffer dispatcher run", e);
-            }
-        }
-    }
-
-    public static String getTransferStatus(int status)
-    {
-        switch(status)
-        {
-            case 0:
-                return "TRANSFER COMPLETED (0)";
-            case 1:
-                return "TRANSFER ERROR (1)";
-            case 2:
-                return "TRANSFER TIMED OUT (2)";
-            case 3:
-                return "TRANSFER CANCELLED (3)";
-            case 4:
-                return "TRANSFER STALL (4)";
-            case 5:
-                return "TRANSFER NO DEVICE (5)";
-            case 6:
-                return "TRANSFER OVERFLOW (6)";
-            default:
-                return "UNKNOWN TRANSFER STATUS (" + status + ")";
-        }
-    }
-
-
-    public class USBEventHandlingThread implements Runnable
-    {
-        /**
-         * If thread should abort.
-         */
-        private volatile boolean mAbort;
-
-        /**
-         * Aborts the event handling thread.
-         */
-        public void abort()
-        {
-            mAbort = true;
-        }
-
-        @Override
-        public void run()
-        {
-            while(!mAbort)
-            {
-                int result = LibUsb.handleEventsTimeout(null, 1000);
-
-                if(result != LibUsb.SUCCESS)
-                {
-                    mAbort = true;
-
-                    mLog.error("error handling usb events [" +
-                        LibUsb.errorName(result) + "]");
-
-                    throw new LibUsbException("Unable to handle USB "
-                        + "events", result);
-                }
-            }
-        }
-    }
-
     public enum TunerTypeCheck
     {
         E4K(0xC8, 0x02, 0x40),
@@ -1679,102 +1324,6 @@ public abstract class RTL2832TunerController extends TunerController
         }
     }
 
-
-    /**
-     * Averages the sample rate over a 10-second period.  The count is for the
-     * number of bytes received from the tuner.  There are two bytes for each
-     * sample.  So, we divide by 20 to get the average sample rate.
-     */
-    public class SampleRateMonitor implements Runnable
-    {
-        private static final int BUFFER_SIZE = 5;
-        private int mTargetSampleRate;
-        private int mSampleRateMinimum;
-        private int mSampleRateMaximum;
-        private int mNewTargetSampleRate;
-        private FloatAveragingBuffer mRateErrorBuffer =
-            new FloatAveragingBuffer(BUFFER_SIZE);
-        private AtomicBoolean mRateChanged = new AtomicBoolean();
-
-        public SampleRateMonitor(int sampleRate)
-        {
-            setTargetRate(sampleRate);
-        }
-
-        public void setSampleRate(int sampleRate)
-        {
-            mNewTargetSampleRate = sampleRate;
-            mRateChanged.set(true);
-        }
-
-        private void setTargetRate(int rate)
-        {
-            mTargetSampleRate = rate;
-            mSampleRateMinimum = (int) ((float) mTargetSampleRate * 0.95f);
-            mSampleRateMaximum = (int) ((float) mTargetSampleRate * 1.05f);
-
-            for(int x = 0; x < BUFFER_SIZE; x++)
-            {
-                mRateErrorBuffer.get(0);
-            }
-        }
-
-        @Override
-        public void run()
-        {
-            if(mRateChanged.compareAndSet(true, false))
-            {
-                setTargetRate(mNewTargetSampleRate);
-
-				/* Reset the sample counter */
-                mSampleCounter.set(0);
-
-                mLog.info("monitor reset for new sample rate [" + mTargetSampleRate + "]");
-            }
-            else
-            {
-                int count = mSampleCounter.getAndSet(0);
-
-                float current = (float) count / 20.0f;
-
-                /**
-                 * Only accept values +/- 5% of target rate
-                 */
-                float average;
-
-                if(mSampleRateMinimum < current && current < mSampleRateMaximum)
-                {
-                    average = mRateErrorBuffer.get((float) mTargetSampleRate - current);
-                }
-                else
-                {
-                    average = mTargetSampleRate;
-                }
-
-                StringBuilder sb = new StringBuilder();
-                sb.append("[");
-                if(mDescriptor != null)
-                {
-                    sb.append(mDescriptor.getSerial());
-                }
-                else
-                {
-                    sb.append("DESCRIPTOR IS NULL");
-                }
-                sb.append("] sample rate current [");
-                sb.append(mDecimalFormatter.format(current));
-                sb.append(" Hz ");
-                sb.append(mPercentFormatter.format(100.0f * (current / (float) mTargetSampleRate)));
-                sb.append("% ] error [");
-                sb.append(average);
-                sb.append(" Hz ] target ");
-                sb.append(mDecimalFormatter.format(mTargetSampleRate));
-
-                mLog.info(sb.toString());
-            }
-        }
-    }
-
     /**
      * RTL2832 EEPROM byte array descriptor parsing class
      */
@@ -1945,4 +1494,47 @@ public abstract class RTL2832TunerController extends TunerController
         }
     }
 
+    /**
+     * Adds a sample listener.  If the USB transfer processor is not currently running, it will auto-start
+     */
+    public void addListener(Listener<ComplexBuffer> listener)
+    {
+        mUSBTransferProcessor.addListener(listener);
+    }
+
+    /**
+     * Removes the sample listener.  If this is the last registered listener, the USB transfer processor will
+     * halt buffer processing
+     */
+    public void removeListener(Listener<ComplexBuffer> listener)
+    {
+        mUSBTransferProcessor.removeListener(listener);
+    }
+
+    /**
+     * RTL-2832 USB transfer processor.  Extends USB transfer processor and allows resetting the device USB
+     * buffer prior to starting streaming.
+     */
+    public class RTL2832USBTransferProcessor extends USBTransferProcessor
+    {
+        /**
+         * Manages stream of USB transfer buffers and converts buffers to complex buffer samples for distribution to
+         * any registered listeners.
+         *
+         * @param deviceName to use when logging information or errors
+         * @param deviceHandle to the USB bulk transfer device
+         * @param sampleAdapter specific to the tuner's byte buffer format for converting to floating point I/Q samples
+         * @param bufferSize in bytes.  Should be a multiple of two: 65536, 131072 or 262144.
+         */
+        public RTL2832USBTransferProcessor(String deviceName, DeviceHandle deviceHandle, ISampleAdapter sampleAdapter, int bufferSize)
+        {
+            super(deviceName, deviceHandle, sampleAdapter, bufferSize);
+        }
+
+        @Override
+        protected void prepareDeviceStart()
+        {
+            resetUSBBuffer();
+        }
+    }
 }

src/source/tuner/usb/USBTransferProcessor.java

@@ -0,0 +1,390 @@
+/*******************************************************************************
+ * 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.usb;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.usb4java.DeviceHandle;
+import org.usb4java.LibUsb;
+import org.usb4java.LibUsbException;
+import org.usb4java.Transfer;
+import org.usb4java.TransferCallback;
+import sample.Broadcaster;
+import sample.Listener;
+import sample.OverflowableTransferQueue;
+import sample.adapter.ISampleAdapter;
+import sample.complex.ComplexBuffer;
+import source.tuner.TunerManager;
+import util.ThreadPool;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class USBTransferProcessor implements TransferCallback
+{
+    private final static Logger mLog = LoggerFactory.getLogger(USBTransferProcessor.class);
+
+    private static final byte USB_BULK_TRANSFER_ENDPOINT = (byte) 0x81;
+    private static final long USB_TIMEOUT_MS = 2000l; //milliseconds
+
+    //Number of native byte buffers to allocate for transferring data from the USB device
+    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;
+
+    //Threshold for resetting buffer overflow condition
+    private static final int FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD = 200;
+
+    private String mDeviceName;
+
+    //Handle to the USB bulk transfer device
+    private DeviceHandle mDeviceHandle;
+
+    //Tuner format-specific byte to IQ float sample converter
+    private ISampleAdapter mSampleAdapter;
+
+    //Byte array transfer buffers size in bytes
+    private int mBufferSize;
+
+    private Broadcaster<ComplexBuffer> mComplexBufferBroadcaster = new Broadcaster<>();
+    private OverflowableTransferQueue<byte[]> mFilledBuffers;
+    private LinkedTransferQueue<Transfer> mAvailableTransfers = new LinkedTransferQueue<>();
+    private LinkedTransferQueue<Transfer> mTransfersInProgress = new LinkedTransferQueue<>();
+
+    private AtomicBoolean mRunning = new AtomicBoolean();
+
+    private ByteBuffer mLibUsbHandlerStatus = ByteBuffer.allocateDirect(4);
+
+    private BufferDispatcher mBufferDispatcher = new BufferDispatcher();
+    private ScheduledFuture mBufferDispatcherFuture;
+
+    /**
+     * Manages stream of USB transfer buffers and converts buffers to complex buffer samples for distribution to
+     * any registered listeners.
+     *
+     * @param deviceName to use when logging information or errors
+     * @param deviceHandle to the USB bulk transfer device
+     * @param sampleAdapter specific to the tuner's byte buffer format for converting to floating point I/Q samples
+     * @param bufferSize in bytes.  Should be a multiple of two: 65536, 131072 or 262144.
+     */
+    public USBTransferProcessor(String deviceName,
+                                DeviceHandle deviceHandle,
+                                ISampleAdapter sampleAdapter,
+                                int bufferSize)
+    {
+        mDeviceName = deviceName;
+        mDeviceHandle = deviceHandle;
+        mSampleAdapter = sampleAdapter;
+        mBufferSize = bufferSize;
+
+        mFilledBuffers = new OverflowableTransferQueue<>(FILLED_BUFFER_MAX_CAPACITY, FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD);
+        mFilledBuffers.setStateListener(new Listener<OverflowableTransferQueue.State>()
+        {
+            @Override
+            public void receive(OverflowableTransferQueue.State state)
+            {
+                if(state == OverflowableTransferQueue.State.NORMAL)
+                {
+                    mLog.debug(mDeviceName + " - buffer overflow - temporary pause until processing catches up");
+                }
+                else
+                {
+                    mLog.debug(mDeviceName + " - buffer overflow cleared - resuming normal processing");
+                }
+            }
+        });
+    }
+
+    /**
+     * Start USB transfer buffer processing.  Subsequent calls to this method after started will be ignored.
+     */
+    private void start()
+    {
+        if(mRunning.compareAndSet(false, true))
+        {
+            prepareDeviceStart();
+
+            prepareTransfers();
+
+            while(!mAvailableTransfers.isEmpty())
+            {
+                Transfer transfer = mAvailableTransfers.poll();
+
+                if(transfer != null)
+                {
+                    int result = LibUsb.submitTransfer(transfer);
+
+                    if(result == LibUsb.SUCCESS)
+                    {
+                        mTransfersInProgress.add(transfer);
+                    }
+                    else
+                    {
+                        //TODO: broadcast to each listener that this source has an error and is shutting down
+                        mLog.error(mDeviceName + "- error submitting transfer [" + LibUsb.errorName(result) + "]");
+                    }
+                }
+            }
+
+            //Start transferred buffer dispatcher
+            mBufferDispatcherFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(mBufferDispatcher,
+                20, 20, TimeUnit.MILLISECONDS);
+
+            //Register with LibUSB processor so that it auto-starts LibUSB processing
+            TunerManager.LIBUSB_TRANSFER_PROCESSOR.registerTransferProcessor(this);
+        }
+    }
+
+    /**
+     * Stop USB transfer buffer processing.  Subsequent calls to this method after stopped will be ignored.
+     */
+    private void stop()
+    {
+        if(mRunning.compareAndSet(true, false))
+        {
+            mBufferDispatcherFuture.cancel(true);
+
+            mFilledBuffers.clear();
+
+            //Cancel the lib usb process timer
+            for(Transfer transfer : mTransfersInProgress)
+            {
+                LibUsb.cancelTransfer(transfer);
+            }
+
+
+            //Unregister from LibUSB processor so that it auto-stops LibUSB processing
+            TunerManager.LIBUSB_TRANSFER_PROCESSOR.registerTransferProcessor(this);
+
+            //Directly invoke the timeout handler to ensure that our cancelled transfer buffers are flushed.
+            int result = LibUsb.handleEventsTimeoutCompleted(null, USB_TIMEOUT_MS,
+                mLibUsbHandlerStatus.asIntBuffer());
+
+            if(result != LibUsb.SUCCESS)
+            {
+                mLog.error(mDeviceName + " - error while cancelling transfer buffers during shutdown/pause - error code:" + result);
+            }
+
+            mLibUsbHandlerStatus.rewind();
+
+            executeDeviceStop();
+        }
+    }
+
+    /**
+     * Allows sub-class implementations to execute any device-specific operations to prepare for starting USB transfers
+     */
+    protected void prepareDeviceStart()
+    {
+
+    }
+
+    /**
+     * Allows sub-class implementations to execute any device-specific operations after stopping USB transfers
+     */
+    protected void executeDeviceStop()
+    {
+
+    }
+
+    /**
+     * Indicates if there are complex buffer listeners registered with this processor
+     */
+    public boolean hasListeners()
+    {
+        return mComplexBufferBroadcaster.hasListeners();
+    }
+
+    public void addListener(Listener<ComplexBuffer> listener)
+    {
+        mComplexBufferBroadcaster.addListener(listener);
+
+        start();
+    }
+
+    public void removeListener(Listener<ComplexBuffer> listener)
+    {
+        mComplexBufferBroadcaster.removeListener(listener);
+
+        if(!hasListeners())
+        {
+            stop();
+        }
+    }
+
+    public void removeAllListeners()
+    {
+        mComplexBufferBroadcaster.clear();
+        stop();
+    }
+
+    /**
+     * Prepares (allocates) a set of transfer buffers for use in transferring data from the USB device via the bulk
+     * interface.  Since we're using direct allocation (native), buffers are retained and reused across multiple
+     * start/stop cycles.
+     */
+    private void prepareTransfers() throws LibUsbException
+    {
+        while(mAvailableTransfers.size() < TRANSFER_BUFFER_POOL_SIZE)
+        {
+            Transfer transfer = LibUsb.allocTransfer();
+
+            if(transfer == null)
+            {
+                throw new LibUsbException("Couldn't allocate USB transfer buffer", LibUsb.ERROR_NO_MEM);
+            }
+
+            final ByteBuffer buffer = ByteBuffer.allocateDirect(mBufferSize);
+
+            LibUsb.fillBulkTransfer(transfer, mDeviceHandle, USB_BULK_TRANSFER_ENDPOINT, buffer, this,
+                "Buffer", USB_TIMEOUT_MS);
+
+            mAvailableTransfers.add(transfer);
+        }
+    }
+
+    /**
+     * Process a filled transfer buffer received back from the USB device
+     */
+    @Override
+    public void processTransfer(Transfer transfer)
+    {
+        mTransfersInProgress.remove(transfer);
+
+        switch(transfer.status())
+        {
+            case LibUsb.TRANSFER_COMPLETED:
+            case LibUsb.TRANSFER_STALL:
+                if(transfer.actualLength() > 0)
+                {
+                    ByteBuffer buffer = transfer.buffer();
+
+                    byte[] data = new byte[transfer.actualLength()];
+                    buffer.get(data);
+                    buffer.rewind();
+
+                    if(mRunning.get())
+                    {
+                        mFilledBuffers.offer(data);
+                    }
+                }
+                break;
+            case LibUsb.TRANSFER_CANCELLED:
+                break;
+            default:
+                //Unexpected transfer error
+                mLog.error(mDeviceName + " - transfer error [" + getTransferStatus(transfer.status()) +
+                    "] transferred actual: " + transfer.actualLength());
+        }
+
+
+        if(mRunning.get())
+        {
+            int result = LibUsb.submitTransfer(transfer);
+
+            if(result == LibUsb.SUCCESS)
+            {
+                mTransfersInProgress.add(transfer);
+            }
+            else
+            {
+                mAvailableTransfers.add(transfer);
+                mLog.error(mDeviceName + " - error submitting transfer [" + LibUsb.errorName(result) + "]");
+
+                if(mTransfersInProgress.isEmpty())
+                {
+                    mLog.warn(mDeviceName + " - all transfer buffer processing is stopped");
+                    //TODO: no transfers are in progress ... need to alert the user and the registered complex
+                    //TODO: buffer listeners so that they can respond accordingly
+                }
+            }
+        }
+        else
+        {
+            //We're stopping - park the transfer buffers
+            if(!mAvailableTransfers.contains(transfer))
+            {
+                mAvailableTransfers.add(transfer);
+            }
+        }
+    }
+
+    /**
+     * Converts the USB transfer status number into a descriptive label
+     */
+    public static String getTransferStatus(int status)
+    {
+        switch(status)
+        {
+            case 0:
+                return "TRANSFER COMPLETED (0)";
+            case 1:
+                return "TRANSFER ERROR (1)";
+            case 2:
+                return "TRANSFER TIMED OUT (2)";
+            case 3:
+                return "TRANSFER CANCELLED (3)";
+            case 4:
+                return "TRANSFER STALL (4)";
+            case 5:
+                return "TRANSFER NO DEVICE (5)";
+            case 6:
+                return "TRANSFER OVERFLOW (6)";
+            default:
+                return "UNKNOWN TRANSFER STATUS (" + status + ")";
+        }
+    }
+
+    /**
+     * Fetches byte[] chunks from the raw sample buffer.  Converts each byte
+     * array and broadcasts the array to all registered listeners
+     */
+    public class BufferDispatcher implements Runnable
+    {
+        private List<byte[]> mBuffersToDispatch = new ArrayList<>();
+
+        @Override
+        public void run()
+        {
+            try
+            {
+                mFilledBuffers.drainTo(mBuffersToDispatch, 15);
+
+                for(byte[] buffer : mBuffersToDispatch)
+                {
+                    float[] complexSamples = mSampleAdapter.convert(buffer);
+
+                    mComplexBufferBroadcaster.broadcast(new ComplexBuffer(complexSamples));
+                }
+            }
+            catch(Exception e)
+            {
+                mLog.error(mDeviceName + " - error while dispatching complex IQ buffer samples", e);
+            }
+
+            mBuffersToDispatch.clear();
+        }
+    }
+}