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.6 by jsr166, Mon Sep 7 20:53:10 2015 UTC vs.
Revision 1.20 by jsr166, Sat Mar 18 20:42:20 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());
# Line 285 | Line 265 | public class SubmissionPublisherTest ext
265          assertEquals(0, s2.nexts);
266          assertEquals(0, s2.errors);
267          assertEquals(0, s2.completes);
268 +        p.close();
269      }
270  
271      /**
# Line 360 | Line 341 | public class SubmissionPublisherTest ext
341      }
342  
343      /**
344 <     * subscribe(null) thows NPE
344 >     * subscribe(null) throws NPE
345       */
346      public void testSubscribe6() {
347          SubmissionPublisher<Integer> p = basicPublisher();
# Line 394 | 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 404 | 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 455 | Line 439 | public class SubmissionPublisherTest ext
439      }
440  
441      /**
442 <     * If a handler is supplied in conctructor, it is invoked when
442 >     * If a handler is supplied in constructor, it is invoked when
443       * subscriber throws an exception in onNext
444       */
445      public void testThrowOnNextHandler() {
446          AtomicInteger calls = new AtomicInteger();
447 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
448 <            (basicExecutor, 8,
465 <             (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 509 | Line 492 | public class SubmissionPublisherTest ext
492          s1.request = false;
493          p.subscribe(s1);
494          s1.awaitSubscribe();
495 <        assertTrue(p.estimateMinimumDemand() == 0);
495 >        assertEquals(0, p.estimateMinimumDemand());
496          TestSubscriber s2 = new TestSubscriber();
497          p.subscribe(s2);
498          p.submit(1);
# Line 550 | Line 533 | public class SubmissionPublisherTest ext
533      }
534  
535      /**
536 <     * Negative request causes error
536 >     * Non-positive request causes error
537       */
538      public void testRequest3() {
539          SubmissionPublisher<Integer> p = basicPublisher();
540          TestSubscriber s1 = new TestSubscriber();
541          TestSubscriber s2 = new TestSubscriber();
542 +        TestSubscriber s3 = new TestSubscriber();
543          p.subscribe(s1);
544          p.subscribe(s2);
545 +        p.subscribe(s3);
546 +        s3.awaitSubscribe();
547          s2.awaitSubscribe();
548          s1.awaitSubscribe();
549          s1.sn.request(-1L);
550 +        s3.sn.request(0L);
551          p.submit(1);
552          p.submit(2);
553          p.close();
# Line 570 | Line 557 | public class SubmissionPublisherTest ext
557          s1.awaitError();
558          assertEquals(1, s1.errors);
559          assertTrue(s1.lastError instanceof IllegalArgumentException);
560 +        s3.awaitError();
561 +        assertEquals(1, s3.errors);
562 +        assertTrue(s3.lastError instanceof IllegalArgumentException);
563      }
564  
565      /**
# 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);
# Line 722 | Line 712 | public class SubmissionPublisherTest ext
712       * offer reports drops if saturated
713       */
714      public void testDroppedOffer() {
715 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
716 <            basicExecutor, 4);
715 >        SubmissionPublisher<Integer> p
716 >            = new SubmissionPublisher<>(basicExecutor, 4);
717          TestSubscriber s1 = new TestSubscriber();
718          s1.request = false;
719          TestSubscriber s2 = new TestSubscriber();
# Line 751 | Line 741 | public class SubmissionPublisherTest ext
741       */
742      public void testHandledDroppedOffer() {
743          AtomicInteger calls = new AtomicInteger();
744 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
745 <            basicExecutor, 4);
744 >        SubmissionPublisher<Integer> p
745 >            = new SubmissionPublisher<>(basicExecutor, 4);
746          TestSubscriber s1 = new TestSubscriber();
747          s1.request = false;
748          TestSubscriber s2 = new TestSubscriber();
# Line 779 | Line 769 | public class SubmissionPublisherTest ext
769       */
770      public void testRecoveredHandledDroppedOffer() {
771          AtomicInteger calls = new AtomicInteger();
772 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
773 <            basicExecutor, 4);
772 >        SubmissionPublisher<Integer> p
773 >            = new SubmissionPublisher<>(basicExecutor, 4);
774          TestSubscriber s1 = new TestSubscriber();
775          s1.request = false;
776          TestSubscriber s2 = new TestSubscriber();
# Line 806 | Line 796 | public class SubmissionPublisherTest ext
796       */
797      public void testEmptyTimedOffer() {
798          SubmissionPublisher<Integer> p = basicPublisher();
799 +        long startTime = System.nanoTime();
800          assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
801 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
802      }
803  
804      /**
# Line 814 | Line 806 | public class SubmissionPublisherTest ext
806       */
807      public void testNullTimedOffer() {
808          SubmissionPublisher<Integer> p = basicPublisher();
809 +        long startTime = System.nanoTime();
810          try {
811 <            p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
811 >            p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
812              shouldThrow();
813          } catch (NullPointerException success) {}
814          try {
815 <            p.offer(1, SHORT_DELAY_MS, null, null);
815 >            p.offer(1, LONG_DELAY_MS, null, null);
816              shouldThrow();
817          } catch (NullPointerException success) {}
818 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
819      }
820  
821      /**
# Line 837 | Line 831 | public class SubmissionPublisherTest ext
831          p.subscribe(s2);
832          s2.awaitSubscribe();
833          s1.awaitSubscribe();
834 <        assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
835 <        assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
834 >        long startTime = System.nanoTime();
835 >        assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
836 >        assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
837          s1.sn.request(4);
838 <        assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
838 >        assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
839          s2.sn.request(4);
840 <        p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
840 >        p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
841          p.close();
842          s2.awaitComplete();
843          assertEquals(4, s2.nexts);
844          s1.awaitComplete();
845          assertEquals(4, s2.nexts);
846 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
847      }
848  
849      /**
850       * Timed offer reports drops if saturated
851       */
852      public void testDroppedTimedOffer() {
853 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
854 <            basicExecutor, 4);
853 >        SubmissionPublisher<Integer> p
854 >            = new SubmissionPublisher<>(basicExecutor, 4);
855          TestSubscriber s1 = new TestSubscriber();
856          s1.request = false;
857          TestSubscriber s2 = new TestSubscriber();
# Line 864 | Line 860 | public class SubmissionPublisherTest ext
860          p.subscribe(s2);
861          s2.awaitSubscribe();
862          s1.awaitSubscribe();
863 +        long delay = timeoutMillis();
864          for (int i = 1; i <= 4; ++i)
865 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
866 <        p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
867 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
865 >            assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
866 >        long startTime = System.nanoTime();
867 >        assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
868          s1.sn.request(64);
869 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
869 >        assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
870 >        // 2 * delay should elapse but check only 1 * delay to allow timer slop
871 >        assertTrue(millisElapsedSince(startTime) >= delay);
872          s2.sn.request(64);
873          p.close();
874          s2.awaitComplete();
# Line 883 | Line 882 | public class SubmissionPublisherTest ext
882       */
883      public void testHandledDroppedTimedOffer() {
884          AtomicInteger calls = new AtomicInteger();
885 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
886 <            basicExecutor, 4);
885 >        SubmissionPublisher<Integer> p
886 >            = new SubmissionPublisher<>(basicExecutor, 4);
887          TestSubscriber s1 = new TestSubscriber();
888          s1.request = false;
889          TestSubscriber s2 = new TestSubscriber();
# Line 893 | Line 892 | public class SubmissionPublisherTest ext
892          p.subscribe(s2);
893          s2.awaitSubscribe();
894          s1.awaitSubscribe();
895 +        long delay = timeoutMillis();
896          for (int i = 1; i <= 4; ++i)
897 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
898 <        p.offer(5, (s, x) -> noopHandle(calls));
899 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
897 >            assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
898 >        long startTime = System.nanoTime();
899 >        assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
900          s1.sn.request(64);
901 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
901 >        assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
902 >        assertTrue(millisElapsedSince(startTime) >= delay);
903          s2.sn.request(64);
904          p.close();
905          s2.awaitComplete();
# Line 911 | Line 912 | public class SubmissionPublisherTest ext
912       */
913      public void testRecoveredHandledDroppedTimedOffer() {
914          AtomicInteger calls = new AtomicInteger();
915 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
916 <            basicExecutor, 4);
915 >        SubmissionPublisher<Integer> p
916 >            = new SubmissionPublisher<>(basicExecutor, 4);
917          TestSubscriber s1 = new TestSubscriber();
918          s1.request = false;
919          TestSubscriber s2 = new TestSubscriber();
# Line 922 | Line 923 | public class SubmissionPublisherTest ext
923          s2.awaitSubscribe();
924          s1.awaitSubscribe();
925          int n = 0;
926 <        for (int i = 1; i <= 8; ++i) {
927 <            int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
926 >        long delay = timeoutMillis();
927 >        long startTime = System.nanoTime();
928 >        for (int i = 1; i <= 6; ++i) {
929 >            int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
930              n = n + 2 + (d < 0 ? d : 0);
931          }
932 +        assertTrue(millisElapsedSince(startTime) >= delay);
933          p.close();
934          s2.awaitComplete();
935          s1.awaitComplete();
# Line 933 | Line 937 | public class SubmissionPublisherTest ext
937          assertTrue(calls.get() >= 2);
938      }
939  
940 +    /**
941 +     * consume returns a CompletableFuture that is done when
942 +     * publisher completes
943 +     */
944 +    public void testConsume() {
945 +        AtomicInteger sum = new AtomicInteger();
946 +        SubmissionPublisher<Integer> p = basicPublisher();
947 +        CompletableFuture<Void> f =
948 +            p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
949 +        int n = 20;
950 +        for (int i = 1; i <= n; ++i)
951 +            p.submit(i);
952 +        p.close();
953 +        f.join();
954 +        assertEquals((n * (n + 1)) / 2, sum.get());
955 +    }
956 +
957 +    /**
958 +     * consume(null) throws NPE
959 +     */
960 +    public void testConsumeNPE() {
961 +        SubmissionPublisher<Integer> p = basicPublisher();
962 +        try {
963 +            CompletableFuture<Void> f = p.consume(null);
964 +            shouldThrow();
965 +        } catch (NullPointerException success) {}
966 +    }
967 +
968 +    /**
969 +     * consume eventually stops processing published items if cancelled
970 +     */
971 +    public void testCancelledConsume() {
972 +        AtomicInteger count = new AtomicInteger();
973 +        SubmissionPublisher<Integer> p = basicPublisher();
974 +        CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
975 +        f.cancel(true);
976 +        int n = 1000000; // arbitrary limit
977 +        for (int i = 1; i <= n; ++i)
978 +            p.submit(i);
979 +        assertTrue(count.get() < n);
980 +    }
981 +
982   }

Diff Legend

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