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