ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/test/tck/SubmissionPublisherTest.java
(Generate patch)

Comparing jsr166/src/test/tck/SubmissionPublisherTest.java (file contents):
Revision 1.13 by jsr166, Sat Sep 12 18:19:57 2015 UTC vs.
Revision 1.21 by jsr166, Mon May 29 22:44:27 2017 UTC

# Line 10 | Line 10 | import java.util.concurrent.Executor;
10   import java.util.concurrent.Executors;
11   import java.util.concurrent.Flow;
12   import java.util.concurrent.ForkJoinPool;
13 – import java.util.concurrent.LinkedBlockingQueue;
13   import java.util.concurrent.SubmissionPublisher;
15 – import java.util.concurrent.ThreadFactory;
16 – import java.util.concurrent.ThreadPoolExecutor;
17 – import java.util.concurrent.TimeUnit;
14   import java.util.concurrent.atomic.AtomicInteger;
19 – import java.util.function.BiConsumer;
20 – import java.util.function.BiFunction;
21 – import java.util.function.BiPredicate;
22 – import java.util.stream.Stream;
15   import junit.framework.Test;
16   import junit.framework.TestSuite;
17  
26 – import static java.util.concurrent.Flow.Publisher;
18   import static java.util.concurrent.Flow.Subscriber;
19   import static java.util.concurrent.Flow.Subscription;
20   import static java.util.concurrent.TimeUnit.MILLISECONDS;
30 – import static java.util.concurrent.TimeUnit.SECONDS;
21  
22   public class SubmissionPublisherTest extends JSR166TestCase {
23  
# Line 156 | Line 146 | public class SubmissionPublisherTest ext
146       * defaultExecutor
147       */
148      public void testConstructor1() {
149 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>();
149 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>();
150          checkInitialState(p);
151          assertEquals(p.getMaxBufferCapacity(), Flow.defaultBufferSize());
152          Executor e = p.getExecutor(), c = ForkJoinPool.commonPool();
# Line 172 | Line 162 | public class SubmissionPublisherTest ext
162       */
163      public void testConstructor2() {
164          Executor e = Executors.newFixedThreadPool(1);
165 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, 8);
165 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(e, 8);
166          checkInitialState(p);
167          assertSame(p.getExecutor(), e);
168          assertEquals(8, p.getMaxBufferCapacity());
169      }
170  
171      /**
172 <     * A null Executor argument to SubmissionPublisher constructor throws NPE
172 >     * A null Executor argument to SubmissionPublisher constructor
173 >     * throws NullPointerException
174       */
175      public void testConstructor3() {
176          try {
# Line 190 | Line 181 | public class SubmissionPublisherTest ext
181  
182      /**
183       * A negative capacity argument to SubmissionPublisher constructor
184 <     * throws IAE
184 >     * throws IllegalArgumentException
185       */
186      public void testConstructor4() {
187          Executor e = Executors.newFixedThreadPool(1);
# Line 202 | Line 193 | public class SubmissionPublisherTest ext
193  
194      /**
195       * A closed publisher reports isClosed with no closedException and
196 <     * throws ISE upon attempted submission; a subsequent close or
197 <     * closeExceptionally has no additional effect.
196 >     * throws IllegalStateException upon attempted submission; a
197 >     * subsequent close or closeExceptionally has no additional
198 >     * effect.
199       */
200      public void testClose() {
201          SubmissionPublisher<Integer> p = basicPublisher();
# Line 223 | Line 215 | public class SubmissionPublisherTest ext
215  
216      /**
217       * A publisher closedExceptionally reports isClosed with the
218 <     * closedException and throws ISE upon attempted submission; a
219 <     * subsequent close or closeExceptionally has no additional
220 <     * effect.
218 >     * closedException and throws IllegalStateException upon attempted
219 >     * submission; a subsequent close or closeExceptionally has no
220 >     * additional effect.
221       */
222      public void testCloseExceptionally() {
223          SubmissionPublisher<Integer> p = basicPublisher();
# Line 385 | Line 377 | public class SubmissionPublisherTest ext
377  
378      /**
379       * Closing a publisher exceptionally causes onError to subscribers
380 +     * after they are subscribed
381       */
382      public void testCloseExceptionallyError() {
383          SubmissionPublisher<Integer> p = basicPublisher();
# Line 395 | Line 388 | public class SubmissionPublisherTest ext
388          p.submit(1);
389          p.closeExceptionally(new SPException());
390          assertTrue(p.isClosed());
391 +        s1.awaitSubscribe();
392          s1.awaitError();
393          assertTrue(s1.nexts <= 1);
394          assertEquals(1, s1.errors);
395 +        s2.awaitSubscribe();
396          s2.awaitError();
397          assertTrue(s2.nexts <= 1);
398          assertEquals(1, s2.errors);
# Line 451 | Line 446 | public class SubmissionPublisherTest ext
446       */
447      public void testThrowOnNextHandler() {
448          AtomicInteger calls = new AtomicInteger();
449 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
450 <            (basicExecutor, 8,
456 <             (s, e) -> calls.getAndIncrement());
449 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
450 >            basicExecutor, 8, (s, e) -> calls.getAndIncrement());
451          TestSubscriber s1 = new TestSubscriber();
452          TestSubscriber s2 = new TestSubscriber();
453          p.subscribe(s1);
# Line 500 | Line 494 | public class SubmissionPublisherTest ext
494          s1.request = false;
495          p.subscribe(s1);
496          s1.awaitSubscribe();
497 <        assertTrue(p.estimateMinimumDemand() == 0);
497 >        assertEquals(0, p.estimateMinimumDemand());
498          TestSubscriber s2 = new TestSubscriber();
499          p.subscribe(s2);
500          p.submit(1);
# Line 541 | Line 535 | public class SubmissionPublisherTest ext
535      }
536  
537      /**
538 <     * Negative request causes error
538 >     * Non-positive request causes error
539       */
540      public void testRequest3() {
541          SubmissionPublisher<Integer> p = basicPublisher();
542          TestSubscriber s1 = new TestSubscriber();
543          TestSubscriber s2 = new TestSubscriber();
544 +        TestSubscriber s3 = new TestSubscriber();
545          p.subscribe(s1);
546          p.subscribe(s2);
547 +        p.subscribe(s3);
548 +        s3.awaitSubscribe();
549          s2.awaitSubscribe();
550          s1.awaitSubscribe();
551          s1.sn.request(-1L);
552 +        s3.sn.request(0L);
553          p.submit(1);
554          p.submit(2);
555          p.close();
# Line 561 | Line 559 | public class SubmissionPublisherTest ext
559          s1.awaitError();
560          assertEquals(1, s1.errors);
561          assertTrue(s1.lastError instanceof IllegalArgumentException);
562 +        s3.awaitError();
563 +        assertEquals(1, s3.errors);
564 +        assertTrue(s3.lastError instanceof IllegalArgumentException);
565      }
566  
567      /**
# Line 634 | Line 635 | public class SubmissionPublisherTest ext
635       * submit eventually issues requested items when buffer capacity is 1
636       */
637      public void testCap1Submit() {
638 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
639 <            basicExecutor, 1);
638 >        SubmissionPublisher<Integer> p
639 >            = new SubmissionPublisher<>(basicExecutor, 1);
640          TestSubscriber s1 = new TestSubscriber();
641          TestSubscriber s2 = new TestSubscriber();
642          p.subscribe(s1);
# Line 713 | Line 714 | public class SubmissionPublisherTest ext
714       * offer reports drops if saturated
715       */
716      public void testDroppedOffer() {
717 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
718 <            basicExecutor, 4);
717 >        SubmissionPublisher<Integer> p
718 >            = new SubmissionPublisher<>(basicExecutor, 4);
719          TestSubscriber s1 = new TestSubscriber();
720          s1.request = false;
721          TestSubscriber s2 = new TestSubscriber();
# Line 742 | Line 743 | public class SubmissionPublisherTest ext
743       */
744      public void testHandledDroppedOffer() {
745          AtomicInteger calls = new AtomicInteger();
746 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
747 <            basicExecutor, 4);
746 >        SubmissionPublisher<Integer> p
747 >            = new SubmissionPublisher<>(basicExecutor, 4);
748          TestSubscriber s1 = new TestSubscriber();
749          s1.request = false;
750          TestSubscriber s2 = new TestSubscriber();
# Line 770 | Line 771 | public class SubmissionPublisherTest ext
771       */
772      public void testRecoveredHandledDroppedOffer() {
773          AtomicInteger calls = new AtomicInteger();
774 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
775 <            basicExecutor, 4);
774 >        SubmissionPublisher<Integer> p
775 >            = new SubmissionPublisher<>(basicExecutor, 4);
776          TestSubscriber s1 = new TestSubscriber();
777          s1.request = false;
778          TestSubscriber s2 = new TestSubscriber();
# Line 851 | Line 852 | public class SubmissionPublisherTest ext
852       * Timed offer reports drops if saturated
853       */
854      public void testDroppedTimedOffer() {
855 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
856 <            basicExecutor, 4);
855 >        SubmissionPublisher<Integer> p
856 >            = new SubmissionPublisher<>(basicExecutor, 4);
857          TestSubscriber s1 = new TestSubscriber();
858          s1.request = false;
859          TestSubscriber s2 = new TestSubscriber();
# Line 883 | Line 884 | public class SubmissionPublisherTest ext
884       */
885      public void testHandledDroppedTimedOffer() {
886          AtomicInteger calls = new AtomicInteger();
887 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
888 <            basicExecutor, 4);
887 >        SubmissionPublisher<Integer> p
888 >            = new SubmissionPublisher<>(basicExecutor, 4);
889          TestSubscriber s1 = new TestSubscriber();
890          s1.request = false;
891          TestSubscriber s2 = new TestSubscriber();
# Line 913 | Line 914 | public class SubmissionPublisherTest ext
914       */
915      public void testRecoveredHandledDroppedTimedOffer() {
916          AtomicInteger calls = new AtomicInteger();
917 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
918 <            basicExecutor, 4);
917 >        SubmissionPublisher<Integer> p
918 >            = new SubmissionPublisher<>(basicExecutor, 4);
919          TestSubscriber s1 = new TestSubscriber();
920          s1.request = false;
921          TestSubscriber s2 = new TestSubscriber();
# Line 946 | Line 947 | public class SubmissionPublisherTest ext
947          AtomicInteger sum = new AtomicInteger();
948          SubmissionPublisher<Integer> p = basicPublisher();
949          CompletableFuture<Void> f =
950 <            p.consume((Integer x) -> { sum.getAndAdd(x.intValue()); });
950 >            p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
951          int n = 20;
952          for (int i = 1; i <= n; ++i)
953              p.submit(i);

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines