ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/test/tck/SubmissionPublisherTest.java

Comparing jsr166/src/test/tck/SubmissionPublisherTest.java (file contents):
Revision 1.9 by dl, Tue Sep 8 19:44:10 2015 UTC vs.
Revision 1.17 by dl, Thu Dec 15 17:34:51 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() > 1) ?
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 395 | 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 405 | 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 948 | Line 931 | public class SubmissionPublisherTest ext
931		assertTrue(calls.get() >= 2);
932		}
933
934	+	/**
935	+	* consume returns a CompletableFuture that is done when
936	+	* publisher completes
937	+	*/
938	+	public void testConsume() {
939	+	AtomicInteger sum = new AtomicInteger();
940	+	SubmissionPublisher<Integer> p = basicPublisher();
941	+	CompletableFuture<Void> f =
942	+	p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
943	+	int n = 20;
944	+	for (int i = 1; i <= n; ++i)
945	+	p.submit(i);
946	+	p.close();
947	+	f.join();
948	+	assertEquals((n * (n + 1)) / 2, sum.get());
949	+	}
950	+
951	+	/**
952	+	* consume(null) throws NPE
953	+	*/
954	+	public void testConsumeNPE() {
955	+	SubmissionPublisher<Integer> p = basicPublisher();
956	+	try {
957	+	CompletableFuture<Void> f = p.consume(null);
958	+	shouldThrow();
959	+	} catch (NullPointerException success) {}
960	+	}
961	+
962	+	/**
963	+	* consume eventually stops processing published items if cancelled
964	+	*/
965	+	public void testCancelledConsume() {
966	+	AtomicInteger count = new AtomicInteger();
967	+	SubmissionPublisher<Integer> p = basicPublisher();
968	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
969	+	f.cancel(true);
970	+	int n = 1000000; // arbitrary limit
971	+	for (int i = 1; i <= n; ++i)
972	+	p.submit(i);
973	+	assertTrue(count.get() < n);
974	+	}
975	+
976		}

Diff Legend

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