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.4 by jsr166, Mon Sep 7 20:33:41 2015 UTC vs.
Revision 1.15 by jsr166, Sun Nov 6 22:50:32 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 153 | Line 129 | public class SubmissionPublisherTest ext
129       */
130      void checkInitialState(SubmissionPublisher<?> p) {
131          assertFalse(p.hasSubscribers());
132 <        assertEquals(p.getNumberOfSubscribers(), 0);
132 >        assertEquals(0, p.getNumberOfSubscribers());
133          assertTrue(p.getSubscribers().isEmpty());
134          assertFalse(p.isClosed());
135          assertNull(p.getClosedException());
136          int n = p.getMaxBufferCapacity();
137          assertTrue((n & (n - 1)) == 0); // power of two
138          assertNotNull(p.getExecutor());
139 <        assertEquals(p.estimateMinimumDemand(), 0);
140 <        assertEquals(p.estimateMaximumLag(), 0);
139 >        assertEquals(0, p.estimateMinimumDemand());
140 >        assertEquals(0, p.estimateMaximumLag());
141      }
142  
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 185 | Line 165 | public class SubmissionPublisherTest ext
165          SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, 8);
166          checkInitialState(p);
167          assertSame(p.getExecutor(), e);
168 <        assertEquals(p.getMaxBufferCapacity(), 8);
168 >        assertEquals(8, p.getMaxBufferCapacity());
169      }
170  
171      /**
# Line 264 | Line 244 | public class SubmissionPublisherTest ext
244          SubmissionPublisher<Integer> p = basicPublisher();
245          p.subscribe(s);
246          assertTrue(p.hasSubscribers());
247 <        assertEquals(p.getNumberOfSubscribers(), 1);
247 >        assertEquals(1, p.getNumberOfSubscribers());
248          assertTrue(p.getSubscribers().contains(s));
249          assertTrue(p.isSubscribed(s));
250          s.awaitSubscribe();
251          assertNotNull(s.sn);
252 <        assertEquals(s.nexts, 0);
253 <        assertEquals(s.errors, 0);
254 <        assertEquals(s.completes, 0);
252 >        assertEquals(0, s.nexts);
253 >        assertEquals(0, s.errors);
254 >        assertEquals(0, s.completes);
255          TestSubscriber s2 = new TestSubscriber();
256          p.subscribe(s2);
257          assertTrue(p.hasSubscribers());
258 <        assertEquals(p.getNumberOfSubscribers(), 2);
258 >        assertEquals(2, p.getNumberOfSubscribers());
259          assertTrue(p.getSubscribers().contains(s));
260          assertTrue(p.getSubscribers().contains(s2));
261          assertTrue(p.isSubscribed(s));
262          assertTrue(p.isSubscribed(s2));
263          s2.awaitSubscribe();
264          assertNotNull(s2.sn);
265 <        assertEquals(s2.nexts, 0);
266 <        assertEquals(s2.errors, 0);
267 <        assertEquals(s2.completes, 0);
265 >        assertEquals(0, s2.nexts);
266 >        assertEquals(0, s2.errors);
267 >        assertEquals(0, s2.completes);
268 >        p.close();
269      }
270  
271      /**
# Line 297 | Line 278 | public class SubmissionPublisherTest ext
278          p.close();
279          p.subscribe(s);
280          s.awaitComplete();
281 <        assertEquals(s.nexts, 0);
282 <        assertEquals(s.errors, 0);
283 <        assertEquals(s.completes, 1);
281 >        assertEquals(0, s.nexts);
282 >        assertEquals(0, s.errors);
283 >        assertEquals(1, s.completes, 1);
284      }
285  
286      /**
# Line 315 | Line 296 | public class SubmissionPublisherTest ext
296          assertSame(p.getClosedException(), ex);
297          p.subscribe(s);
298          s.awaitError();
299 <        assertEquals(s.nexts, 0);
300 <        assertEquals(s.errors, 1);
299 >        assertEquals(0, s.nexts);
300 >        assertEquals(1, s.errors);
301      }
302  
303      /**
# Line 328 | Line 309 | public class SubmissionPublisherTest ext
309          SubmissionPublisher<Integer> p = basicPublisher();
310          p.subscribe(s);
311          assertTrue(p.hasSubscribers());
312 <        assertEquals(p.getNumberOfSubscribers(), 1);
312 >        assertEquals(1, p.getNumberOfSubscribers());
313          assertTrue(p.getSubscribers().contains(s));
314          assertTrue(p.isSubscribed(s));
315          s.awaitSubscribe();
316          assertNotNull(s.sn);
317 <        assertEquals(s.nexts, 0);
318 <        assertEquals(s.errors, 0);
319 <        assertEquals(s.completes, 0);
317 >        assertEquals(0, s.nexts);
318 >        assertEquals(0, s.errors);
319 >        assertEquals(0, s.completes);
320          p.subscribe(s);
321          s.awaitError();
322 <        assertEquals(s.nexts, 0);
323 <        assertEquals(s.errors, 1);
322 >        assertEquals(0, s.nexts);
323 >        assertEquals(1, s.errors);
324          assertFalse(p.isSubscribed(s));
325      }
326  
# Line 354 | Line 335 | public class SubmissionPublisherTest ext
335              p.subscribe(s);
336          } catch (Exception ok) {}
337          s.awaitError();
338 <        assertEquals(s.nexts, 0);
339 <        assertEquals(s.errors, 1);
340 <        assertEquals(s.completes, 0);
338 >        assertEquals(0, s.nexts);
339 >        assertEquals(1, s.errors);
340 >        assertEquals(0, s.completes);
341      }
342  
343      /**
344 <     * subscribe(null) thows NPE
344 >     * subscribe(null) throws NPE
345       */
346      public void testSubscribe6() {
347          SubmissionPublisher<Integer> p = basicPublisher();
# Line 385 | Line 366 | public class SubmissionPublisherTest ext
366          assertTrue(p.isClosed());
367          assertNull(p.getClosedException());
368          s1.awaitComplete();
369 <        assertEquals(s1.nexts, 1);
370 <        assertEquals(s1.completes, 1);
369 >        assertEquals(1, s1.nexts);
370 >        assertEquals(1, s1.completes);
371          s2.awaitComplete();
372 <        assertEquals(s2.nexts, 1);
373 <        assertEquals(s2.completes, 1);
372 >        assertEquals(1, s2.nexts);
373 >        assertEquals(1, s2.completes);
374      }
375  
376      /**
# Line 406 | Line 387 | public class SubmissionPublisherTest ext
387          assertTrue(p.isClosed());
388          s1.awaitError();
389          assertTrue(s1.nexts <= 1);
390 <        assertEquals(s1.errors, 1);
390 >        assertEquals(1, s1.errors);
391          s2.awaitError();
392          assertTrue(s2.nexts <= 1);
393 <        assertEquals(s2.errors, 1);
393 >        assertEquals(1, s2.errors);
394      }
395  
396      /**
# Line 428 | Line 409 | public class SubmissionPublisherTest ext
409              p.submit(i);
410          p.close();
411          s2.awaitComplete();
412 <        assertEquals(s2.nexts, 20);
413 <        assertEquals(s2.completes, 1);
412 >        assertEquals(20, s2.nexts);
413 >        assertEquals(1, s2.completes);
414          assertTrue(s1.nexts < 20);
415          assertFalse(p.isSubscribed(s1));
416      }
# Line 449 | Line 430 | public class SubmissionPublisherTest ext
430          p.submit(2);
431          p.close();
432          s2.awaitComplete();
433 <        assertEquals(s2.nexts, 2);
433 >        assertEquals(2, s2.nexts);
434          s1.awaitComplete();
435 <        assertEquals(s1.errors, 1);
435 >        assertEquals(1, s1.errors);
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 473 | Line 454 | public class SubmissionPublisherTest ext
454          p.submit(2);
455          p.close();
456          s2.awaitComplete();
457 <        assertEquals(s2.nexts, 2);
458 <        assertEquals(s2.completes, 1);
457 >        assertEquals(2, s2.nexts);
458 >        assertEquals(1, s2.completes);
459          s1.awaitError();
460 <        assertEquals(s1.errors, 1);
461 <        assertEquals(calls.get(), 1);
460 >        assertEquals(1, s1.errors);
461 >        assertEquals(1, calls.get());
462      }
463  
464      /**
# Line 494 | Line 475 | public class SubmissionPublisherTest ext
475          p.close();
476          s2.awaitComplete();
477          s1.awaitComplete();
478 <        assertEquals(s2.nexts, 20);
479 <        assertEquals(s2.completes, 1);
480 <        assertEquals(s1.nexts, 20);
481 <        assertEquals(s1.completes, 1);
478 >        assertEquals(20, s2.nexts);
479 >        assertEquals(1, s2.completes);
480 >        assertEquals(20, s1.nexts);
481 >        assertEquals(1, s1.completes);
482      }
483  
484      /**
# Line 515 | Line 496 | public class SubmissionPublisherTest ext
496          p.submit(1);
497          p.submit(2);
498          s2.awaitNext(1);
499 <        assertEquals(s1.nexts, 0);
499 >        assertEquals(0, s1.nexts);
500          s1.sn.request(3);
501          p.submit(3);
502          p.close();
503          s2.awaitComplete();
504 <        assertEquals(s2.nexts, 3);
505 <        assertEquals(s2.completes, 1);
504 >        assertEquals(3, s2.nexts);
505 >        assertEquals(1, s2.completes);
506          s1.awaitComplete();
507          assertTrue(s1.nexts > 0);
508 <        assertEquals(s1.completes, 1);
508 >        assertEquals(1, s1.completes);
509      }
510  
511      /**
# Line 543 | Line 524 | public class SubmissionPublisherTest ext
524          p.submit(2);
525          p.close();
526          s2.awaitComplete();
527 <        assertEquals(s2.nexts, 2);
528 <        assertEquals(s2.completes, 1);
527 >        assertEquals(2, s2.nexts);
528 >        assertEquals(1, s2.completes);
529          s1.awaitNext(1);
530 <        assertEquals(s1.nexts, 1);
530 >        assertEquals(1, s1.nexts);
531      }
532  
533      /**
# Line 565 | Line 546 | public class SubmissionPublisherTest ext
546          p.submit(2);
547          p.close();
548          s2.awaitComplete();
549 <        assertEquals(s2.nexts, 2);
550 <        assertEquals(s2.completes, 1);
549 >        assertEquals(2, s2.nexts);
550 >        assertEquals(1, s2.completes);
551          s1.awaitError();
552 <        assertEquals(s1.errors, 1);
552 >        assertEquals(1, s1.errors);
553          assertTrue(s1.lastError instanceof IllegalArgumentException);
554      }
555  
# Line 582 | Line 563 | public class SubmissionPublisherTest ext
563          s.request = false;
564          p.subscribe(s);
565          s.awaitSubscribe();
566 <        assertEquals(p.estimateMinimumDemand(), 0);
566 >        assertEquals(0, p.estimateMinimumDemand());
567          s.sn.request(1);
568 <        assertEquals(p.estimateMinimumDemand(), 1);
568 >        assertEquals(1, p.estimateMinimumDemand());
569          p.submit(1);
570          s.awaitNext(1);
571 <        assertEquals(p.estimateMinimumDemand(), 0);
571 >        assertEquals(0, p.estimateMinimumDemand());
572      }
573  
574      /**
# Line 595 | Line 576 | public class SubmissionPublisherTest ext
576       */
577      public void testEmptySubmit() {
578          SubmissionPublisher<Integer> p = basicPublisher();
579 <        assertEquals(p.submit(1), 0);
579 >        assertEquals(0, p.submit(1));
580      }
581  
582      /**
# Line 623 | Line 604 | public class SubmissionPublisherTest ext
604          p.subscribe(s2);
605          s2.awaitSubscribe();
606          s1.awaitSubscribe();
607 <        assertEquals(p.submit(1), 1);
607 >        assertEquals(1, p.submit(1));
608          assertTrue(p.estimateMaximumLag() >= 1);
609          assertTrue(p.submit(2) >= 2);
610          assertTrue(p.estimateMaximumLag() >= 2);
# Line 634 | Line 615 | public class SubmissionPublisherTest ext
615          p.submit(4);
616          p.close();
617          s2.awaitComplete();
618 <        assertEquals(s2.nexts, 4);
618 >        assertEquals(4, s2.nexts);
619          s1.awaitComplete();
620 <        assertEquals(s2.nexts, 4);
620 >        assertEquals(4, s2.nexts);
621      }
622  
623      /**
# Line 656 | Line 637 | public class SubmissionPublisherTest ext
637          p.close();
638          s2.awaitComplete();
639          s1.awaitComplete();
640 <        assertEquals(s2.nexts, 20);
641 <        assertEquals(s2.completes, 1);
642 <        assertEquals(s1.nexts, 20);
643 <        assertEquals(s1.completes, 1);
640 >        assertEquals(20, s2.nexts);
641 >        assertEquals(1, s2.completes);
642 >        assertEquals(20, s1.nexts);
643 >        assertEquals(1, s1.completes);
644      }
645  
646      static boolean noopHandle(AtomicInteger count) {
# Line 713 | Line 694 | public class SubmissionPublisherTest ext
694          p.offer(4, null);
695          p.close();
696          s2.awaitComplete();
697 <        assertEquals(s2.nexts, 4);
697 >        assertEquals(4, s2.nexts);
698          s1.awaitComplete();
699 <        assertEquals(s2.nexts, 4);
699 >        assertEquals(4, s2.nexts);
700      }
701  
702      /**
# Line 774 | Line 755 | public class SubmissionPublisherTest ext
755          assertTrue(calls.get() >= 4);
756      }
757  
777 –
758      /**
759       * offer succeeds if drop handler forces request
760       */
# Line 798 | Line 778 | public class SubmissionPublisherTest ext
778          p.close();
779          s2.awaitComplete();
780          s1.awaitComplete();
781 <        assertEquals(s1.nexts + s2.nexts, n);
781 >        assertEquals(n, s1.nexts + s2.nexts);
782          assertTrue(calls.get() >= 2);
783      }
784  
805 –
785      /**
786       * Timed offer to a publisher with no subscribers returns lag 0
787       */
788      public void testEmptyTimedOffer() {
789          SubmissionPublisher<Integer> p = basicPublisher();
790 +        long startTime = System.nanoTime();
791          assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
792 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
793      }
794  
795      /**
# Line 816 | Line 797 | public class SubmissionPublisherTest ext
797       */
798      public void testNullTimedOffer() {
799          SubmissionPublisher<Integer> p = basicPublisher();
800 +        long startTime = System.nanoTime();
801          try {
802 <            p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
802 >            p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
803              shouldThrow();
804          } catch (NullPointerException success) {}
805          try {
806 <            p.offer(1, SHORT_DELAY_MS, null, null);
806 >            p.offer(1, LONG_DELAY_MS, null, null);
807              shouldThrow();
808          } catch (NullPointerException success) {}
809 +        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
810      }
811  
812      /**
# Line 839 | Line 822 | public class SubmissionPublisherTest ext
822          p.subscribe(s2);
823          s2.awaitSubscribe();
824          s1.awaitSubscribe();
825 <        assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
826 <        assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
825 >        long startTime = System.nanoTime();
826 >        assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
827 >        assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
828          s1.sn.request(4);
829 <        assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
829 >        assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
830          s2.sn.request(4);
831 <        p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
831 >        p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
832          p.close();
833          s2.awaitComplete();
834 <        assertEquals(s2.nexts, 4);
834 >        assertEquals(4, s2.nexts);
835          s1.awaitComplete();
836 <        assertEquals(s2.nexts, 4);
836 >        assertEquals(4, s2.nexts);
837 >        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
838      }
839  
840      /**
# Line 866 | 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 895 | 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 924 | 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();
927 <        assertEquals(s1.nexts + s2.nexts, n);
927 >        assertEquals(n, s1.nexts + s2.nexts);
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