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.10 by dl, Sat Sep 12 11:25:15 2015 UTC vs.
Revision 1.18 by jsr166, Wed Jan 4 06:09:58 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 38 | Line 28 | public class SubmissionPublisherTest ext
28          return new TestSuite(SubmissionPublisherTest.class);
29      }
30  
31 <    // Factory for single thread pool in case commonPool parallelism is zero
42 <    static final class DaemonThreadFactory implements ThreadFactory {
43 <        public Thread newThread(Runnable r) {
44 <            Thread t = new Thread(r);
45 <            t.setDaemon(true);
46 <            return t;
47 <        }
48 <    }
49 <
50 <    static final Executor basicExecutor =
51 <        (ForkJoinPool.getCommonPoolParallelism() > 1) ?
52 <        ForkJoinPool.commonPool() :
53 <        new ThreadPoolExecutor(1, 1, 60, SECONDS,
54 <                               new LinkedBlockingQueue<Runnable>(),
55 <                               new DaemonThreadFactory());
31 >    final Executor basicExecutor = basicPublisher().getExecutor();
32  
33      static SubmissionPublisher<Integer> basicPublisher() {
34 <        return new SubmissionPublisher<Integer>(basicExecutor,
59 <                                                Flow.defaultBufferSize());
34 >        return new SubmissionPublisher<Integer>();
35      }
36  
37      static class SPException extends RuntimeException {}
# Line 168 | Line 143 | public class SubmissionPublisherTest ext
143      /**
144       * A default-constructed SubmissionPublisher has no subscribers,
145       * is not closed, has default buffer size, and uses the
146 <     * ForkJoinPool.commonPool executor
146 >     * defaultExecutor
147       */
148      public void testConstructor1() {
149 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>();
149 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>();
150          checkInitialState(p);
176 –        assertSame(p.getExecutor(), ForkJoinPool.commonPool());
151          assertEquals(p.getMaxBufferCapacity(), Flow.defaultBufferSize());
152 +        Executor e = p.getExecutor(), c = ForkJoinPool.commonPool();
153 +        if (ForkJoinPool.getCommonPoolParallelism() > 1)
154 +            assertSame(e, c);
155 +        else
156 +            assertNotSame(e, c);
157      }
158  
159      /**
# Line 183 | 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());
# Line 396 | Line 375 | public class SubmissionPublisherTest ext
375  
376      /**
377       * Closing a publisher exceptionally causes onError to subscribers
378 +     * after they are subscribed
379       */
380      public void testCloseExceptionallyError() {
381          SubmissionPublisher<Integer> p = basicPublisher();
# Line 406 | Line 386 | public class SubmissionPublisherTest ext
386          p.submit(1);
387          p.closeExceptionally(new SPException());
388          assertTrue(p.isClosed());
389 +        s1.awaitSubscribe();
390          s1.awaitError();
391          assertTrue(s1.nexts <= 1);
392          assertEquals(1, s1.errors);
393 +        s2.awaitSubscribe();
394          s2.awaitError();
395          assertTrue(s2.nexts <= 1);
396          assertEquals(1, s2.errors);
# Line 462 | Line 444 | public class SubmissionPublisherTest ext
444       */
445      public void testThrowOnNextHandler() {
446          AtomicInteger calls = new AtomicInteger();
447 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
448 <            (basicExecutor, 8,
467 <             (s, e) -> calls.getAndIncrement());
447 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
448 >            basicExecutor, 8, (s, e) -> calls.getAndIncrement());
449          TestSubscriber s1 = new TestSubscriber();
450          TestSubscriber s2 = new TestSubscriber();
451          p.subscribe(s1);
# Line 645 | Line 626 | public class SubmissionPublisherTest ext
626       * submit eventually issues requested items when buffer capacity is 1
627       */
628      public void testCap1Submit() {
629 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
630 <            basicExecutor, 1);
629 >        SubmissionPublisher<Integer> p
630 >            = new SubmissionPublisher<>(basicExecutor, 1);
631          TestSubscriber s1 = new TestSubscriber();
632          TestSubscriber s2 = new TestSubscriber();
633          p.subscribe(s1);
# Line 724 | Line 705 | public class SubmissionPublisherTest ext
705       * offer reports drops if saturated
706       */
707      public void testDroppedOffer() {
708 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
709 <            basicExecutor, 4);
708 >        SubmissionPublisher<Integer> p
709 >            = new SubmissionPublisher<>(basicExecutor, 4);
710          TestSubscriber s1 = new TestSubscriber();
711          s1.request = false;
712          TestSubscriber s2 = new TestSubscriber();
# Line 753 | Line 734 | public class SubmissionPublisherTest ext
734       */
735      public void testHandledDroppedOffer() {
736          AtomicInteger calls = new AtomicInteger();
737 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
738 <            basicExecutor, 4);
737 >        SubmissionPublisher<Integer> p
738 >            = new SubmissionPublisher<>(basicExecutor, 4);
739          TestSubscriber s1 = new TestSubscriber();
740          s1.request = false;
741          TestSubscriber s2 = new TestSubscriber();
# Line 781 | Line 762 | public class SubmissionPublisherTest ext
762       */
763      public void testRecoveredHandledDroppedOffer() {
764          AtomicInteger calls = new AtomicInteger();
765 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
766 <            basicExecutor, 4);
765 >        SubmissionPublisher<Integer> p
766 >            = new SubmissionPublisher<>(basicExecutor, 4);
767          TestSubscriber s1 = new TestSubscriber();
768          s1.request = false;
769          TestSubscriber s2 = new TestSubscriber();
# Line 862 | Line 843 | public class SubmissionPublisherTest ext
843       * Timed offer reports drops if saturated
844       */
845      public void testDroppedTimedOffer() {
846 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
847 <            basicExecutor, 4);
846 >        SubmissionPublisher<Integer> p
847 >            = new SubmissionPublisher<>(basicExecutor, 4);
848          TestSubscriber s1 = new TestSubscriber();
849          s1.request = false;
850          TestSubscriber s2 = new TestSubscriber();
# Line 894 | Line 875 | public class SubmissionPublisherTest ext
875       */
876      public void testHandledDroppedTimedOffer() {
877          AtomicInteger calls = new AtomicInteger();
878 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
879 <            basicExecutor, 4);
878 >        SubmissionPublisher<Integer> p
879 >            = new SubmissionPublisher<>(basicExecutor, 4);
880          TestSubscriber s1 = new TestSubscriber();
881          s1.request = false;
882          TestSubscriber s2 = new TestSubscriber();
# Line 924 | Line 905 | public class SubmissionPublisherTest ext
905       */
906      public void testRecoveredHandledDroppedTimedOffer() {
907          AtomicInteger calls = new AtomicInteger();
908 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
909 <            basicExecutor, 4);
908 >        SubmissionPublisher<Integer> p
909 >            = new SubmissionPublisher<>(basicExecutor, 4);
910          TestSubscriber s1 = new TestSubscriber();
911          s1.request = false;
912          TestSubscriber s2 = new TestSubscriber();
# Line 956 | Line 937 | public class SubmissionPublisherTest ext
937      public void testConsume() {
938          AtomicInteger sum = new AtomicInteger();
939          SubmissionPublisher<Integer> p = basicPublisher();
940 <        CompletableFuture<Void> f =
941 <            p.consume((Integer x) -> { sum.getAndAdd(x.intValue()); });
940 >        CompletableFuture<Void> f =
941 >            p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
942          int n = 20;
943          for (int i = 1; i <= n; ++i)
944              p.submit(i);
# Line 974 | Line 955 | public class SubmissionPublisherTest ext
955          try {
956              CompletableFuture<Void> f = p.consume(null);
957              shouldThrow();
958 <        } catch(NullPointerException success) {
978 <        }
958 >        } catch (NullPointerException success) {}
959      }
960  
961      /**
# Line 991 | Line 971 | public class SubmissionPublisherTest ext
971              p.submit(i);
972          assertTrue(count.get() < n);
973      }
974 <    
974 >
975   }

Diff Legend

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