37607: Enhance BlockingBuffer to allow for a timeout value

git-svn-id: https://svn.apache.org/repos/asf/jakarta/commons/proper/collections/trunk@348428 13f79535-47bb-0310-9956-ffa450edef68

RELEASE-NOTES.html

@@ -72,6 +72,7 @@ If this causes major headaches to anyone please contact commons-dev at jakarta.a
 <li>ExtendedProperties - No longer uses an exception in normal processing [30497]</li>
 <li>BlockingBuffer - now includes stack trace if InterupttedException occurs [33700]</li>
 <li>BlockingBuffer - new methods that allow get and remove with a timeout [27691]</li>
+<li>BlockingBuffer - now allows you to specify a default timeout value for get/remove operations [37607]</li>
 <li>Transformed*Map - new factory decorateTransform() that transforms any existing entries in the map [30959]</li>
 <li>ListOrderedMap - values can now be accessed as a List using valueList() [37015]</li>
 <li>ListOrderedMap - additional list-like method, setValue(int,Object)</li>

STATUS.html

@@ -83,6 +83,7 @@ component to ensure that it continues to meet a variety of needs.</p>
    <li><a href="mailto:rdonkin@apache.org">Robert Burrell Donkin</a></li>
    <li><a href="mailto:psteitz@apache.org">Phil Steitz</a></li>
    <li><a href="mailto:matth@apache.org">Matthew Hawthorne</a></li>
+   <li><a href="mailto:jcarman@apache.org">James Carman</a></li>
    <li>Arun M. Thomas</li>
 </ul>
 

src/java/org/apache/commons/collections/BufferUtils.java

@@ -21,7 +21,6 @@ import org.apache.commons.collections.buffer.SynchronizedBuffer;
 import org.apache.commons.collections.buffer.TransformedBuffer;
 import org.apache.commons.collections.buffer.TypedBuffer;
 import org.apache.commons.collections.buffer.UnmodifiableBuffer;
-import org.apache.commons.collections.buffer.TimeoutBuffer;
 
 /**
  * Provides utility methods and decorators for {@link Buffer} instances.
@@ -99,8 +98,8 @@ public class BufferUtils {
      * @throws IllegalArgumentException  if the Buffer is null
      * @since Commons Collections 3.2
      */
-    public static Buffer timeoutBuffer(Buffer buffer, long timeout) {
+    public static Buffer blockingBuffer(Buffer buffer, long timeout) {
-        return TimeoutBuffer.decorate(buffer, timeout);
+        return BlockingBuffer.decorate(buffer, timeout);
     }
 
     /**

src/java/org/apache/commons/collections/buffer/BlockingBuffer.java

@@ -15,41 +15,42 @@
  */
 package org.apache.commons.collections.buffer;
 
+import org.apache.commons.collections.Buffer;
+import org.apache.commons.collections.BufferUnderflowException;
+
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.Collection;
 
-import org.apache.commons.collections.Buffer;
-import org.apache.commons.collections.BufferUnderflowException;
-
 /**
- * Decorates another <code>Buffer</code> to make {@link #get()} and
+ * Decorates another <code>Buffer</code> to make {@link #get()} and {@link #remove()} block when the <code>Buffer</code>
- * {@link #remove()} block when the <code>Buffer</code> is empty.
+ * is empty.
- * <p>
+ * <p/>
- * If either <code>get</code> or <code>remove</code> is called on an empty
+ * If either <code>get</code> or <code>remove</code> is called on an empty <code>Buffer</code>, the calling thread waits
- * <code>Buffer</code>, the calling thread waits for notification that
+ * for notification that an <code>add</code> or <code>addAll</code> operation has completed.
- * an <code>add</code> or <code>addAll</code> operation has completed.
+ * <p/>
- * <p>
+ * When one or more entries are added to an empty <code>Buffer</code>, all threads blocked in <code>get</code> or
- * When one or more entries are added to an empty <code>Buffer</code>,
+ * <code>remove</code> are notified. There is no guarantee that concurrent blocked <code>get</code> or
- * all threads blocked in <code>get</code> or <code>remove</code> are notified.
+ * <code>remove</code> requests will be "unblocked" and receive data in the order that they arrive.
- * There is no guarantee that concurrent blocked <code>get</code> or
+ * <p/>
- * <code>remove</code> requests will be "unblocked" and receive data in the
- * order that they arrive.
- * <p>
  * This class is Serializable from Commons Collections 3.1.
  *
  * @author Stephen Colebourne
  * @author Janek Bogucki
  * @author Phil Steitz
+ * @author James Carman
  * @version $Revision$ $Date$
  * @since Commons Collections 3.0
  */
 public class BlockingBuffer extends SynchronizedBuffer {
+
     /**
      * Serialization version
      */
     private static final long serialVersionUID = 1719328905017860541L;
 
+    private long timeout;
+
     /**
      * Factory method to create a blocking buffer.
      *
@@ -57,53 +58,83 @@ public class BlockingBuffer extends SynchronizedBuffer {
      * @return a new blocking Buffer
      * @throws IllegalArgumentException if buffer is null
      */
-    public static Buffer decorate(Buffer buffer) {
+    public static Buffer decorate( Buffer buffer ) {
-        return new BlockingBuffer(buffer);
+        return new BlockingBuffer( buffer );
+    }
+
+    /**
+     * Factory method to create a blocking buffer with a timeout value.
+     *
+     * @param buffer  the buffer to decorate, must not be null
+     * @param timeout the maximum amount of time to block
+     * @return a new blocking buffer
+     * @throws IllegalArgumentException if the buffer is null
+     */
+    public static Buffer decorate( Buffer buffer, long timeout ) {
+        return new BlockingBuffer( buffer, timeout );
     }
 
     //-----------------------------------------------------------------------
+
     /**
      * Constructor that wraps (not copies).
      *
      * @param buffer the buffer to decorate, must not be null
      * @throws IllegalArgumentException if the buffer is null
      */
-    protected BlockingBuffer(Buffer buffer) {
+    protected BlockingBuffer( Buffer buffer ) {
-        super(buffer);
+        super( buffer );
+    }
+
+    /**
+     * Constructor that wraps (not copies).
+     *
+     * @param buffer  the buffer to decorate, must not be null
+     * @param timeout the maximum amount of time to block
+     * @throws IllegalArgumentException if the buffer is null
+     */
+    protected BlockingBuffer( Buffer buffer, long timeout ) {
+        super( buffer );
+        this.timeout = timeout < 0 ? 0 : timeout;
     }
 
     //-----------------------------------------------------------------------
-    public boolean add(Object o) {
+    public boolean add( Object o ) {
-        synchronized (lock) {
+        synchronized( lock ) {
-            boolean result = collection.add(o);
+            boolean result = collection.add( o );
             lock.notifyAll();
             return result;
         }
     }
 
-    public boolean addAll(Collection c) {
+    public boolean addAll( Collection c ) {
-        synchronized (lock) {
+        synchronized( lock ) {
-            boolean result = collection.addAll(c);
+            boolean result = collection.addAll( c );
             lock.notifyAll();
             return result;
         }
     }
 
     /**
-     * Gets the next value from the buffer, waiting until an object is
+     * Gets the next value from the buffer, waiting until an object is added if the buffer is empty.
-     * added if the buffer is empty.
      *
      * @throws BufferUnderflowException if an interrupt is received
      */
     public Object get() {
-        synchronized (lock) {
+        synchronized( lock ) {
-            while (collection.isEmpty()) {
+            while( collection.isEmpty() ) {
                 try {
+                    if( timeout <= 0 ) {
                         lock.wait();
-                } catch (InterruptedException e) {
+                    }
-                    PrintWriter out = new PrintWriter(new StringWriter());
+                    else {
-                    e.printStackTrace(out);
+                        return get( timeout );
-                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
+                    }
+                }
+                catch( InterruptedException e ) {
+                    PrintWriter out = new PrintWriter( new StringWriter() );
+                    e.printStackTrace( out );
+                    throw new BufferUnderflowException( "Caused by InterruptedException: " + out.toString() );
                 }
             }
             return getBuffer().get();
@@ -111,50 +142,56 @@ public class BlockingBuffer extends SynchronizedBuffer {
     }
 
     /**
-     * Gets the next value from the buffer, waiting until an object is
+     * Gets the next value from the buffer, waiting until an object is added for up to the specified timeout value if
-     * added for up to the specified timeout value if the buffer is empty.
+     * the buffer is empty.
      *
      * @param timeout the timeout value in milliseconds
      * @throws BufferUnderflowException if an interrupt is received
      * @throws BufferUnderflowException if the timeout expires
      * @since Commons Collections 3.2
      */
-    public Object get(final long timeout) {
+    public Object get( final long timeout ) {
-        synchronized (lock) {
+        synchronized( lock ) {
             final long expiration = System.currentTimeMillis() + timeout;
             long timeLeft = expiration - System.currentTimeMillis();
-            while (timeLeft > 0 && collection.isEmpty()) {
+            while( timeLeft > 0 && collection.isEmpty() ) {
                 try {
-                    lock.wait(timeLeft);
+                    lock.wait( timeLeft );
                     timeLeft = expiration - System.currentTimeMillis();
-                } catch(InterruptedException e) {
+                }
-                    PrintWriter out = new PrintWriter(new StringWriter());
+                catch( InterruptedException e ) {
-                    e.printStackTrace(out);
+                    PrintWriter out = new PrintWriter( new StringWriter() );
-                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
+                    e.printStackTrace( out );
+                    throw new BufferUnderflowException( "Caused by InterruptedException: " + out.toString() );
                 }
             }
-            if (collection.isEmpty()) {
+            if( collection.isEmpty() ) {
-                throw new BufferUnderflowException("Timeout expired.");
+                throw new BufferUnderflowException( "Timeout expired." );
             }
             return getBuffer().get();
         }
     }
 
     /**
-     * Removes the next value from the buffer, waiting until an object is
+     * Removes the next value from the buffer, waiting until an object is added if the buffer is empty.
-     * added if the buffer is empty.
      *
      * @throws BufferUnderflowException if an interrupt is received
      */
     public Object remove() {
-        synchronized (lock) {
+        synchronized( lock ) {
-            while (collection.isEmpty()) {
+            while( collection.isEmpty() ) {
                 try {
+                    if( timeout <= 0 ) {
                         lock.wait();
-                } catch (InterruptedException e) {
+                    }
-                    PrintWriter out = new PrintWriter(new StringWriter());
+                    else {
-                    e.printStackTrace(out);
+                        return remove( timeout );
-                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
+                    }
+                }
+                catch( InterruptedException e ) {
+                    PrintWriter out = new PrintWriter( new StringWriter() );
+                    e.printStackTrace( out );
+                    throw new BufferUnderflowException( "Caused by InterruptedException: " + out.toString() );
                 }
             }
             return getBuffer().remove();
@@ -162,33 +199,33 @@ public class BlockingBuffer extends SynchronizedBuffer {
     }
 
     /**
-     * Removes the next value from the buffer, waiting until an object is
+     * Removes the next value from the buffer, waiting until an object is added for up to the specified timeout value if
-     * added for up to the specified timeout value if the buffer is empty.
+     * the buffer is empty.
      *
      * @param timeout the timeout value in milliseconds
      * @throws BufferUnderflowException if an interrupt is received
      * @throws BufferUnderflowException if the timeout expires
      * @since Commons Collections 3.2
      */
-    public Object remove(final long timeout) {
+    public Object remove( final long timeout ) {
-        synchronized (lock) {
+        synchronized( lock ) {
             final long expiration = System.currentTimeMillis() + timeout;
             long timeLeft = expiration - System.currentTimeMillis();
-            while (timeLeft > 0 && collection.isEmpty()) {
+            while( timeLeft > 0 && collection.isEmpty() ) {
                 try {
-                    lock.wait(timeLeft);
+                    lock.wait( timeLeft );
                     timeLeft = expiration - System.currentTimeMillis();
-                } catch(InterruptedException e) {
+                }
-                    PrintWriter out = new PrintWriter(new StringWriter());
+                catch( InterruptedException e ) {
-                    e.printStackTrace(out);
+                    PrintWriter out = new PrintWriter( new StringWriter() );
-                    throw new BufferUnderflowException("Caused by InterruptedException: " + out.toString());
+                    e.printStackTrace( out );
+                    throw new BufferUnderflowException( "Caused by InterruptedException: " + out.toString() );
                 }
             }
-            if (collection.isEmpty()) {
+            if( collection.isEmpty() ) {
-                throw new BufferUnderflowException("Timeout expired.");
+                throw new BufferUnderflowException( "Timeout expired." );
             }
             return getBuffer().remove();
         }
     }
-
 }

src/test/org/apache/commons/collections/buffer/TestBlockingBuffer.java

@@ -15,46 +15,43 @@
  */
 package org.apache.commons.collections.buffer;
 
+import junit.framework.Test;
+import junit.framework.TestSuite;
+import org.apache.commons.collections.AbstractTestObject;
+import org.apache.commons.collections.Buffer;
+import org.apache.commons.collections.BufferUnderflowException;
+
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.Set;
 
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.commons.collections.AbstractTestObject;
-import org.apache.commons.collections.Buffer;
-import org.apache.commons.collections.BufferUnderflowException;
-
 /**
- * Extension of {@link TestObject} for exercising the {@link BlockingBuffer}
+ * Extension of {@link AbstractTestObject} for exercising the {@link BlockingBuffer} implementation.
- * implementation.
- *
- * @since Commons Collections 3.0
- * @version $Revision$
  *
  * @author Janek Bogucki
  * @author Phil Steitz
+ * @version $Revision$
+ * @since Commons Collections 3.0
  */
 public class TestBlockingBuffer extends AbstractTestObject {
 
-    public TestBlockingBuffer(String testName) {
+    public TestBlockingBuffer( String testName ) {
-        super(testName);
+        super( testName );
     }
 
     public static Test suite() {
-        return new TestSuite(TestBlockingBuffer.class);
+        return new TestSuite( TestBlockingBuffer.class );
     }
 
-    public static void main(String args[]) {
+    public static void main( String args[] ) {
-        String[] testCaseName = { TestBlockingBuffer.class.getName()};
+        String[] testCaseName = {TestBlockingBuffer.class.getName()};
-        junit.textui.TestRunner.main(testCaseName);
+        junit.textui.TestRunner.main( testCaseName );
     }
 
     public Object makeObject() {
-        return BlockingBuffer.decorate(new MyBuffer());
+        return BlockingBuffer.decorate( new MyBuffer() );
     }
 
     public boolean isEqualsCheckable() {
@@ -62,80 +59,114 @@ public class TestBlockingBuffer extends AbstractTestObject {
     }
 
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#add()}.
+     * Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#add(Object)}.
      */
     public void testGetWithAdd() {
-      
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
-
+        new DelayedAdd( blockingBuffer, obj ).start();
-        new DelayedAdd(blockingBuffer, obj).start();
 
         // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
-        assertSame(obj, blockingBuffer.get());
+        assertSame( obj, blockingBuffer.get() );
+    }
+
+    public void testGetWithAddTimeout() {
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer(), 500 );
+        Object obj = new Object();
+        new DelayedAdd( blockingBuffer, obj, 100 ).start();
+
+        // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
+        assertSame( obj, blockingBuffer.get() );
     }
 
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#addAll()}.
+     * Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)}.
      */
     public void testGetWithAddAll() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
-
+        new DelayedAddAll( blockingBuffer, obj ).start();
-        new DelayedAddAll(blockingBuffer, obj).start();
 
         // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
-        assertSame(obj, blockingBuffer.get());
+        assertSame( obj, blockingBuffer.get() );
+    }
+
+    public void testGetWithAddAllTimeout() {
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer(), 500 );
+        Object obj = new Object();
+        new DelayedAddAll( blockingBuffer, obj, 100 ).start();
+
+        // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
+        assertSame( obj, blockingBuffer.get() );
     }
 
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#add()}.
+     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#add(Object)}.
      */
     public void testRemoveWithAdd() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
-
+        new DelayedAdd( blockingBuffer, obj ).start();
-        new DelayedAdd(blockingBuffer, obj).start();
 
         // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
-        assertSame(obj, blockingBuffer.remove());
+        assertSame( obj, blockingBuffer.remove() );
     }
 
+    public void testRemoveWithAddTimeout() {
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer(), 100 );
+        Object obj = new Object();
+        new DelayedAdd( blockingBuffer, obj, 500 ).start();
+        try {
+            blockingBuffer.remove();
+        }
+        catch( BufferUnderflowException e ) {
+        }
+    }
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll()}.
+     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)}.
      */
     public void testRemoveWithAddAll() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
-
+        new DelayedAddAll( blockingBuffer, obj ).start();
-        new DelayedAddAll(blockingBuffer, obj).start();
 
         // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
-        assertSame(obj, blockingBuffer.remove());
+        assertSame( obj, blockingBuffer.remove() );
     }
 
+    public void testRemoveWithAddAllTimeout() {
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer(), 100 );
+        Object obj = new Object();
+        new DelayedAddAll( blockingBuffer, obj, 500 ).start();
+        try {
+            blockingBuffer.remove();
+        }
+        catch( BufferUnderflowException e ) {
+        }
+    }
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#add()} using multiple read threads.
+     * Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#add(Object)} using multiple read
-     *
+     * threads.
-     *  Two read threads should block on an empty buffer until one object
+     * <p/>
-     *  is added then both threads should complete.
+     * Two read threads should block on an empty buffer until one object is added then both threads should complete.
      */
     public void testBlockedGetWithAdd() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
 
         // run methods will get and compare -- must wait for add
-        Thread thread1 = new ReadThread(blockingBuffer, obj);
+        Thread thread1 = new ReadThread( blockingBuffer, obj );
-        Thread thread2 = new ReadThread(blockingBuffer, obj);
+        Thread thread2 = new ReadThread( blockingBuffer, obj );
         thread1.start();
         thread2.start();
 
@@ -143,31 +174,32 @@ public class TestBlockingBuffer extends AbstractTestObject {
         delay();
 
         // notifyAll should allow both read threads to complete
-        blockingBuffer.add(obj);
+        blockingBuffer.add( obj );
 
         // allow notified threads to complete 
         delay();
 
         // There should not be any threads waiting.
-        if (thread1.isAlive() || thread2.isAlive())
+        if( thread1.isAlive() || thread2.isAlive() ) {
-            fail("Live thread(s) when both should be dead.");
+            fail( "Live thread(s) when both should be dead." );
+        }
     }
 
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#addAll()} using multiple read threads.
+     * Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)} using
-     *
+     * multiple read threads.
-     *  Two read threads should block on an empty buffer until a
+     * <p/>
-     *  singleton is added then both threads should complete.
+     * Two read threads should block on an empty buffer until a singleton is added then both threads should complete.
      */
     public void testBlockedGetWithAddAll() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
 
         // run methods will get and compare -- must wait for addAll
-        Thread thread1 = new ReadThread(blockingBuffer, obj);
+        Thread thread1 = new ReadThread( blockingBuffer, obj );
-        Thread thread2 = new ReadThread(blockingBuffer, obj);
+        Thread thread2 = new ReadThread( blockingBuffer, obj );
         thread1.start();
         thread2.start();
 
@@ -175,28 +207,29 @@ public class TestBlockingBuffer extends AbstractTestObject {
         delay();
 
         // notifyAll should allow both read threads to complete
-        blockingBuffer.addAll(Collections.singleton(obj));
+        blockingBuffer.addAll( Collections.singleton( obj ) );
 
         // allow notified threads to complete 
         delay();
 
         // There should not be any threads waiting.
-        if (thread1.isAlive() || thread2.isAlive())
+        if( thread1.isAlive() || thread2.isAlive() ) {
-            fail("Live thread(s) when both should be dead.");
+            fail( "Live thread(s) when both should be dead." );
+        }
     }
 
     //-----------------------------------------------------------------------
+
     /**
      * Tests interrupted {@link BlockingBuffer#get()}.
      */
     public void testInterruptedGet() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
 
         // spawn a read thread to wait on the empty buffer
         ArrayList exceptionList = new ArrayList();
-        Thread thread = new ReadThread(blockingBuffer, obj, exceptionList);
+        Thread thread = new ReadThread( blockingBuffer, obj, exceptionList );
         thread.start();
 
         // Interrupting the thread should cause it to throw BufferUnderflowException
@@ -204,150 +237,141 @@ public class TestBlockingBuffer extends AbstractTestObject {
 
         // Chill, so thread can throw and add message to exceptionList
         delay();
-        
+        assertTrue( "Thread interrupt should have led to underflow",
-        assertTrue("Thread interrupt should have led to underflow", 
+                    exceptionList.contains( "BufferUnderFlow" ) );
-            exceptionList.contains("BufferUnderFlow"));
+        if( thread.isAlive() ) {
-        
+            fail( "Read thread has hung." );
-        if (thread.isAlive()) {
-            fail("Read thread has hung.");
         }
 
     }
 
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#add()} using multiple read threads.
+     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#add(Object)} using multiple read
-     *
+     * threads.
-     *  Two read threads should block on an empty buffer until one
+     * <p/>
-     *  object is added then one thread should complete. The remaining
+     * Two read threads should block on an empty buffer until one object is added then one thread should complete. The
-     *  thread should complete after the addition of a second object.
+     * remaining thread should complete after the addition of a second object.
      */
     public void testBlockedRemoveWithAdd() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
 
         // run methods will remove and compare -- must wait for add
-        Thread thread1 = new ReadThread(blockingBuffer, obj, null, "remove");
+        Thread thread1 = new ReadThread( blockingBuffer, obj, null, "remove" );
-        Thread thread2 = new ReadThread(blockingBuffer, obj, null, "remove");
+        Thread thread2 = new ReadThread( blockingBuffer, obj, null, "remove" );
         thread1.start();
         thread2.start();
 
         // give hungry read threads ample time to hang
         delay();
-           
+        blockingBuffer.add( obj );
-        blockingBuffer.add(obj);
 
         // allow notified threads to complete 
         delay();
 
         // There should be one thread waiting.
-        assertTrue ("There is one thread waiting", thread1.isAlive() ^ thread2.isAlive());
+        assertTrue( "There is one thread waiting", thread1.isAlive() ^ thread2.isAlive() );
-           
+        blockingBuffer.add( obj );
-        blockingBuffer.add(obj);
 
         // allow notified thread to complete 
         delay();
 
         // There should not be any threads waiting.
-        if(thread1.isAlive() || thread2.isAlive())
+        if( thread1.isAlive() || thread2.isAlive() ) {
-            fail("Live thread(s) when both should be dead.");
+            fail( "Live thread(s) when both should be dead." );
+        }
     }
 
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll()} using multiple read threads.
+     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)}
-     *
+     * using multiple read threads.
-     *  Two read threads should block on an empty buffer until a
+     * <p/>
-     *  singleton collection is added then one thread should
+     * Two read threads should block on an empty buffer until a singleton collection is added then one thread should
-     *  complete. The remaining thread should complete after the
+     * complete. The remaining thread should complete after the addition of a second singleton.
-     *  addition of a second singleton.
      */
     public void testBlockedRemoveWithAddAll1() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
 
         // run methods will remove and compare -- must wait for addAll
-        Thread thread1 = new ReadThread(blockingBuffer, obj, null, "remove");
+        Thread thread1 = new ReadThread( blockingBuffer, obj, null, "remove" );
-        Thread thread2 = new ReadThread(blockingBuffer, obj, null, "remove");
+        Thread thread2 = new ReadThread( blockingBuffer, obj, null, "remove" );
         thread1.start();
         thread2.start();
 
         // give hungry read threads ample time to hang
         delay();
-           
+        blockingBuffer.addAll( Collections.singleton( obj ) );
-        blockingBuffer.addAll(Collections.singleton(obj));
 
         // allow notified threads to complete 
         delay();
 
         // There should be one thread waiting.
-        assertTrue ("There is one thread waiting", thread1.isAlive() ^ thread2.isAlive());
+        assertTrue( "There is one thread waiting", thread1.isAlive() ^ thread2.isAlive() );
-           
+        blockingBuffer.addAll( Collections.singleton( obj ) );
-        blockingBuffer.addAll(Collections.singleton(obj));
 
         // allow notified thread to complete 
         delay();
 
         // There should not be any threads waiting.
-        if(thread1.isAlive() || thread2.isAlive())
+        if( thread1.isAlive() || thread2.isAlive() ) {
-            fail("Live thread(s) when both should be dead.");
+            fail( "Live thread(s) when both should be dead." );
+        }
     }
 
-   
     //-----------------------------------------------------------------------
+
     /**
-     *  Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll()} using multiple read threads.
+     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)}
-     *
+     * using multiple read threads.
-     *  Two read threads should block on an empty buffer until a
+     * <p/>
-     *  collection with two distinct objects is added then both
+     * Two read threads should block on an empty buffer until a collection with two distinct objects is added then both
-     *  threads should complete. Each thread should have read a
+     * threads should complete. Each thread should have read a different object.
-     *  different object.
      */
     public void testBlockedRemoveWithAddAll2() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj1 = new Object();
         Object obj2 = new Object();
-        
+        Set objs = Collections.synchronizedSet( new HashSet() );
-        Set objs = Collections.synchronizedSet(new HashSet());
+        objs.add( obj1 );
-        objs.add(obj1);
+        objs.add( obj2 );
-        objs.add(obj2);
 
         // run methods will remove and compare -- must wait for addAll
-        Thread thread1 = new ReadThread(blockingBuffer, objs, "remove");
+        Thread thread1 = new ReadThread( blockingBuffer, objs, "remove" );
-        Thread thread2 = new ReadThread(blockingBuffer, objs, "remove");
+        Thread thread2 = new ReadThread( blockingBuffer, objs, "remove" );
         thread1.start();
         thread2.start();
 
         // give hungry read threads ample time to hang
         delay();
-           
+        blockingBuffer.addAll( objs );
-        blockingBuffer.addAll(objs);
 
         // allow notified threads to complete 
         delay();
-        
+        assertEquals( "Both objects were removed", 0, objs.size() );
-        assertEquals("Both objects were removed", 0, objs.size());
 
         // There should not be any threads waiting.
-        if(thread1.isAlive() || thread2.isAlive())
+        if( thread1.isAlive() || thread2.isAlive() ) {
-            fail("Live thread(s) when both should be dead.");
+            fail( "Live thread(s) when both should be dead." );
+        }
     }
 
     //-----------------------------------------------------------------------
+
     /**
      * Tests interrupted remove.
      */
     public void testInterruptedRemove() {
-        
+        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
-        Buffer blockingBuffer = BlockingBuffer.decorate(new MyBuffer());
         Object obj = new Object();
 
         // spawn a read thread to wait on the empty buffer
         ArrayList exceptionList = new ArrayList();
-        Thread thread = new ReadThread(blockingBuffer, obj, exceptionList, "remove");
+        Thread thread = new ReadThread( blockingBuffer, obj, exceptionList, "remove" );
         thread.start();
 
         // Interrupting the thread should cause it to throw BufferUnderflowException
@@ -355,103 +379,122 @@ public class TestBlockingBuffer extends AbstractTestObject {
 
         // Chill, so thread can throw and add message to exceptionList
         delay();
-        
+        assertTrue( "Thread interrupt should have led to underflow",
-        assertTrue("Thread interrupt should have led to underflow", 
+                    exceptionList.contains( "BufferUnderFlow" ) );
-            exceptionList.contains("BufferUnderFlow"));
+        if( thread.isAlive() ) {
-        
+            fail( "Read thread has hung." );
-        if (thread.isAlive()) {
-            fail("Read thread has hung.");
         }
 
     }
 
     public void testTimeoutGet() {
-        final BlockingBuffer buffer = new BlockingBuffer(new MyBuffer());
+        final BlockingBuffer buffer = new BlockingBuffer( new MyBuffer() );
         try {
             buffer.get( 100 );
             fail( "Get should have timed out." );
         }
-        catch( BufferUnderflowException e ){
+        catch( BufferUnderflowException e ) {
         }
     }
 
     public void testTimeoutRemove() {
-        final BlockingBuffer buffer = new BlockingBuffer(new MyBuffer());
+        final BlockingBuffer buffer = new BlockingBuffer( new MyBuffer() );
         try {
             buffer.remove( 100 );
             fail( "Get should have timed out." );
         }
-        catch( BufferUnderflowException e ){
+        catch( BufferUnderflowException e ) {
         }
     }
+
     protected static class DelayedAdd extends Thread {
 
         Buffer buffer;
+
         Object obj;
 
-        DelayedAdd (Buffer buffer, Object obj) {
+        long delay = 1000;
+
+        public DelayedAdd( Buffer buffer, Object obj, long delay ) {
+            this.buffer = buffer;
+            this.obj = obj;
+            this.delay = delay;
+        }
+
+        DelayedAdd( Buffer buffer, Object obj ) {
             super();
             this.buffer = buffer;
             this.obj = obj;
         }
 
         public void run() {
-
             try {
                 // wait for other thread to block on get() or remove()
-                Thread.sleep(100);
+                Thread.sleep( delay );
             }
-            catch (InterruptedException e) {}
+            catch( InterruptedException e ) {
-
+            }
-            buffer.add(obj);
+            buffer.add( obj );
         }
     }
 
     protected static class DelayedAddAll extends Thread {
 
         Buffer buffer;
+
         Object obj;
 
-        DelayedAddAll (Buffer buffer, Object obj) {
+        long delay = 100;
+
+        public DelayedAddAll( Buffer buffer, Object obj, long delay ) {
+            this.buffer = buffer;
+            this.obj = obj;
+            this.delay = delay;
+        }
+
+        DelayedAddAll( Buffer buffer, Object obj ) {
             super();
             this.buffer = buffer;
             this.obj = obj;
         }
 
         public void run() {
-
             try {
                 // wait for other thread to block on get() or remove()
-                Thread.sleep(100);
+                Thread.sleep( delay );
             }
-            catch (InterruptedException e) {}
+            catch( InterruptedException e ) {
-
+            }
-            buffer.addAll(Collections.singleton(obj));
+            buffer.addAll( Collections.singleton( obj ) );
         }
     }
 
     protected static class ReadThread extends Thread {
 
         Buffer buffer;
+
         Object obj;
+
         ArrayList exceptionList = null;
+
         String action = "get";
+
         Set objs;
 
-        ReadThread (Buffer buffer, Object obj) {
+        ReadThread( Buffer buffer, Object obj ) {
             super();
             this.buffer = buffer;
             this.obj = obj;
         }
 
-        ReadThread (Buffer buffer, Object obj, ArrayList exceptionList) {
+        ReadThread( Buffer buffer, Object obj, ArrayList exceptionList ) {
             super();
             this.buffer = buffer;
             this.obj = obj;
             this.exceptionList = exceptionList;
         }
 
-        ReadThread (Buffer buffer, Object obj, ArrayList exceptionList, String action) {
+        ReadThread( Buffer buffer, Object obj, ArrayList exceptionList, String action ) {
             super();
             this.buffer = buffer;
             this.obj = obj;
@@ -459,7 +502,7 @@ public class TestBlockingBuffer extends AbstractTestObject {
             this.action = action;
         }
 
-        ReadThread (Buffer buffer, Set objs, String action) {
+        ReadThread( Buffer buffer, Set objs, String action ) {
             super();
             this.buffer = buffer;
             this.objs = objs;
@@ -468,40 +511,47 @@ public class TestBlockingBuffer extends AbstractTestObject {
 
         public void run() {
             try {
-                if (action == "get") {
+                if( action == "get" ) {
-                    assertSame(obj, buffer.get());
+                    assertSame( obj, buffer.get() );
-                } else {
-                    if (null != obj)
-                        assertSame(obj, buffer.remove());
-                    else
-                        assertTrue(objs.remove(buffer.remove()));
                 }
-            } catch (BufferUnderflowException ex) {
+                else {
-                exceptionList.add("BufferUnderFlow");
+                    if( null != obj ) {
+                        assertSame( obj, buffer.remove() );
+                    }
+                    else {
+                        assertTrue( objs.remove( buffer.remove() ) );
+                    }
+                }
+            }
+            catch( BufferUnderflowException ex ) {
+                exceptionList.add( "BufferUnderFlow" );
             }
         }
     }
-        
 
     protected static class MyBuffer extends LinkedList implements Buffer {
 
         public Object get() {
-            if(isEmpty())
+            if( isEmpty() ) {
                 throw new BufferUnderflowException();
-            return get(0);
+            }
+            return get( 0 );
         }
 
         public Object remove() {
-            if(isEmpty())
+            if( isEmpty() ) {
                 throw new BufferUnderflowException();
-            return remove(0);
+            }
+            return remove( 0 );
         }
     }
 
-    private void delay(){
+    private void delay() {
         try {
-            Thread.sleep(100);
+            Thread.sleep( 100 );
-        } catch (InterruptedException e) {}
+        }
+        catch( InterruptedException e ) {
+        }
     }
 
     public String getCompatibilityVersion() {
@@ -510,12 +560,13 @@ public class TestBlockingBuffer extends AbstractTestObject {
 
 //    public void testCreate() throws Exception {
 //        Buffer buffer = BlockingBuffer.decorate(new UnboundedFifoBuffer());
-//        writeExternalFormToDisk((java.io.Serializable) buffer, "D:/dev/collections/data/test/BlockingBuffer.emptyCollection.version3.1.obj");
+//        writeExternalFormToDisk((java.io.Serializable) buffer,
+//        "D:/dev/collections/data/test/BlockingBuffer.emptyCollection.version3.1.obj");
 //        buffer = BlockingBuffer.decorate(new UnboundedFifoBuffer());
 //        buffer.add("A");
 //        buffer.add("B");
 //        buffer.add("C");
-//        writeExternalFormToDisk((java.io.Serializable) buffer, "D:/dev/collections/data/test/BlockingBuffer.fullCollection.version3.1.obj");
+//        writeExternalFormToDisk((java.io.Serializable) buffer,
+//        "D:/dev/collections/data/test/BlockingBuffer.fullCollection.version3.1.obj");
 //    }
-
 }