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.7 by jsr166, Mon Sep 7 21:25:50 2015 UTC vs.
Revision 1.14 by jsr166, Wed Aug 24 22:22:39 2016 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>();
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 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 455 | Line 436 | public class SubmissionPublisherTest ext
436      }
437  
438      /**
439 <     * If a handler is supplied in conctructor, it is invoked when
439 >     * If a handler is supplied in constructor, it is invoked when
440       * subscriber throws an exception in onNext
441       */
442      public void testThrowOnNextHandler() {
# Line 870 | Line 851 | public class SubmissionPublisherTest ext
851          p.subscribe(s2);
852          s2.awaitSubscribe();
853          s1.awaitSubscribe();
854 +        long delay = timeoutMillis();
855          for (int i = 1; i <= 4; ++i)
856 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
857 <        p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
858 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
856 >            assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
857 >        long startTime = System.nanoTime();
858 >        assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
859          s1.sn.request(64);
860 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
860 >        assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
861 >        // 2 * delay should elapse but check only 1 * delay to allow timer slop
862 >        assertTrue(millisElapsedSince(startTime) >= delay);
863          s2.sn.request(64);
864          p.close();
865          s2.awaitComplete();
# Line 899 | Line 883 | public class SubmissionPublisherTest ext
883          p.subscribe(s2);
884          s2.awaitSubscribe();
885          s1.awaitSubscribe();
886 +        long delay = timeoutMillis();
887          for (int i = 1; i <= 4; ++i)
888 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
889 <        p.offer(5, (s, x) -> noopHandle(calls));
890 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
888 >            assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
889 >        long startTime = System.nanoTime();
890 >        assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
891          s1.sn.request(64);
892 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
892 >        assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
893 >        assertTrue(millisElapsedSince(startTime) >= delay);
894          s2.sn.request(64);
895          p.close();
896          s2.awaitComplete();
# Line 928 | Line 914 | public class SubmissionPublisherTest ext
914          s2.awaitSubscribe();
915          s1.awaitSubscribe();
916          int n = 0;
917 <        for (int i = 1; i <= 8; ++i) {
918 <            int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
917 >        long delay = timeoutMillis();
918 >        long startTime = System.nanoTime();
919 >        for (int i = 1; i <= 6; ++i) {
920 >            int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
921              n = n + 2 + (d < 0 ? d : 0);
922          }
923 +        assertTrue(millisElapsedSince(startTime) >= delay);
924          p.close();
925          s2.awaitComplete();
926          s1.awaitComplete();
# Line 939 | Line 928 | public class SubmissionPublisherTest ext
928          assertTrue(calls.get() >= 2);
929      }
930  
931 +    /**
932 +     * consume returns a CompletableFuture that is done when
933 +     * publisher completes
934 +     */
935 +    public void testConsume() {
936 +        AtomicInteger sum = new AtomicInteger();
937 +        SubmissionPublisher<Integer> p = basicPublisher();
938 +        CompletableFuture<Void> f =
939 +            p.consume((Integer x) -> { sum.getAndAdd(x.intValue()); });
940 +        int n = 20;
941 +        for (int i = 1; i <= n; ++i)
942 +            p.submit(i);
943 +        p.close();
944 +        f.join();
945 +        assertEquals((n * (n + 1)) / 2, sum.get());
946 +    }
947 +
948 +    /**
949 +     * consume(null) throws NPE
950 +     */
951 +    public void testConsumeNPE() {
952 +        SubmissionPublisher<Integer> p = basicPublisher();
953 +        try {
954 +            CompletableFuture<Void> f = p.consume(null);
955 +            shouldThrow();
956 +        } catch (NullPointerException success) {}
957 +    }
958 +
959 +    /**
960 +     * consume eventually stops processing published items if cancelled
961 +     */
962 +    public void testCancelledConsume() {
963 +        AtomicInteger count = new AtomicInteger();
964 +        SubmissionPublisher<Integer> p = basicPublisher();
965 +        CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
966 +        f.cancel(true);
967 +        int n = 1000000; // arbitrary limit
968 +        for (int i = 1; i <= n; ++i)
969 +            p.submit(i);
970 +        assertTrue(count.get() < n);
971 +    }
972 +
973   }

Diff Legend

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