--- jsr166/src/test/tck/SubmissionPublisherTest.java 2015/09/12 17:18:13 1.12 +++ jsr166/src/test/tck/SubmissionPublisherTest.java 2017/03/09 00:11:16 1.19 @@ -10,24 +10,14 @@ import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.Flow; import java.util.concurrent.ForkJoinPool; -import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.SubmissionPublisher; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.BiConsumer; -import java.util.function.BiFunction; -import java.util.function.BiPredicate; -import java.util.stream.Stream; import junit.framework.Test; import junit.framework.TestSuite; -import static java.util.concurrent.Flow.Publisher; import static java.util.concurrent.Flow.Subscriber; import static java.util.concurrent.Flow.Subscription; import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static java.util.concurrent.TimeUnit.SECONDS; public class SubmissionPublisherTest extends JSR166TestCase { @@ -39,7 +29,7 @@ public class SubmissionPublisherTest ext } final Executor basicExecutor = basicPublisher().getExecutor(); - + static SubmissionPublisher basicPublisher() { return new SubmissionPublisher(); } @@ -156,7 +146,7 @@ public class SubmissionPublisherTest ext * defaultExecutor */ public void testConstructor1() { - SubmissionPublisher p = new SubmissionPublisher(); + SubmissionPublisher p = new SubmissionPublisher<>(); checkInitialState(p); assertEquals(p.getMaxBufferCapacity(), Flow.defaultBufferSize()); Executor e = p.getExecutor(), c = ForkJoinPool.commonPool(); @@ -172,7 +162,7 @@ public class SubmissionPublisherTest ext */ public void testConstructor2() { Executor e = Executors.newFixedThreadPool(1); - SubmissionPublisher p = new SubmissionPublisher(e, 8); + SubmissionPublisher p = new SubmissionPublisher<>(e, 8); checkInitialState(p); assertSame(p.getExecutor(), e); assertEquals(8, p.getMaxBufferCapacity()); @@ -385,6 +375,7 @@ public class SubmissionPublisherTest ext /** * Closing a publisher exceptionally causes onError to subscribers + * after they are subscribed */ public void testCloseExceptionallyError() { SubmissionPublisher p = basicPublisher(); @@ -395,9 +386,11 @@ public class SubmissionPublisherTest ext p.submit(1); p.closeExceptionally(new SPException()); assertTrue(p.isClosed()); + s1.awaitSubscribe(); s1.awaitError(); assertTrue(s1.nexts <= 1); assertEquals(1, s1.errors); + s2.awaitSubscribe(); s2.awaitError(); assertTrue(s2.nexts <= 1); assertEquals(1, s2.errors); @@ -451,9 +444,8 @@ public class SubmissionPublisherTest ext */ public void testThrowOnNextHandler() { AtomicInteger calls = new AtomicInteger(); - SubmissionPublisher p = new SubmissionPublisher - (basicExecutor, 8, - (s, e) -> calls.getAndIncrement()); + SubmissionPublisher p = new SubmissionPublisher<>( + basicExecutor, 8, (s, e) -> calls.getAndIncrement()); TestSubscriber s1 = new TestSubscriber(); TestSubscriber s2 = new TestSubscriber(); p.subscribe(s1); @@ -541,17 +533,21 @@ public class SubmissionPublisherTest ext } /** - * Negative request causes error + * Non-positive request causes error */ public void testRequest3() { SubmissionPublisher p = basicPublisher(); TestSubscriber s1 = new TestSubscriber(); TestSubscriber s2 = new TestSubscriber(); + TestSubscriber s3 = new TestSubscriber(); p.subscribe(s1); p.subscribe(s2); + p.subscribe(s3); + s3.awaitSubscribe(); s2.awaitSubscribe(); s1.awaitSubscribe(); s1.sn.request(-1L); + s3.sn.request(0L); p.submit(1); p.submit(2); p.close(); @@ -561,6 +557,9 @@ public class SubmissionPublisherTest ext s1.awaitError(); assertEquals(1, s1.errors); assertTrue(s1.lastError instanceof IllegalArgumentException); + s3.awaitError(); + assertEquals(1, s3.errors); + assertTrue(s3.lastError instanceof IllegalArgumentException); } /** @@ -634,8 +633,8 @@ public class SubmissionPublisherTest ext * submit eventually issues requested items when buffer capacity is 1 */ public void testCap1Submit() { - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 1); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 1); TestSubscriber s1 = new TestSubscriber(); TestSubscriber s2 = new TestSubscriber(); p.subscribe(s1); @@ -713,8 +712,8 @@ public class SubmissionPublisherTest ext * offer reports drops if saturated */ public void testDroppedOffer() { - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 4); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 4); TestSubscriber s1 = new TestSubscriber(); s1.request = false; TestSubscriber s2 = new TestSubscriber(); @@ -742,8 +741,8 @@ public class SubmissionPublisherTest ext */ public void testHandledDroppedOffer() { AtomicInteger calls = new AtomicInteger(); - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 4); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 4); TestSubscriber s1 = new TestSubscriber(); s1.request = false; TestSubscriber s2 = new TestSubscriber(); @@ -770,8 +769,8 @@ public class SubmissionPublisherTest ext */ public void testRecoveredHandledDroppedOffer() { AtomicInteger calls = new AtomicInteger(); - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 4); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 4); TestSubscriber s1 = new TestSubscriber(); s1.request = false; TestSubscriber s2 = new TestSubscriber(); @@ -851,8 +850,8 @@ public class SubmissionPublisherTest ext * Timed offer reports drops if saturated */ public void testDroppedTimedOffer() { - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 4); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 4); TestSubscriber s1 = new TestSubscriber(); s1.request = false; TestSubscriber s2 = new TestSubscriber(); @@ -883,8 +882,8 @@ public class SubmissionPublisherTest ext */ public void testHandledDroppedTimedOffer() { AtomicInteger calls = new AtomicInteger(); - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 4); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 4); TestSubscriber s1 = new TestSubscriber(); s1.request = false; TestSubscriber s2 = new TestSubscriber(); @@ -913,8 +912,8 @@ public class SubmissionPublisherTest ext */ public void testRecoveredHandledDroppedTimedOffer() { AtomicInteger calls = new AtomicInteger(); - SubmissionPublisher p = new SubmissionPublisher( - basicExecutor, 4); + SubmissionPublisher p + = new SubmissionPublisher<>(basicExecutor, 4); TestSubscriber s1 = new TestSubscriber(); s1.request = false; TestSubscriber s2 = new TestSubscriber(); @@ -946,7 +945,7 @@ public class SubmissionPublisherTest ext AtomicInteger sum = new AtomicInteger(); SubmissionPublisher p = basicPublisher(); CompletableFuture f = - p.consume((Integer x) -> { sum.getAndAdd(x.intValue()); }); + p.consume((Integer x) -> sum.getAndAdd(x.intValue())); int n = 20; for (int i = 1; i <= n; ++i) p.submit(i);