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.8 by jsr166, Mon Sep 7 21:36:03 2015 UTC vs.
Revision 1.23 by dl, Sun Nov 26 21:37:56 2017 UTC

# Line 5 | Line 5
5   * http://creativecommons.org/publicdomain/zero/1.0/
6   */
7  
8 < import static java.util.concurrent.TimeUnit.MILLISECONDS;
9 < import static java.util.concurrent.TimeUnit.SECONDS;
10 <
8 > import java.util.concurrent.CompletableFuture;
9   import java.util.concurrent.Executor;
10   import java.util.concurrent.Executors;
11   import java.util.concurrent.Flow;
14 – import static java.util.concurrent.Flow.Publisher;
15 – import static java.util.concurrent.Flow.Subscriber;
16 – import static java.util.concurrent.Flow.Subscription;
17 – import java.util.concurrent.LinkedBlockingQueue;
12   import java.util.concurrent.ForkJoinPool;
13   import java.util.concurrent.SubmissionPublisher;
20 – import java.util.concurrent.ThreadFactory;
21 – import java.util.concurrent.ThreadPoolExecutor;
22 – import java.util.concurrent.TimeUnit;
14   import java.util.concurrent.atomic.AtomicInteger;
24 – import java.util.function.BiConsumer;
25 – import java.util.function.BiPredicate;
26 – import java.util.function.BiFunction;
27 –
15   import junit.framework.Test;
16   import junit.framework.TestSuite;
17  
18 + import static java.util.concurrent.Flow.Subscriber;
19 + import static java.util.concurrent.Flow.Subscription;
20 + import static java.util.concurrent.TimeUnit.MILLISECONDS;
21 +
22   public class SubmissionPublisherTest extends JSR166TestCase {
23  
24      public static void main(String[] args) {
# Line 37 | 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
41 <    static final class DaemonThreadFactory implements ThreadFactory {
42 <        public Thread newThread(Runnable r) {
43 <            Thread t = new Thread(r);
44 <            t.setDaemon(true);
45 <            return t;
46 <        }
47 <    }
48 <
49 <    static final Executor basicExecutor =
50 <        (ForkJoinPool.getCommonPoolParallelism() > 0) ?
51 <        ForkJoinPool.commonPool() :
52 <        new ThreadPoolExecutor(1, 1, 60, SECONDS,
53 <                               new LinkedBlockingQueue<Runnable>(),
54 <                               new DaemonThreadFactory());
31 >    final Executor basicExecutor = basicPublisher().getExecutor();
32  
33      static SubmissionPublisher<Integer> basicPublisher() {
34 <        return new SubmissionPublisher<Integer>(basicExecutor,
58 <                                                Flow.defaultBufferSize());
34 >        return new SubmissionPublisher<Integer>();
35      }
36  
37      static class SPException extends RuntimeException {}
# Line 167 | 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);
175 –        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 182 | 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 200 | 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 212 | 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 233 | 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 285 | Line 267 | public class SubmissionPublisherTest ext
267          assertEquals(0, s2.nexts);
268          assertEquals(0, s2.errors);
269          assertEquals(0, s2.completes);
270 +        p.close();
271      }
272  
273      /**
# Line 394 | 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 404 | 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 416 | Line 402 | public class SubmissionPublisherTest ext
402       * Cancelling a subscription eventually causes no more onNexts to be issued
403       */
404      public void testCancel() {
405 <        SubmissionPublisher<Integer> p = basicPublisher();
405 >        SubmissionPublisher<Integer> p =
406 >            new SubmissionPublisher<Integer>(basicExecutor, 4); // must be < 20
407          TestSubscriber s1 = new TestSubscriber();
408          TestSubscriber s2 = new TestSubscriber();
409          p.subscribe(s1);
# Line 460 | Line 447 | public class SubmissionPublisherTest ext
447       */
448      public void testThrowOnNextHandler() {
449          AtomicInteger calls = new AtomicInteger();
450 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
451 <            (basicExecutor, 8,
465 <             (s, e) -> calls.getAndIncrement());
450 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
451 >            basicExecutor, 8, (s, e) -> calls.getAndIncrement());
452          TestSubscriber s1 = new TestSubscriber();
453          TestSubscriber s2 = new TestSubscriber();
454          p.subscribe(s1);
# Line 509 | Line 495 | public class SubmissionPublisherTest ext
495          s1.request = false;
496          p.subscribe(s1);
497          s1.awaitSubscribe();
498 <        assertTrue(p.estimateMinimumDemand() == 0);
498 >        assertEquals(0, p.estimateMinimumDemand());
499          TestSubscriber s2 = new TestSubscriber();
500          p.subscribe(s2);
501          p.submit(1);
# Line 550 | Line 536 | public class SubmissionPublisherTest ext
536      }
537  
538      /**
539 <     * Negative request causes error
539 >     * Non-positive request causes error
540       */
541      public void testRequest3() {
542          SubmissionPublisher<Integer> p = basicPublisher();
543          TestSubscriber s1 = new TestSubscriber();
544          TestSubscriber s2 = new TestSubscriber();
545 +        TestSubscriber s3 = new TestSubscriber();
546          p.subscribe(s1);
547          p.subscribe(s2);
548 +        p.subscribe(s3);
549 +        s3.awaitSubscribe();
550          s2.awaitSubscribe();
551          s1.awaitSubscribe();
552          s1.sn.request(-1L);
553 +        s3.sn.request(0L);
554          p.submit(1);
555          p.submit(2);
556          p.close();
# Line 570 | Line 560 | public class SubmissionPublisherTest ext
560          s1.awaitError();
561          assertEquals(1, s1.errors);
562          assertTrue(s1.lastError instanceof IllegalArgumentException);
563 +        s3.awaitError();
564 +        assertEquals(1, s3.errors);
565 +        assertTrue(s3.lastError instanceof IllegalArgumentException);
566      }
567  
568      /**
569       * estimateMinimumDemand reports 0 until request, nonzero after
570 <     * request, and zero again after delivery
570 >     * request
571       */
572      public void testEstimateMinimumDemand() {
573          TestSubscriber s = new TestSubscriber();
# Line 585 | Line 578 | public class SubmissionPublisherTest ext
578          assertEquals(0, p.estimateMinimumDemand());
579          s.sn.request(1);
580          assertEquals(1, p.estimateMinimumDemand());
588 –        p.submit(1);
589 –        s.awaitNext(1);
590 –        assertEquals(0, p.estimateMinimumDemand());
581      }
582  
583      /**
# Line 643 | Line 633 | public class SubmissionPublisherTest ext
633       * submit eventually issues requested items when buffer capacity is 1
634       */
635      public void testCap1Submit() {
636 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
637 <            basicExecutor, 1);
636 >        SubmissionPublisher<Integer> p
637 >            = new SubmissionPublisher<>(basicExecutor, 1);
638          TestSubscriber s1 = new TestSubscriber();
639          TestSubscriber s2 = new TestSubscriber();
640          p.subscribe(s1);
641          p.subscribe(s2);
642          for (int i = 1; i <= 20; ++i) {
653 –            assertTrue(p.estimateMinimumDemand() <= 1);
643              assertTrue(p.submit(i) >= 0);
644          }
645          p.close();
# Line 722 | Line 711 | public class SubmissionPublisherTest ext
711       * offer reports drops if saturated
712       */
713      public void testDroppedOffer() {
714 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
715 <            basicExecutor, 4);
714 >        SubmissionPublisher<Integer> p
715 >            = new SubmissionPublisher<>(basicExecutor, 4);
716          TestSubscriber s1 = new TestSubscriber();
717          s1.request = false;
718          TestSubscriber s2 = new TestSubscriber();
# Line 751 | Line 740 | public class SubmissionPublisherTest ext
740       */
741      public void testHandledDroppedOffer() {
742          AtomicInteger calls = new AtomicInteger();
743 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
744 <            basicExecutor, 4);
743 >        SubmissionPublisher<Integer> p
744 >            = new SubmissionPublisher<>(basicExecutor, 4);
745          TestSubscriber s1 = new TestSubscriber();
746          s1.request = false;
747          TestSubscriber s2 = new TestSubscriber();
# Line 779 | Line 768 | public class SubmissionPublisherTest ext
768       */
769      public void testRecoveredHandledDroppedOffer() {
770          AtomicInteger calls = new AtomicInteger();
771 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
772 <            basicExecutor, 4);
771 >        SubmissionPublisher<Integer> p
772 >            = new SubmissionPublisher<>(basicExecutor, 4);
773          TestSubscriber s1 = new TestSubscriber();
774          s1.request = false;
775          TestSubscriber s2 = new TestSubscriber();
# Line 860 | Line 849 | public class SubmissionPublisherTest ext
849       * Timed offer reports drops if saturated
850       */
851      public void testDroppedTimedOffer() {
852 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
853 <            basicExecutor, 4);
852 >        SubmissionPublisher<Integer> p
853 >            = new SubmissionPublisher<>(basicExecutor, 4);
854          TestSubscriber s1 = new TestSubscriber();
855          s1.request = false;
856          TestSubscriber s2 = new TestSubscriber();
# Line 870 | Line 859 | public class SubmissionPublisherTest ext
859          p.subscribe(s2);
860          s2.awaitSubscribe();
861          s1.awaitSubscribe();
862 +        long delay = timeoutMillis();
863          for (int i = 1; i <= 4; ++i)
864 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
865 <        p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
866 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
864 >            assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
865 >        long startTime = System.nanoTime();
866 >        assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
867          s1.sn.request(64);
868 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
868 >        assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
869 >        // 2 * delay should elapse but check only 1 * delay to allow timer slop
870 >        assertTrue(millisElapsedSince(startTime) >= delay);
871          s2.sn.request(64);
872          p.close();
873          s2.awaitComplete();
# Line 889 | Line 881 | public class SubmissionPublisherTest ext
881       */
882      public void testHandledDroppedTimedOffer() {
883          AtomicInteger calls = new AtomicInteger();
884 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
885 <            basicExecutor, 4);
884 >        SubmissionPublisher<Integer> p
885 >            = new SubmissionPublisher<>(basicExecutor, 4);
886          TestSubscriber s1 = new TestSubscriber();
887          s1.request = false;
888          TestSubscriber s2 = new TestSubscriber();
# Line 899 | Line 891 | public class SubmissionPublisherTest ext
891          p.subscribe(s2);
892          s2.awaitSubscribe();
893          s1.awaitSubscribe();
894 +        long delay = timeoutMillis();
895          for (int i = 1; i <= 4; ++i)
896 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
897 <        p.offer(5, (s, x) -> noopHandle(calls));
898 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
896 >            assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
897 >        long startTime = System.nanoTime();
898 >        assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
899          s1.sn.request(64);
900 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
900 >        assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
901 >        assertTrue(millisElapsedSince(startTime) >= delay);
902          s2.sn.request(64);
903          p.close();
904          s2.awaitComplete();
# Line 917 | Line 911 | public class SubmissionPublisherTest ext
911       */
912      public void testRecoveredHandledDroppedTimedOffer() {
913          AtomicInteger calls = new AtomicInteger();
914 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
915 <            basicExecutor, 4);
914 >        SubmissionPublisher<Integer> p
915 >            = new SubmissionPublisher<>(basicExecutor, 4);
916          TestSubscriber s1 = new TestSubscriber();
917          s1.request = false;
918          TestSubscriber s2 = new TestSubscriber();
# Line 928 | Line 922 | public class SubmissionPublisherTest ext
922          s2.awaitSubscribe();
923          s1.awaitSubscribe();
924          int n = 0;
925 <        for (int i = 1; i <= 8; ++i) {
926 <            int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
925 >        long delay = timeoutMillis();
926 >        long startTime = System.nanoTime();
927 >        for (int i = 1; i <= 6; ++i) {
928 >            int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
929              n = n + 2 + (d < 0 ? d : 0);
930          }
931 +        assertTrue(millisElapsedSince(startTime) >= delay);
932          p.close();
933          s2.awaitComplete();
934          s1.awaitComplete();
# Line 939 | Line 936 | public class SubmissionPublisherTest ext
936          assertTrue(calls.get() >= 2);
937      }
938  
939 +    /**
940 +     * consume returns a CompletableFuture that is done when
941 +     * publisher completes
942 +     */
943 +    public void testConsume() {
944 +        AtomicInteger sum = new AtomicInteger();
945 +        SubmissionPublisher<Integer> p = basicPublisher();
946 +        CompletableFuture<Void> f =
947 +            p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
948 +        int n = 20;
949 +        for (int i = 1; i <= n; ++i)
950 +            p.submit(i);
951 +        p.close();
952 +        f.join();
953 +        assertEquals((n * (n + 1)) / 2, sum.get());
954 +    }
955 +
956 +    /**
957 +     * consume(null) throws NPE
958 +     */
959 +    public void testConsumeNPE() {
960 +        SubmissionPublisher<Integer> p = basicPublisher();
961 +        try {
962 +            CompletableFuture<Void> f = p.consume(null);
963 +            shouldThrow();
964 +        } catch (NullPointerException success) {}
965 +    }
966 +
967 +    /**
968 +     * consume eventually stops processing published items if cancelled
969 +     */
970 +    public void testCancelledConsume() {
971 +        AtomicInteger count = new AtomicInteger();
972 +        SubmissionPublisher<Integer> p = basicPublisher();
973 +        CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
974 +        f.cancel(true);
975 +        int n = 1000000; // arbitrary limit
976 +        for (int i = 1; i <= n; ++i)
977 +            p.submit(i);
978 +        assertTrue(count.get() < n);
979 +    }
980 +
981   }

Diff Legend

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