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.13 by jsr166, Sat Sep 12 18:19:57 2015 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.LinkedBlockingQueue;
14   import java.util.concurrent.SubmissionPublisher;
15   import java.util.concurrent.ThreadFactory;
16   import java.util.concurrent.ThreadPoolExecutor;
17   import java.util.concurrent.TimeUnit;
18   import java.util.concurrent.atomic.AtomicInteger;
19   import java.util.function.BiConsumer;
25 – import java.util.function.BiPredicate;
20   import java.util.function.BiFunction;
21 <
21 > import java.util.function.BiPredicate;
22 > import java.util.stream.Stream;
23   import junit.framework.Test;
24   import junit.framework.TestSuite;
25  
26 + import static java.util.concurrent.Flow.Publisher;
27 + import static java.util.concurrent.Flow.Subscriber;
28 + import static java.util.concurrent.Flow.Subscription;
29 + import static java.util.concurrent.TimeUnit.MILLISECONDS;
30 + import static java.util.concurrent.TimeUnit.SECONDS;
31 +
32   public class SubmissionPublisherTest extends JSR166TestCase {
33  
34      public static void main(String[] args) {
# Line 37 | Line 38 | public class SubmissionPublisherTest ext
38          return new TestSuite(SubmissionPublisherTest.class);
39      }
40  
41 <    // 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());
41 >    final Executor basicExecutor = basicPublisher().getExecutor();
42  
43      static SubmissionPublisher<Integer> basicPublisher() {
44 <        return new SubmissionPublisher<Integer>(basicExecutor,
58 <                                                Flow.defaultBufferSize());
44 >        return new SubmissionPublisher<Integer>();
45      }
46  
47      static class SPException extends RuntimeException {}
# Line 167 | Line 153 | public class SubmissionPublisherTest ext
153      /**
154       * A default-constructed SubmissionPublisher has no subscribers,
155       * is not closed, has default buffer size, and uses the
156 <     * ForkJoinPool.commonPool executor
156 >     * defaultExecutor
157       */
158      public void testConstructor1() {
159          SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>();
160          checkInitialState(p);
175 –        assertSame(p.getExecutor(), ForkJoinPool.commonPool());
161          assertEquals(p.getMaxBufferCapacity(), Flow.defaultBufferSize());
162 +        Executor e = p.getExecutor(), c = ForkJoinPool.commonPool();
163 +        if (ForkJoinPool.getCommonPoolParallelism() > 1)
164 +            assertSame(e, c);
165 +        else
166 +            assertNotSame(e, c);
167      }
168  
169      /**
# Line 285 | Line 275 | public class SubmissionPublisherTest ext
275          assertEquals(0, s2.nexts);
276          assertEquals(0, s2.errors);
277          assertEquals(0, s2.completes);
278 +        p.close();
279      }
280  
281      /**
# Line 360 | Line 351 | public class SubmissionPublisherTest ext
351      }
352  
353      /**
354 <     * subscribe(null) thows NPE
354 >     * subscribe(null) throws NPE
355       */
356      public void testSubscribe6() {
357          SubmissionPublisher<Integer> p = basicPublisher();
# Line 455 | Line 446 | public class SubmissionPublisherTest ext
446      }
447  
448      /**
449 <     * If a handler is supplied in conctructor, it is invoked when
449 >     * If a handler is supplied in constructor, it is invoked when
450       * subscriber throws an exception in onNext
451       */
452      public void testThrowOnNextHandler() {
# Line 806 | Line 797 | public class SubmissionPublisherTest ext
797       */
798      public void testEmptyTimedOffer() {
799          SubmissionPublisher<Integer> p = basicPublisher();
800 +        long startTime = System.nanoTime();
801          assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
802 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
803      }
804  
805      /**
# Line 814 | Line 807 | public class SubmissionPublisherTest ext
807       */
808      public void testNullTimedOffer() {
809          SubmissionPublisher<Integer> p = basicPublisher();
810 +        long startTime = System.nanoTime();
811          try {
812 <            p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
812 >            p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
813              shouldThrow();
814          } catch (NullPointerException success) {}
815          try {
816 <            p.offer(1, SHORT_DELAY_MS, null, null);
816 >            p.offer(1, LONG_DELAY_MS, null, null);
817              shouldThrow();
818          } catch (NullPointerException success) {}
819 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
820      }
821  
822      /**
# Line 837 | Line 832 | public class SubmissionPublisherTest ext
832          p.subscribe(s2);
833          s2.awaitSubscribe();
834          s1.awaitSubscribe();
835 <        assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
836 <        assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
835 >        long startTime = System.nanoTime();
836 >        assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
837 >        assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
838          s1.sn.request(4);
839 <        assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
839 >        assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
840          s2.sn.request(4);
841 <        p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
841 >        p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
842          p.close();
843          s2.awaitComplete();
844          assertEquals(4, s2.nexts);
845          s1.awaitComplete();
846          assertEquals(4, s2.nexts);
847 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
848      }
849  
850      /**
# Line 864 | Line 861 | public class SubmissionPublisherTest ext
861          p.subscribe(s2);
862          s2.awaitSubscribe();
863          s1.awaitSubscribe();
864 +        long delay = timeoutMillis();
865          for (int i = 1; i <= 4; ++i)
866 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
867 <        p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
868 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
866 >            assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
867 >        long startTime = System.nanoTime();
868 >        assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
869          s1.sn.request(64);
870 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
870 >        assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
871 >        // 2 * delay should elapse but check only 1 * delay to allow timer slop
872 >        assertTrue(millisElapsedSince(startTime) >= delay);
873          s2.sn.request(64);
874          p.close();
875          s2.awaitComplete();
# Line 893 | Line 893 | public class SubmissionPublisherTest ext
893          p.subscribe(s2);
894          s2.awaitSubscribe();
895          s1.awaitSubscribe();
896 +        long delay = timeoutMillis();
897          for (int i = 1; i <= 4; ++i)
898 <            assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
899 <        p.offer(5, (s, x) -> noopHandle(calls));
900 <        assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
898 >            assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
899 >        long startTime = System.nanoTime();
900 >        assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
901          s1.sn.request(64);
902 <        assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
902 >        assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
903 >        assertTrue(millisElapsedSince(startTime) >= delay);
904          s2.sn.request(64);
905          p.close();
906          s2.awaitComplete();
# Line 922 | Line 924 | public class SubmissionPublisherTest ext
924          s2.awaitSubscribe();
925          s1.awaitSubscribe();
926          int n = 0;
927 <        for (int i = 1; i <= 8; ++i) {
928 <            int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
927 >        long delay = timeoutMillis();
928 >        long startTime = System.nanoTime();
929 >        for (int i = 1; i <= 6; ++i) {
930 >            int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
931              n = n + 2 + (d < 0 ? d : 0);
932          }
933 +        assertTrue(millisElapsedSince(startTime) >= delay);
934          p.close();
935          s2.awaitComplete();
936          s1.awaitComplete();
# Line 933 | Line 938 | public class SubmissionPublisherTest ext
938          assertTrue(calls.get() >= 2);
939      }
940  
941 +    /**
942 +     * consume returns a CompletableFuture that is done when
943 +     * publisher completes
944 +     */
945 +    public void testConsume() {
946 +        AtomicInteger sum = new AtomicInteger();
947 +        SubmissionPublisher<Integer> p = basicPublisher();
948 +        CompletableFuture<Void> f =
949 +            p.consume((Integer x) -> { sum.getAndAdd(x.intValue()); });
950 +        int n = 20;
951 +        for (int i = 1; i <= n; ++i)
952 +            p.submit(i);
953 +        p.close();
954 +        f.join();
955 +        assertEquals((n * (n + 1)) / 2, sum.get());
956 +    }
957 +
958 +    /**
959 +     * consume(null) throws NPE
960 +     */
961 +    public void testConsumeNPE() {
962 +        SubmissionPublisher<Integer> p = basicPublisher();
963 +        try {
964 +            CompletableFuture<Void> f = p.consume(null);
965 +            shouldThrow();
966 +        } catch (NullPointerException success) {}
967 +    }
968 +
969 +    /**
970 +     * consume eventually stops processing published items if cancelled
971 +     */
972 +    public void testCancelledConsume() {
973 +        AtomicInteger count = new AtomicInteger();
974 +        SubmissionPublisher<Integer> p = basicPublisher();
975 +        CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
976 +        f.cancel(true);
977 +        int n = 1000000; // arbitrary limit
978 +        for (int i = 1; i <= n; ++i)
979 +            p.submit(i);
980 +        assertTrue(count.get() < n);
981 +    }
982 +
983   }

Diff Legend

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