[WIP]-39 updates to resolve issue with stopping channel streams.

src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java

@@ -15,10 +15,15 @@
  ******************************************************************************/
 package io.github.dsheirer.sample.buffer;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.concurrent.atomic.AtomicInteger;
 
 public abstract class AbstractReusableBuffer
 {
+    private final static Logger mLog = LoggerFactory.getLogger(AbstractReusableBuffer.class);
+
     protected IReusableBufferDisposedListener mBufferDisposedListener;
     protected long mTimestamp;
     protected AtomicInteger mUserCount = new AtomicInteger();
@@ -35,6 +40,11 @@ public abstract class AbstractReusableBuffer
         mBufferDisposedListener = bufferDisposedListener;
     }
 
+    public void dispose()
+    {
+        mBufferDisposedListener = null;
+    }
+
     public void setDebugName(String debugName)
     {
         mDebugName = debugName;
@@ -65,15 +75,12 @@ public abstract class AbstractReusableBuffer
     }
 
     /**
-     * Sets the number of users that will receive this buffer.  Each user is expected to invoke the
-     * decrementUserCount() method to indicate that they are finished using this buffer so that this buffer can be
-     * reused once the user count is zero.
-     *
-     * @param userCount that will receive this reusable buffer
+     * Sets the user count to 0 and recycles this buffer
      */
-    public void setUserCount(int userCount)
+    public void clearUserCount()
     {
-        mUserCount.set(userCount);
+        mUserCount.set(0);
+        recycle();
     }
 
     /**
@@ -88,14 +95,25 @@ public abstract class AbstractReusableBuffer
      */
     public void decrementUserCount()
     {
-        int currentCount = mUserCount.decrementAndGet();
-
-        if(currentCount == 0)
+        if(mUserCount.decrementAndGet() <= 0)
         {
             recycle();
+        }
+    }
+
+    /**
+     * Sends this buffer back to the owning buffer queue for reuse
+     */
+    private void recycle()
+    {
+        prepareForRecycle();
+
+        if(mBufferDisposedListener != null)
+        {
             mBufferDisposedListener.disposed(this);
         }
-        else if(currentCount < 0)
+
+        if(mUserCount.get() < 0)
         {
             throw new IllegalStateException("User count is below zero.  This indicates that this buffer's decrement" +
                 " user count was invoked by more than the expected user count");
@@ -103,10 +121,10 @@ public abstract class AbstractReusableBuffer
     }
 
     /**
-     * Invoked just prior to notifying the owner that this buffer is ready for recycle.  This method
-     * is intended for sub-class implementations to perform any recycle cleanup actions.
+     * Invoked just prior to notifying the owner that this buffer is ready for prepareForRecycle.  This method
+     * is intended for sub-class implementations to perform any prepareForRecycle cleanup actions.
      */
-    protected void recycle()
+    protected void prepareForRecycle()
     {
         //No-op in this abstract class
     }
@@ -138,7 +156,14 @@ public abstract class AbstractReusableBuffer
      */
     public void incrementUserCount(int additionalUserCount)
     {
-        mUserCount.addAndGet(additionalUserCount);
+        if(additionalUserCount >= 0)
+        {
+            mUserCount.addAndGet(additionalUserCount);
+        }
+        else
+        {
+            mLog.debug("Request to add negative user count [" + additionalUserCount + "] ignored");
+        }
     }
 
     /**

src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java

@@ -30,18 +30,37 @@ public abstract class AbstractReusableBufferQueue<T extends AbstractReusableBuff
     private int mBufferCount = 0;
     private String mDebugName;
 
+    /**
+     * Base queue for managing reusable buffers.
+     *
+     * @param debugName to associate a debug string with this buffer instance
+     */
     public AbstractReusableBufferQueue(String debugName)
     {
         mDebugName = debugName;
     }
 
+    /**
+     * Base queue for managing reusable buffers.
+     */
     public AbstractReusableBufferQueue()
     {
     }
 
+    /**
+     * Disposes of any reclaimed buffers to prepare this queue for disposal.
+     */
     public void dispose()
     {
-        mReusableBufferQueue.clear();
+        T buffer = mReusableBufferQueue.poll();
+
+        while(buffer != null)
+        {
+            buffer.dispose();
+            buffer = mReusableBufferQueue.poll();
+        }
+
+        mBufferCount = 0;
     }
 
     /**
@@ -72,11 +91,17 @@ public abstract class AbstractReusableBufferQueue<T extends AbstractReusableBuff
         mBufferCount++;
     }
 
+    /**
+     * Current count of buffers created by this queue
+     */
     protected int getBufferCount()
     {
         return mBufferCount;
     }
 
+    /**
+     * Debug name for this queue instance
+     */
     protected String getDebugName()
     {
         return mDebugName;

src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferAssembler.java

@@ -59,8 +59,8 @@ public class ReusableBufferAssembler
 
     public void dispose()
     {
+        mReusableComplexBufferQueue.dispose();
         mListener = null;
-        mBuffer = null;
     }
 
     /**
@@ -117,7 +117,7 @@ public class ReusableBufferAssembler
             mBuffer.rewind();
             ReusableComplexBuffer reusableComplexBuffer = mReusableComplexBufferQueue.getBuffer(mBufferSize);
             reusableComplexBuffer.reloadFrom(mBuffer, mCurrentBufferTimestamp);
-            reusableComplexBuffer.setUserCount(1);
+            reusableComplexBuffer.incrementUserCount();
             mListener.receive(reusableComplexBuffer);
         }
 

src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBuffer.java

@@ -26,10 +26,10 @@ public class ReusableChannelResultsBuffer extends AbstractReusableBuffer
 
     /**
      * Reusable buffer for storing polyphase channelizer results arrays.  This buffer is NOT thread safe and is
-     * designed to be used by only one consumer.  The consumer must invoke the recycle() method to indicate that this
+     * designed to be used by only one consumer.  The consumer must invoke the prepareForRecycle() method to indicate that this
      * buffer can be reused.
      *
-     * @param disposedListener to receive notification when this buffer is ready for recycle
+     * @param disposedListener to receive notification when this buffer is ready for prepareForRecycle
      */
     ReusableChannelResultsBuffer(IReusableBufferDisposedListener disposedListener)
     {
@@ -39,7 +39,7 @@ public class ReusableChannelResultsBuffer extends AbstractReusableBuffer
     /**
      * Recycles this buffer for reuse.
      */
-    protected void recycle()
+    protected void prepareForRecycle()
     {
         mEmptyBuffers.addAll(mFilledBuffers);
         mFilledBuffers.clear();

src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java

@@ -22,11 +22,18 @@ public class ReusableComplexBufferQueue extends AbstractReusableBufferQueue<Reus
 {
     private final static Logger mLog = LoggerFactory.getLogger(ReusableComplexBufferQueue.class);
 
+    /**
+     * Queue for managing the creation and reuse of reusable complex sample buffers.
+     * @param debugName to associate with this queue instance
+     */
     public ReusableComplexBufferQueue(String debugName)
     {
         super(debugName);
     }
 
+    /**
+     * Queue for managing the creation and reuse of reusable complex sample buffers.
+     */
     public ReusableComplexBufferQueue()
     {
     }