Commons Collections example source code file (TestBlockingBuffer.java)

This example Commons Collections source code file (TestBlockingBuffer.java) is included in the DevDaily.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Java - Commons Collections tags/keywords

arraylist, buffer, buffer, bufferunderflowexception, delayedadd, delayedaddall, mybuffer, mybuffer, object, object, readthread, string, thread, thread, util

The Commons Collections TestBlockingBuffer.java source code

/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
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;

/**
 * Extension of {@link AbstractTestObject} for exercising the {@link BlockingBuffer} implementation.
 *
 * @author Janek Bogucki
 * @author Phil Steitz
 * @version $Revision: 646780 $
 * @since Commons Collections 3.0
 */
public class TestBlockingBuffer extends AbstractTestObject {

    public TestBlockingBuffer( String testName ) {
        super( testName );
    }

    public static Test suite() {
        return new TestSuite( TestBlockingBuffer.class );
    }

    public static void main( String args[] ) {
        String[] testCaseName = {TestBlockingBuffer.class.getName()};
        junit.textui.TestRunner.main( testCaseName );
    }

    public Object makeObject() {
        return BlockingBuffer.decorate( new MyBuffer() );
    }

    public boolean isEqualsCheckable() {
        return false;
    }

    //-----------------------------------------------------------------------

    /**
     * Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#add(Object)}.
     */
    public void testGetWithAdd() {
        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
        Object obj = new Object();
        new DelayedAdd( blockingBuffer, obj ).start();

        // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
        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(java.util.Collection)}.
     */
    public void testGetWithAddAll() {
        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
        Object obj = new Object();
        new DelayedAddAll( blockingBuffer, obj ).start();

        // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
        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(Object)}.
     */
    public void testRemoveWithAdd() {
        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
        Object obj = new Object();
        new DelayedAdd( blockingBuffer, obj ).start();

        // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
        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(java.util.Collection)}.
     */
    public void testRemoveWithAddAll() {
        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
        Object obj = new Object();
        new DelayedAddAll( blockingBuffer, obj ).start();

        // verify does not throw BufferUnderflowException; should block until other thread has added to the buffer .
        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(Object)} using multiple read
     * threads.
     * <p/>
     * 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() );
        Object obj = new Object();

        // run methods will get and compare -- must wait for add
        Thread thread1 = new ReadThread( blockingBuffer, obj );
        Thread thread2 = new ReadThread( blockingBuffer, obj );
        thread1.start();
        thread2.start();

        // give hungry read threads ample time to hang
        delay();

        // notifyAll should allow both read threads to complete
        blockingBuffer.add( obj );

        // allow notified threads to complete 
        delay();

        // There should not be any threads waiting.
        if( thread1.isAlive() || thread2.isAlive() ) {
            fail( "Live thread(s) when both should be dead." );
        }
    }

    //-----------------------------------------------------------------------

    /**
     * Tests {@link BlockingBuffer#get()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)} using
     * multiple read threads.
     * <p/>
     * 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() );
        Object obj = new Object();

        // run methods will get and compare -- must wait for addAll
        Thread thread1 = new ReadThread( blockingBuffer, obj );
        Thread thread2 = new ReadThread( blockingBuffer, obj );
        thread1.start();
        thread2.start();

        // give hungry read threads ample time to hang
        delay();

        // notifyAll should allow both read threads to complete
        blockingBuffer.addAll( Collections.singleton( obj ) );

        // allow notified threads to complete 
        delay();

        // There should not be any threads waiting.
        if( thread1.isAlive() || thread2.isAlive() ) {
            fail( "Live thread(s) when both should be dead." );
        }
    }

    //-----------------------------------------------------------------------

    /**
     * Tests interrupted {@link BlockingBuffer#get()}.
     */
    public void testInterruptedGet() {
        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.start();

        // Interrupting the thread should cause it to throw BufferUnderflowException
        thread.interrupt();

        // Chill, so thread can throw and add message to exceptionList
        delay();
        assertTrue( "Thread interrupt should have led to underflow",
                    exceptionList.contains( "BufferUnderFlow" ) );
        if( thread.isAlive() ) {
            fail( "Read thread has hung." );
        }

    }

    //-----------------------------------------------------------------------

    /**
     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#add(Object)} using multiple read
     * threads.
     * <p/>
     * Two read threads should block on an empty buffer until one object is added then one thread should complete. The
     * remaining thread should complete after the addition of a second object.
     */
    public void testBlockedRemoveWithAdd() {
        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 thread2 = new ReadThread( blockingBuffer, obj, null, "remove" );
        thread1.start();
        thread2.start();

        // give hungry read threads ample time to hang
        delay();
        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() );
        blockingBuffer.add( obj );

        // allow notified thread to complete 
        delay();

        // There should not be any threads waiting.
        if( thread1.isAlive() || thread2.isAlive() ) {
            fail( "Live thread(s) when both should be dead." );
        }
    }

    //-----------------------------------------------------------------------

    /**
     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)}
     * using multiple read threads.
     * <p/>
     * 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 addition of a second singleton.
     */
    public void testBlockedRemoveWithAddAll1() {
        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 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 ) );

        // allow notified threads to complete 
        delay();

        // There should be one thread waiting.
        assertTrue( "There is one thread waiting", thread1.isAlive() ^ thread2.isAlive() );
        blockingBuffer.addAll( Collections.singleton( obj ) );

        // allow notified thread to complete 
        delay();

        // There should not be any threads waiting.
        if( thread1.isAlive() || thread2.isAlive() ) {
            fail( "Live thread(s) when both should be dead." );
        }
    }

    //-----------------------------------------------------------------------

    /**
     * Tests {@link BlockingBuffer#remove()} in combination with {@link BlockingBuffer#addAll(java.util.Collection)}
     * using multiple read threads.
     * <p/>
     * 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 different object.
     */
    public void testBlockedRemoveWithAddAll2() {
        Buffer blockingBuffer = BlockingBuffer.decorate( new MyBuffer() );
        Object obj1 = new Object();
        Object obj2 = new Object();
        Set objs = Collections.synchronizedSet( new HashSet() );
        objs.add( obj1 );
        objs.add( obj2 );

        // run methods will remove and compare -- must wait for addAll
        Thread thread1 = 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 );

        // allow notified threads to complete 
        delay();
        assertEquals( "Both objects were removed", 0, objs.size() );

        // There should not be any threads waiting.
        if( thread1.isAlive() || thread2.isAlive() ) {
            fail( "Live thread(s) when both should be dead." );
        }
    }

    //-----------------------------------------------------------------------

    /**
     * Tests interrupted remove.
     */
    public void testInterruptedRemove() {
        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.start();

        // Interrupting the thread should cause it to throw BufferUnderflowException
        thread.interrupt();

        // Chill, so thread can throw and add message to exceptionList
        delay();
        assertTrue( "Thread interrupt should have led to underflow",
                    exceptionList.contains( "BufferUnderFlow" ) );
        if( thread.isAlive() ) {
            fail( "Read thread has hung." );
        }

    }

    public void testTimeoutGet() {
        final BlockingBuffer buffer = new BlockingBuffer( new MyBuffer() );
        try {
            buffer.get( 100 );
            fail( "Get should have timed out." );
        }
        catch( BufferUnderflowException e ) {
        }
    }

    public void testTimeoutRemove() {
        final BlockingBuffer buffer = new BlockingBuffer( new MyBuffer() );
        try {
            buffer.remove( 100 );
            fail( "Get should have timed out." );
        }
        catch( BufferUnderflowException e ) {
        }
    }

    protected static class DelayedAdd extends Thread {

        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( delay );
            }
            catch( InterruptedException e ) {
            }
            buffer.add( obj );
        }
    }

    protected static class DelayedAddAll extends Thread {

        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( delay );
            }
            catch( InterruptedException e ) {
            }
            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 ) {
            super();
            this.buffer = buffer;
            this.obj = obj;
        }

        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 ) {
            super();
            this.buffer = buffer;
            this.obj = obj;
            this.exceptionList = exceptionList;
            this.action = action;
        }

        ReadThread( Buffer buffer, Set objs, String action ) {
            super();
            this.buffer = buffer;
            this.objs = objs;
            this.action = action;
        }

        public void run() {
            try {
                if( action == "get" ) {
                    assertSame( obj, buffer.get() );
                }
                else {
                    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() ) {
                throw new BufferUnderflowException();
            }
            return get( 0 );
        }

        public Object remove() {
            if( isEmpty() ) {
                throw new BufferUnderflowException();
            }
            return remove( 0 );
        }
    }

    private void delay() {
        try {
            Thread.sleep( 100 );
        }
        catch( InterruptedException e ) {
        }
    }

    public String getCompatibilityVersion() {
        return "3.1";
    }

//    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");
//        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");
//    }
}