--- jsr166/src/test/tck/SynchronousQueueTest.java 2003/10/05 23:00:40 1.6 +++ jsr166/src/test/tck/SynchronousQueueTest.java 2011/05/06 11:22:07 1.33 @@ -1,56 +1,85 @@ /* - * Written by members of JCP JSR-166 Expert Group and released to the - * public domain. Use, modify, and redistribute this code in any way - * without acknowledgement. Other contributors include Andrew Wright, - * Jeffrey Hayes, Pat Fischer, Mike Judd. + * Written by Doug Lea with assistance from members of JCP JSR-166 + * Expert Group and released to the public domain, as explained at + * http://creativecommons.org/publicdomain/zero/1.0/ + * Other contributors include Andrew Wright, Jeffrey Hayes, + * Pat Fisher, Mike Judd. */ import junit.framework.*; import java.util.*; import java.util.concurrent.*; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import java.io.*; public class SynchronousQueueTest extends JSR166TestCase { + public static class Fair extends BlockingQueueTest { + protected BlockingQueue emptyCollection() { + return new SynchronousQueue(true); + } + } + + public static class NonFair extends BlockingQueueTest { + protected BlockingQueue emptyCollection() { + return new SynchronousQueue(false); + } + } + public static void main(String[] args) { - junit.textui.TestRunner.run (suite()); + junit.textui.TestRunner.run(suite()); } public static Test suite() { - return new TestSuite(SynchronousQueueTest.class); + return newTestSuite(SynchronousQueueTest.class, + new Fair().testSuite(), + new NonFair().testSuite()); } /** - * A SynchronousQueue is both empty and full + * Any SynchronousQueue is both empty and full */ - public void testEmptyFull() { - SynchronousQueue q = new SynchronousQueue(); + public void testEmptyFull(SynchronousQueue q) { assertTrue(q.isEmpty()); - assertEquals(0, q.size()); + assertEquals(0, q.size()); assertEquals(0, q.remainingCapacity()); assertFalse(q.offer(zero)); } /** + * A non-fair SynchronousQueue is both empty and full + */ + public void testEmptyFull() { + testEmptyFull(new SynchronousQueue()); + } + + /** + * A fair SynchronousQueue is both empty and full + */ + public void testFairEmptyFull() { + testEmptyFull(new SynchronousQueue(true)); + } + + /** * offer(null) throws NPE */ public void testOfferNull() { - try { + try { SynchronousQueue q = new SynchronousQueue(); q.offer(null); shouldThrow(); - } catch (NullPointerException success) { } + } catch (NullPointerException success) {} } /** * add(null) throws NPE */ public void testAddNull() { - try { + try { SynchronousQueue q = new SynchronousQueue(); q.add(null); shouldThrow(); - } catch (NullPointerException success) { } + } catch (NullPointerException success) {} } /** @@ -65,13 +94,12 @@ public class SynchronousQueueTest extend * add throws ISE if no active taker */ public void testAdd() { - try { + try { SynchronousQueue q = new SynchronousQueue(); assertEquals(0, q.remainingCapacity()); q.add(one); shouldThrow(); - } catch (IllegalStateException success){ - } + } catch (IllegalStateException success) {} } /** @@ -82,8 +110,7 @@ public class SynchronousQueueTest extend SynchronousQueue q = new SynchronousQueue(); q.addAll(null); shouldThrow(); - } - catch (NullPointerException success) {} + } catch (NullPointerException success) {} } /** @@ -94,8 +121,7 @@ public class SynchronousQueueTest extend SynchronousQueue q = new SynchronousQueue(); q.addAll(q); shouldThrow(); - } - catch (IllegalArgumentException success) {} + } catch (IllegalArgumentException success) {} } /** @@ -107,9 +133,9 @@ public class SynchronousQueueTest extend Integer[] ints = new Integer[1]; q.addAll(Arrays.asList(ints)); shouldThrow(); - } - catch (NullPointerException success) {} + } catch (NullPointerException success) {} } + /** * addAll throws ISE if no active taker */ @@ -121,249 +147,289 @@ public class SynchronousQueueTest extend ints[i] = new Integer(i); q.addAll(Arrays.asList(ints)); shouldThrow(); - } - catch (IllegalStateException success) {} + } catch (IllegalStateException success) {} } /** * put(null) throws NPE */ - public void testPutNull() { - try { + public void testPutNull() throws InterruptedException { + try { SynchronousQueue q = new SynchronousQueue(); q.put(null); shouldThrow(); - } - catch (NullPointerException success){ - } - catch (InterruptedException ie) { - unexpectedException(); - } - } + } catch (NullPointerException success) {} + } /** * put blocks interruptibly if no active taker */ - public void testBlockingPut() { - Thread t = new Thread(new Runnable() { - public void run() { - try { - SynchronousQueue q = new SynchronousQueue(); - q.put(zero); - threadShouldThrow(); - } catch (InterruptedException ie){ - } - }}); + public void testBlockingPut() throws InterruptedException { + Thread t = new Thread(new CheckedInterruptedRunnable() { + public void realRun() throws InterruptedException { + SynchronousQueue q = new SynchronousQueue(); + q.put(zero); + }}); + t.start(); - try { - Thread.sleep(SHORT_DELAY_MS); - t.interrupt(); - t.join(); - } - catch (InterruptedException ie) { - unexpectedException(); - } + delay(SHORT_DELAY_MS); + t.interrupt(); + t.join(); } /** - * put blocks waiting for take + * put blocks waiting for take */ - public void testPutWithTake() { + public void testPutWithTake() throws InterruptedException { final SynchronousQueue q = new SynchronousQueue(); - Thread t = new Thread(new Runnable() { - public void run() { - int added = 0; - try { - q.put(new Object()); - ++added; - q.put(new Object()); - ++added; - q.put(new Object()); - ++added; - q.put(new Object()); + Thread t = new Thread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + int added = 0; + try { + while (true) { + q.put(added); ++added; - threadShouldThrow(); - } catch (InterruptedException e){ - assertTrue(added >= 1); } + } catch (InterruptedException success) { + assertEquals(1, added); } - }); - try { - t.start(); - Thread.sleep(SHORT_DELAY_MS); - q.take(); - Thread.sleep(SHORT_DELAY_MS); - t.interrupt(); - t.join(); - } catch (Exception e){ - unexpectedException(); - } + }}); + + t.start(); + delay(SHORT_DELAY_MS); + assertEquals(0, q.take()); + delay(SHORT_DELAY_MS); + t.interrupt(); + t.join(); } /** * timed offer times out if elements not taken */ - public void testTimedOffer() { - final SynchronousQueue q = new SynchronousQueue(); - Thread t = new Thread(new Runnable() { - public void run() { - try { - - threadAssertFalse(q.offer(new Object(), SHORT_DELAY_MS, TimeUnit.MILLISECONDS)); - q.offer(new Object(), LONG_DELAY_MS, TimeUnit.MILLISECONDS); - threadShouldThrow(); - } catch (InterruptedException success){} - } - }); - - try { - t.start(); - Thread.sleep(SMALL_DELAY_MS); - t.interrupt(); - t.join(); - } catch (Exception e){ - unexpectedException(); - } + public void testTimedOffer(final SynchronousQueue q) + throws InterruptedException { + final CountDownLatch pleaseInterrupt = new CountDownLatch(1); + Thread t = newStartedThread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + long t0 = System.nanoTime(); + assertFalse(q.offer(new Object(), SHORT_DELAY_MS, MILLISECONDS)); + assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS); + pleaseInterrupt.countDown(); + t0 = System.nanoTime(); + try { + q.offer(new Object(), LONG_DELAY_MS, MILLISECONDS); + shouldThrow(); + } catch (InterruptedException success) {} + assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS); + }}); + + assertTrue(pleaseInterrupt.await(MEDIUM_DELAY_MS, MILLISECONDS)); + t.interrupt(); + awaitTermination(t, MEDIUM_DELAY_MS); } + /** + * timed offer times out if elements not taken + */ + public void testTimedOffer() throws InterruptedException { + testTimedOffer(new SynchronousQueue()); + } /** - * take blocks interruptibly when empty + * timed offer times out if elements not taken */ - public void testTakeFromEmpty() { - final SynchronousQueue q = new SynchronousQueue(); - Thread t = new Thread(new Runnable() { - public void run() { - try { - q.take(); - threadShouldThrow(); - } catch (InterruptedException success){ } + public void testFairTimedOffer() throws InterruptedException { + testTimedOffer(new SynchronousQueue(true)); + } + + /** + * put blocks interruptibly if no active taker + */ + public void testFairBlockingPut() throws InterruptedException { + Thread t = new Thread(new CheckedInterruptedRunnable() { + public void realRun() throws InterruptedException { + SynchronousQueue q = new SynchronousQueue(true); + q.put(zero); + }}); + + t.start(); + delay(SHORT_DELAY_MS); + t.interrupt(); + t.join(); + } + + /** + * put blocks waiting for take + */ + public void testFairPutWithTake() throws InterruptedException { + final SynchronousQueue q = new SynchronousQueue(true); + Thread t = new Thread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + int added = 0; + try { + while (true) { + q.put(added); + ++added; + } + } catch (InterruptedException success) { + assertEquals(1, added); } - }); - try { - t.start(); - Thread.sleep(SHORT_DELAY_MS); - t.interrupt(); - t.join(); - } catch (Exception e){ - unexpectedException(); - } + }}); + + t.start(); + delay(SHORT_DELAY_MS); + assertEquals(0, q.take()); + delay(SHORT_DELAY_MS); + t.interrupt(); + t.join(); + } + + /** + * take blocks interruptibly when empty + */ + public void testFairTakeFromEmpty() throws InterruptedException { + final SynchronousQueue q = new SynchronousQueue(true); + Thread t = new Thread(new CheckedInterruptedRunnable() { + public void realRun() throws InterruptedException { + q.take(); + }}); + + t.start(); + delay(SHORT_DELAY_MS); + t.interrupt(); + t.join(); } /** - * poll fails unless active taker + * poll return null if no active putter */ public void testPoll() { SynchronousQueue q = new SynchronousQueue(); - assertNull(q.poll()); + assertNull(q.poll()); } /** - * timed pool with zero timeout times out if no active taker + * timed poll with zero timeout times out if no active putter */ - public void testTimedPoll0() { - try { - SynchronousQueue q = new SynchronousQueue(); - assertNull(q.poll(0, TimeUnit.MILLISECONDS)); - } catch (InterruptedException e){ - unexpectedException(); - } + public void testTimedPoll0() throws InterruptedException { + SynchronousQueue q = new SynchronousQueue(); + assertNull(q.poll(0, MILLISECONDS)); } /** - * timed pool with nonzero timeout times out if no active taker + * timed poll with nonzero timeout times out if no active putter */ - public void testTimedPoll() { - try { - SynchronousQueue q = new SynchronousQueue(); - assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS)); - } catch (InterruptedException e){ - unexpectedException(); - } + public void testTimedPoll() throws InterruptedException { + SynchronousQueue q = new SynchronousQueue(); + long t0 = System.nanoTime(); + assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS)); + assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS); } /** * Interrupted timed poll throws InterruptedException instead of * returning timeout status */ - public void testInterruptedTimedPoll() { - Thread t = new Thread(new Runnable() { - public void run() { - try { - SynchronousQueue q = new SynchronousQueue(); - assertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS)); - } catch (InterruptedException success){ - } - }}); - t.start(); - try { - Thread.sleep(SHORT_DELAY_MS); - t.interrupt(); - t.join(); - } - catch (InterruptedException ie) { - unexpectedException(); - } + public void testInterruptedTimedPoll(final SynchronousQueue q) + throws InterruptedException { + final CountDownLatch threadStarted = new CountDownLatch(1); + Thread t = newStartedThread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + long t0 = System.nanoTime(); + threadStarted.countDown(); + try { + q.poll(LONG_DELAY_MS, MILLISECONDS); + shouldThrow(); + } catch (InterruptedException success) {} + assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS); + assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS); + }}); + + threadStarted.await(); + delay(SHORT_DELAY_MS); + t.interrupt(); + awaitTermination(t, MEDIUM_DELAY_MS); } /** - * timed poll before a delayed offer fails; after offer succeeds; - * on interruption throws + * Interrupted timed poll throws InterruptedException instead of + * returning timeout status */ - public void testTimedPollWithOffer() { - final SynchronousQueue q = new SynchronousQueue(); - Thread t = new Thread(new Runnable() { - public void run() { - try { - threadAssertNull(q.poll(SHORT_DELAY_MS, TimeUnit.MILLISECONDS)); - q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS); - q.poll(LONG_DELAY_MS, TimeUnit.MILLISECONDS); - threadShouldThrow(); - } catch (InterruptedException success) { } - } - }); - try { - t.start(); - Thread.sleep(SMALL_DELAY_MS); - assertTrue(q.offer(zero, SHORT_DELAY_MS, TimeUnit.MILLISECONDS)); - t.interrupt(); - t.join(); - } catch (Exception e){ - unexpectedException(); - } - } + public void testInterruptedTimedPoll() throws InterruptedException { + testInterruptedTimedPoll(new SynchronousQueue()); + } + + /** + * Interrupted timed poll throws InterruptedException instead of + * returning timeout status + */ + public void testFairInterruptedTimedPoll() throws InterruptedException { + testInterruptedTimedPoll(new SynchronousQueue(true)); + } + + /** + * timed poll before a delayed offer times out, returning null; + * after offer succeeds; on interruption throws + */ + public void testFairTimedPollWithOffer() throws InterruptedException { + final SynchronousQueue q = new SynchronousQueue(true); + final CountDownLatch pleaseOffer = new CountDownLatch(1); + Thread t = newStartedThread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + long t0 = System.nanoTime(); + assertNull(q.poll(SHORT_DELAY_MS, MILLISECONDS)); + assertTrue(millisElapsedSince(t0) >= SHORT_DELAY_MS); + + pleaseOffer.countDown(); + t0 = System.nanoTime(); + assertSame(zero, q.poll(LONG_DELAY_MS, MILLISECONDS)); + assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS); + + t0 = System.nanoTime(); + try { + q.poll(LONG_DELAY_MS, MILLISECONDS); + shouldThrow(); + } catch (InterruptedException success) {} + assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS); + }}); + assertTrue(pleaseOffer.await(MEDIUM_DELAY_MS, MILLISECONDS)); + long t0 = System.nanoTime(); + assertTrue(q.offer(zero, LONG_DELAY_MS, MILLISECONDS)); + assertTrue(millisElapsedSince(t0) < MEDIUM_DELAY_MS); + + t.interrupt(); + awaitTermination(t, MEDIUM_DELAY_MS); + } /** - * peek returns null + * peek() returns null if no active putter */ public void testPeek() { SynchronousQueue q = new SynchronousQueue(); - assertNull(q.peek()); + assertNull(q.peek()); } /** - * element throws NSEE + * element() throws NSEE if no active putter */ public void testElement() { SynchronousQueue q = new SynchronousQueue(); try { q.element(); shouldThrow(); - } - catch (NoSuchElementException success) {} + } catch (NoSuchElementException success) {} } /** - * remove throws NSEE if no active taker + * remove() throws NSEE if no active putter */ public void testRemove() { SynchronousQueue q = new SynchronousQueue(); try { q.remove(); shouldThrow(); - } catch (NoSuchElementException success){ - } + } catch (NoSuchElementException success) {} } /** @@ -374,7 +440,7 @@ public class SynchronousQueueTest extend assertFalse(q.remove(zero)); assertTrue(q.isEmpty()); } - + /** * contains returns false */ @@ -431,7 +497,7 @@ public class SynchronousQueueTest extend */ public void testToArray() { SynchronousQueue q = new SynchronousQueue(); - Object[] o = q.toArray(); + Object[] o = q.toArray(); assertEquals(o.length, 0); } @@ -440,19 +506,19 @@ public class SynchronousQueueTest extend */ public void testToArray2() { SynchronousQueue q = new SynchronousQueue(); - Integer[] ints = new Integer[1]; + Integer[] ints = new Integer[1]; assertNull(ints[0]); } - + /** * toArray(null) throws NPE */ public void testToArray_BadArg() { - try { - SynchronousQueue q = new SynchronousQueue(); - Object o[] = q.toArray(null); - shouldThrow(); - } catch(NullPointerException success){} + SynchronousQueue q = new SynchronousQueue(); + try { + Object o[] = q.toArray(null); + shouldThrow(); + } catch (NullPointerException success) {} } @@ -461,13 +527,12 @@ public class SynchronousQueueTest extend */ public void testIterator() { SynchronousQueue q = new SynchronousQueue(); - Iterator it = q.iterator(); + Iterator it = q.iterator(); assertFalse(it.hasNext()); try { Object x = it.next(); shouldThrow(); - } - catch (NoSuchElementException success) {} + } catch (NoSuchElementException success) {} } /** @@ -475,12 +540,11 @@ public class SynchronousQueueTest extend */ public void testIteratorRemove() { SynchronousQueue q = new SynchronousQueue(); - Iterator it = q.iterator(); + Iterator it = q.iterator(); try { it.remove(); shouldThrow(); - } - catch (IllegalStateException success) {} + } catch (IllegalStateException success) {} } /** @@ -490,7 +554,7 @@ public class SynchronousQueueTest extend SynchronousQueue q = new SynchronousQueue(); String s = q.toString(); assertNotNull(s); - } + } /** @@ -499,35 +563,21 @@ public class SynchronousQueueTest extend public void testOfferInExecutor() { final SynchronousQueue q = new SynchronousQueue(); ExecutorService executor = Executors.newFixedThreadPool(2); - final Integer one = new Integer(1); - executor.execute(new Runnable() { - public void run() { - threadAssertFalse(q.offer(one)); - try { - threadAssertTrue(q.offer(one, MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS)); - threadAssertEquals(0, q.remainingCapacity()); - } - catch (InterruptedException e) { - threadUnexpectedException(); - } - } - }); + executor.execute(new CheckedRunnable() { + public void realRun() throws InterruptedException { + assertFalse(q.offer(one)); + assertTrue(q.offer(one, MEDIUM_DELAY_MS, MILLISECONDS)); + assertEquals(0, q.remainingCapacity()); + }}); + + executor.execute(new CheckedRunnable() { + public void realRun() throws InterruptedException { + delay(SMALL_DELAY_MS); + assertSame(one, q.take()); + }}); - executor.execute(new Runnable() { - public void run() { - try { - Thread.sleep(SMALL_DELAY_MS); - threadAssertEquals(one, q.take()); - } - catch (InterruptedException e) { - threadUnexpectedException(); - } - } - }); - joinPool(executor); - } /** @@ -536,83 +586,65 @@ public class SynchronousQueueTest extend public void testPollInExecutor() { final SynchronousQueue q = new SynchronousQueue(); ExecutorService executor = Executors.newFixedThreadPool(2); - executor.execute(new Runnable() { - public void run() { - threadAssertNull(q.poll()); - try { - threadAssertTrue(null != q.poll(MEDIUM_DELAY_MS, TimeUnit.MILLISECONDS)); - threadAssertTrue(q.isEmpty()); - } - catch (InterruptedException e) { - threadUnexpectedException(); - } - } - }); + executor.execute(new CheckedRunnable() { + public void realRun() throws InterruptedException { + assertNull(q.poll()); + assertSame(one, q.poll(MEDIUM_DELAY_MS, MILLISECONDS)); + assertTrue(q.isEmpty()); + }}); + + executor.execute(new CheckedRunnable() { + public void realRun() throws InterruptedException { + delay(SHORT_DELAY_MS); + q.put(one); + }}); - executor.execute(new Runnable() { - public void run() { - try { - Thread.sleep(SMALL_DELAY_MS); - q.put(new Integer(1)); - } - catch (InterruptedException e) { - threadUnexpectedException(); - } - } - }); - joinPool(executor); } /** * a deserialized serialized queue is usable */ - public void testSerialization() { + public void testSerialization() throws Exception { SynchronousQueue q = new SynchronousQueue(); - try { - ByteArrayOutputStream bout = new ByteArrayOutputStream(10000); - ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout)); - out.writeObject(q); - out.close(); - - ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray()); - ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin)); - SynchronousQueue r = (SynchronousQueue)in.readObject(); - assertEquals(q.size(), r.size()); - while (!q.isEmpty()) - assertEquals(q.remove(), r.remove()); - } catch(Exception e){ - unexpectedException(); - } + ByteArrayOutputStream bout = new ByteArrayOutputStream(10000); + ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(bout)); + out.writeObject(q); + out.close(); + + ByteArrayInputStream bin = new ByteArrayInputStream(bout.toByteArray()); + ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(bin)); + SynchronousQueue r = (SynchronousQueue)in.readObject(); + assertEquals(q.size(), r.size()); + while (!q.isEmpty()) + assertEquals(q.remove(), r.remove()); } /** * drainTo(null) throws NPE - */ + */ public void testDrainToNull() { SynchronousQueue q = new SynchronousQueue(); try { q.drainTo(null); shouldThrow(); - } catch(NullPointerException success) { - } + } catch (NullPointerException success) {} } /** * drainTo(this) throws IAE - */ + */ public void testDrainToSelf() { SynchronousQueue q = new SynchronousQueue(); try { q.drainTo(q); shouldThrow(); - } catch(IllegalArgumentException success) { - } + } catch (IllegalArgumentException success) {} } /** * drainTo(c) of empty queue doesn't transfer elements - */ + */ public void testDrainTo() { SynchronousQueue q = new SynchronousQueue(); ArrayList l = new ArrayList(); @@ -623,98 +655,74 @@ public class SynchronousQueueTest extend /** * drainTo empties queue, unblocking a waiting put. - */ - public void testDrainToWithActivePut() { + */ + public void testDrainToWithActivePut() throws InterruptedException { final SynchronousQueue q = new SynchronousQueue(); - Thread t = new Thread(new Runnable() { - public void run() { - try { - q.put(new Integer(1)); - } catch (InterruptedException ie){ - threadUnexpectedException(); - } - } - }); - try { - t.start(); - ArrayList l = new ArrayList(); - Thread.sleep(SHORT_DELAY_MS); - q.drainTo(l); - assertTrue(l.size() <= 1); - if (l.size() > 0) - assertEquals(l.get(0), new Integer(1)); - t.join(); - assertTrue(l.size() <= 1); - } catch(Exception e){ - unexpectedException(); - } + Thread t = new Thread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + q.put(new Integer(1)); + }}); + + t.start(); + ArrayList l = new ArrayList(); + delay(SHORT_DELAY_MS); + q.drainTo(l); + assertTrue(l.size() <= 1); + if (l.size() > 0) + assertEquals(l.get(0), new Integer(1)); + t.join(); + assertTrue(l.size() <= 1); } /** * drainTo(null, n) throws NPE - */ + */ public void testDrainToNullN() { SynchronousQueue q = new SynchronousQueue(); try { q.drainTo(null, 0); shouldThrow(); - } catch(NullPointerException success) { - } + } catch (NullPointerException success) {} } /** * drainTo(this, n) throws IAE - */ + */ public void testDrainToSelfN() { SynchronousQueue q = new SynchronousQueue(); try { q.drainTo(q, 0); shouldThrow(); - } catch(IllegalArgumentException success) { - } + } catch (IllegalArgumentException success) {} } /** * drainTo(c, n) empties up to n elements of queue into c - */ - public void testDrainToN() { + */ + public void testDrainToN() throws InterruptedException { final SynchronousQueue q = new SynchronousQueue(); - Thread t1 = new Thread(new Runnable() { - public void run() { - try { - q.put(one); - } catch (InterruptedException ie){ - threadUnexpectedException(); - } - } - }); - Thread t2 = new Thread(new Runnable() { - public void run() { - try { - q.put(two); - } catch (InterruptedException ie){ - threadUnexpectedException(); - } - } - }); + Thread t1 = new Thread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + q.put(one); + }}); + + Thread t2 = new Thread(new CheckedRunnable() { + public void realRun() throws InterruptedException { + q.put(two); + }}); - try { - t1.start(); - t2.start(); - ArrayList l = new ArrayList(); - Thread.sleep(SHORT_DELAY_MS); - q.drainTo(l, 1); - assertTrue(l.size() == 1); - q.drainTo(l, 1); - assertTrue(l.size() == 2); - assertTrue(l.contains(one)); - assertTrue(l.contains(two)); - t1.join(); - t2.join(); - } catch(Exception e){ - unexpectedException(); - } + t1.start(); + t2.start(); + ArrayList l = new ArrayList(); + delay(SHORT_DELAY_MS); + q.drainTo(l, 1); + assertEquals(1, l.size()); + q.drainTo(l, 1); + assertEquals(2, l.size()); + assertTrue(l.contains(one)); + assertTrue(l.contains(two)); + t1.join(); + t2.join(); } - }