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.4 by jsr166, Mon Sep 7 20:33:41 2015 UTC vs.
Revision 1.27 by jsr166, Wed Jun 6 16:24:09 2018 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.CountDownLatch;
10		import java.util.concurrent.Executor;
11		import java.util.concurrent.Executors;
12		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;
13		import java.util.concurrent.ForkJoinPool;
14		import java.util.concurrent.SubmissionPublisher;
20	–	import java.util.concurrent.ThreadFactory;
21	–	import java.util.concurrent.ThreadPoolExecutor;
22	–	import java.util.concurrent.TimeUnit;
15		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	–
16		import junit.framework.Test;
17		import junit.framework.TestSuite;
18
19	+	import static java.util.concurrent.Flow.Subscriber;
20	+	import static java.util.concurrent.Flow.Subscription;
21	+	import static java.util.concurrent.TimeUnit.MILLISECONDS;
22	+
23		public class SubmissionPublisherTest extends JSR166TestCase {
24
25		public static void main(String[] args) {
#	Line 37 | Line 29 | public class SubmissionPublisherTest ext
29		return new TestSuite(SubmissionPublisherTest.class);
30		}
31
32	<	// 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());
32	>	final Executor basicExecutor = basicPublisher().getExecutor();
33
34		static SubmissionPublisher<Integer> basicPublisher() {
35	<	return new SubmissionPublisher<Integer>(basicExecutor,
58	<	Flow.defaultBufferSize());
35	>	return new SubmissionPublisher<Integer>();
36		}
37
38		static class SPException extends RuntimeException {}
#	Line 153 | Line 130 | public class SubmissionPublisherTest ext
130		*/
131		void checkInitialState(SubmissionPublisher<?> p) {
132		assertFalse(p.hasSubscribers());
133	<	assertEquals(p.getNumberOfSubscribers(), 0);
133	>	assertEquals(0, p.getNumberOfSubscribers());
134		assertTrue(p.getSubscribers().isEmpty());
135		assertFalse(p.isClosed());
136		assertNull(p.getClosedException());
137		int n = p.getMaxBufferCapacity();
138		assertTrue((n & (n - 1)) == 0); // power of two
139		assertNotNull(p.getExecutor());
140	<	assertEquals(p.estimateMinimumDemand(), 0);
141	<	assertEquals(p.estimateMaximumLag(), 0);
140	>	assertEquals(0, p.estimateMinimumDemand());
141	>	assertEquals(0, p.estimateMaximumLag());
142		}
143
144		/**
145		* A default-constructed SubmissionPublisher has no subscribers,
146		* is not closed, has default buffer size, and uses the
147	<	* ForkJoinPool.commonPool executor
147	>	* defaultExecutor
148		*/
149		public void testConstructor1() {
150	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>();
150	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>();
151		checkInitialState(p);
175	–	assertSame(p.getExecutor(), ForkJoinPool.commonPool());
152		assertEquals(p.getMaxBufferCapacity(), Flow.defaultBufferSize());
153	+	Executor e = p.getExecutor(), c = ForkJoinPool.commonPool();
154	+	if (ForkJoinPool.getCommonPoolParallelism() > 1)
155	+	assertSame(e, c);
156	+	else
157	+	assertNotSame(e, c);
158		}
159
160		/**
#	Line 182 | Line 163 | public class SubmissionPublisherTest ext
163		*/
164		public void testConstructor2() {
165		Executor e = Executors.newFixedThreadPool(1);
166	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, 8);
166	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>(e, 8);
167		checkInitialState(p);
168		assertSame(p.getExecutor(), e);
169	<	assertEquals(p.getMaxBufferCapacity(), 8);
169	>	assertEquals(8, p.getMaxBufferCapacity());
170		}
171
172		/**
173	<	* A null Executor argument to SubmissionPublisher constructor throws NPE
173	>	* A null Executor argument to SubmissionPublisher constructor
174	>	* throws NullPointerException
175		*/
176		public void testConstructor3() {
177		try {
#	Line 200 | Line 182 | public class SubmissionPublisherTest ext
182
183		/**
184		* A negative capacity argument to SubmissionPublisher constructor
185	<	* throws IAE
185	>	* throws IllegalArgumentException
186		*/
187		public void testConstructor4() {
188		Executor e = Executors.newFixedThreadPool(1);
#	Line 212 | Line 194 | public class SubmissionPublisherTest ext
194
195		/**
196		* A closed publisher reports isClosed with no closedException and
197	<	* throws ISE upon attempted submission; a subsequent close or
198	<	* closeExceptionally has no additional effect.
197	>	* throws IllegalStateException upon attempted submission; a
198	>	* subsequent close or closeExceptionally has no additional
199	>	* effect.
200		*/
201		public void testClose() {
202		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 233 | Line 216 | public class SubmissionPublisherTest ext
216
217		/**
218		* A publisher closedExceptionally reports isClosed with the
219	<	* closedException and throws ISE upon attempted submission; a
220	<	* subsequent close or closeExceptionally has no additional
221	<	* effect.
219	>	* closedException and throws IllegalStateException upon attempted
220	>	* submission; a subsequent close or closeExceptionally has no
221	>	* additional effect.
222		*/
223		public void testCloseExceptionally() {
224		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 264 | Line 247 | public class SubmissionPublisherTest ext
247		SubmissionPublisher<Integer> p = basicPublisher();
248		p.subscribe(s);
249		assertTrue(p.hasSubscribers());
250	<	assertEquals(p.getNumberOfSubscribers(), 1);
250	>	assertEquals(1, p.getNumberOfSubscribers());
251		assertTrue(p.getSubscribers().contains(s));
252		assertTrue(p.isSubscribed(s));
253		s.awaitSubscribe();
254		assertNotNull(s.sn);
255	<	assertEquals(s.nexts, 0);
256	<	assertEquals(s.errors, 0);
257	<	assertEquals(s.completes, 0);
255	>	assertEquals(0, s.nexts);
256	>	assertEquals(0, s.errors);
257	>	assertEquals(0, s.completes);
258		TestSubscriber s2 = new TestSubscriber();
259		p.subscribe(s2);
260		assertTrue(p.hasSubscribers());
261	<	assertEquals(p.getNumberOfSubscribers(), 2);
261	>	assertEquals(2, p.getNumberOfSubscribers());
262		assertTrue(p.getSubscribers().contains(s));
263		assertTrue(p.getSubscribers().contains(s2));
264		assertTrue(p.isSubscribed(s));
265		assertTrue(p.isSubscribed(s2));
266		s2.awaitSubscribe();
267		assertNotNull(s2.sn);
268	<	assertEquals(s2.nexts, 0);
269	<	assertEquals(s2.errors, 0);
270	<	assertEquals(s2.completes, 0);
268	>	assertEquals(0, s2.nexts);
269	>	assertEquals(0, s2.errors);
270	>	assertEquals(0, s2.completes);
271	>	p.close();
272		}
273
274		/**
#	Line 297 | Line 281 | public class SubmissionPublisherTest ext
281		p.close();
282		p.subscribe(s);
283		s.awaitComplete();
284	<	assertEquals(s.nexts, 0);
285	<	assertEquals(s.errors, 0);
286	<	assertEquals(s.completes, 1);
284	>	assertEquals(0, s.nexts);
285	>	assertEquals(0, s.errors);
286	>	assertEquals(1, s.completes, 1);
287		}
288
289		/**
#	Line 315 | Line 299 | public class SubmissionPublisherTest ext
299		assertSame(p.getClosedException(), ex);
300		p.subscribe(s);
301		s.awaitError();
302	<	assertEquals(s.nexts, 0);
303	<	assertEquals(s.errors, 1);
302	>	assertEquals(0, s.nexts);
303	>	assertEquals(1, s.errors);
304		}
305
306		/**
#	Line 328 | Line 312 | public class SubmissionPublisherTest ext
312		SubmissionPublisher<Integer> p = basicPublisher();
313		p.subscribe(s);
314		assertTrue(p.hasSubscribers());
315	<	assertEquals(p.getNumberOfSubscribers(), 1);
315	>	assertEquals(1, p.getNumberOfSubscribers());
316		assertTrue(p.getSubscribers().contains(s));
317		assertTrue(p.isSubscribed(s));
318		s.awaitSubscribe();
319		assertNotNull(s.sn);
320	<	assertEquals(s.nexts, 0);
321	<	assertEquals(s.errors, 0);
322	<	assertEquals(s.completes, 0);
320	>	assertEquals(0, s.nexts);
321	>	assertEquals(0, s.errors);
322	>	assertEquals(0, s.completes);
323		p.subscribe(s);
324		s.awaitError();
325	<	assertEquals(s.nexts, 0);
326	<	assertEquals(s.errors, 1);
325	>	assertEquals(0, s.nexts);
326	>	assertEquals(1, s.errors);
327		assertFalse(p.isSubscribed(s));
328		}
329
#	Line 350 | Line 334 | public class SubmissionPublisherTest ext
334		TestSubscriber s = new TestSubscriber();
335		SubmissionPublisher<Integer> p = basicPublisher();
336		s.throwOnCall = true;
337	<	try {
354	<	p.subscribe(s);
355	<	} catch (Exception ok) {}
337	>	p.subscribe(s);
338		s.awaitError();
339	<	assertEquals(s.nexts, 0);
340	<	assertEquals(s.errors, 1);
341	<	assertEquals(s.completes, 0);
339	>	assertEquals(0, s.nexts);
340	>	assertEquals(1, s.errors);
341	>	assertEquals(0, s.completes);
342		}
343
344		/**
345	<	* subscribe(null) thows NPE
345	>	* subscribe(null) throws NPE
346		*/
347		public void testSubscribe6() {
348		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 385 | Line 367 | public class SubmissionPublisherTest ext
367		assertTrue(p.isClosed());
368		assertNull(p.getClosedException());
369		s1.awaitComplete();
370	<	assertEquals(s1.nexts, 1);
371	<	assertEquals(s1.completes, 1);
370	>	assertEquals(1, s1.nexts);
371	>	assertEquals(1, s1.completes);
372		s2.awaitComplete();
373	<	assertEquals(s2.nexts, 1);
374	<	assertEquals(s2.completes, 1);
373	>	assertEquals(1, s2.nexts);
374	>	assertEquals(1, s2.completes);
375		}
376
377		/**
378		* Closing a publisher exceptionally causes onError to subscribers
379	+	* after they are subscribed
380		*/
381		public void testCloseExceptionallyError() {
382		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 404 | Line 387 | public class SubmissionPublisherTest ext
387		p.submit(1);
388		p.closeExceptionally(new SPException());
389		assertTrue(p.isClosed());
390	+	s1.awaitSubscribe();
391		s1.awaitError();
392		assertTrue(s1.nexts <= 1);
393	<	assertEquals(s1.errors, 1);
393	>	assertEquals(1, s1.errors);
394	>	s2.awaitSubscribe();
395		s2.awaitError();
396		assertTrue(s2.nexts <= 1);
397	<	assertEquals(s2.errors, 1);
397	>	assertEquals(1, s2.errors);
398		}
399
400		/**
401		* Cancelling a subscription eventually causes no more onNexts to be issued
402		*/
403		public void testCancel() {
404	<	SubmissionPublisher<Integer> p = basicPublisher();
404	>	SubmissionPublisher<Integer> p =
405	>	new SubmissionPublisher<>(basicExecutor, 4); // must be < 20
406		TestSubscriber s1 = new TestSubscriber();
407		TestSubscriber s2 = new TestSubscriber();
408		p.subscribe(s1);
#	Line 428 | Line 414 | public class SubmissionPublisherTest ext
414		p.submit(i);
415		p.close();
416		s2.awaitComplete();
417	<	assertEquals(s2.nexts, 20);
418	<	assertEquals(s2.completes, 1);
417	>	assertEquals(20, s2.nexts);
418	>	assertEquals(1, s2.completes);
419		assertTrue(s1.nexts < 20);
420		assertFalse(p.isSubscribed(s1));
421		}
#	Line 449 | Line 435 | public class SubmissionPublisherTest ext
435		p.submit(2);
436		p.close();
437		s2.awaitComplete();
438	<	assertEquals(s2.nexts, 2);
438	>	assertEquals(2, s2.nexts);
439		s1.awaitComplete();
440	<	assertEquals(s1.errors, 1);
440	>	assertEquals(1, s1.errors);
441		}
442
443		/**
444	<	* If a handler is supplied in conctructor, it is invoked when
444	>	* If a handler is supplied in constructor, it is invoked when
445		* subscriber throws an exception in onNext
446		*/
447		public void testThrowOnNextHandler() {
448		AtomicInteger calls = new AtomicInteger();
449	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
450	<	(basicExecutor, 8,
465	<	(s, e) -> calls.getAndIncrement());
449	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
450	>	basicExecutor, 8, (s, e) -> calls.getAndIncrement());
451		TestSubscriber s1 = new TestSubscriber();
452		TestSubscriber s2 = new TestSubscriber();
453		p.subscribe(s1);
#	Line 473 | Line 458 | public class SubmissionPublisherTest ext
458		p.submit(2);
459		p.close();
460		s2.awaitComplete();
461	<	assertEquals(s2.nexts, 2);
462	<	assertEquals(s2.completes, 1);
461	>	assertEquals(2, s2.nexts);
462	>	assertEquals(1, s2.completes);
463		s1.awaitError();
464	<	assertEquals(s1.errors, 1);
465	<	assertEquals(calls.get(), 1);
464	>	assertEquals(1, s1.errors);
465	>	assertEquals(1, calls.get());
466		}
467
468		/**
#	Line 494 | Line 479 | public class SubmissionPublisherTest ext
479		p.close();
480		s2.awaitComplete();
481		s1.awaitComplete();
482	<	assertEquals(s2.nexts, 20);
483	<	assertEquals(s2.completes, 1);
484	<	assertEquals(s1.nexts, 20);
485	<	assertEquals(s1.completes, 1);
482	>	assertEquals(20, s2.nexts);
483	>	assertEquals(1, s2.completes);
484	>	assertEquals(20, s1.nexts);
485	>	assertEquals(1, s1.completes);
486		}
487
488		/**
#	Line 509 | Line 494 | public class SubmissionPublisherTest ext
494		s1.request = false;
495		p.subscribe(s1);
496		s1.awaitSubscribe();
497	<	assertTrue(p.estimateMinimumDemand() == 0);
497	>	assertEquals(0, p.estimateMinimumDemand());
498		TestSubscriber s2 = new TestSubscriber();
499		p.subscribe(s2);
500		p.submit(1);
501		p.submit(2);
502		s2.awaitNext(1);
503	<	assertEquals(s1.nexts, 0);
503	>	assertEquals(0, s1.nexts);
504		s1.sn.request(3);
505		p.submit(3);
506		p.close();
507		s2.awaitComplete();
508	<	assertEquals(s2.nexts, 3);
509	<	assertEquals(s2.completes, 1);
508	>	assertEquals(3, s2.nexts);
509	>	assertEquals(1, s2.completes);
510		s1.awaitComplete();
511		assertTrue(s1.nexts > 0);
512	<	assertEquals(s1.completes, 1);
512	>	assertEquals(1, s1.completes);
513		}
514
515		/**
#	Line 543 | Line 528 | public class SubmissionPublisherTest ext
528		p.submit(2);
529		p.close();
530		s2.awaitComplete();
531	<	assertEquals(s2.nexts, 2);
532	<	assertEquals(s2.completes, 1);
531	>	assertEquals(2, s2.nexts);
532	>	assertEquals(1, s2.completes);
533		s1.awaitNext(1);
534	<	assertEquals(s1.nexts, 1);
534	>	assertEquals(1, s1.nexts);
535		}
536
537		/**
538	<	* Negative request causes error
538	>	* Non-positive request causes error
539		*/
540		public void testRequest3() {
541		SubmissionPublisher<Integer> p = basicPublisher();
542		TestSubscriber s1 = new TestSubscriber();
543		TestSubscriber s2 = new TestSubscriber();
544	+	TestSubscriber s3 = new TestSubscriber();
545		p.subscribe(s1);
546		p.subscribe(s2);
547	+	p.subscribe(s3);
548	+	s3.awaitSubscribe();
549		s2.awaitSubscribe();
550		s1.awaitSubscribe();
551		s1.sn.request(-1L);
552	+	s3.sn.request(0L);
553		p.submit(1);
554		p.submit(2);
555		p.close();
556		s2.awaitComplete();
557	<	assertEquals(s2.nexts, 2);
558	<	assertEquals(s2.completes, 1);
557	>	assertEquals(2, s2.nexts);
558	>	assertEquals(1, s2.completes);
559		s1.awaitError();
560	<	assertEquals(s1.errors, 1);
560	>	assertEquals(1, s1.errors);
561		assertTrue(s1.lastError instanceof IllegalArgumentException);
562	+	s3.awaitError();
563	+	assertEquals(1, s3.errors);
564	+	assertTrue(s3.lastError instanceof IllegalArgumentException);
565		}
566
567		/**
568		* estimateMinimumDemand reports 0 until request, nonzero after
569	<	* request, and zero again after delivery
569	>	* request
570		*/
571		public void testEstimateMinimumDemand() {
572		TestSubscriber s = new TestSubscriber();
#	Line 582 | Line 574 | public class SubmissionPublisherTest ext
574		s.request = false;
575		p.subscribe(s);
576		s.awaitSubscribe();
577	<	assertEquals(p.estimateMinimumDemand(), 0);
577	>	assertEquals(0, p.estimateMinimumDemand());
578		s.sn.request(1);
579	<	assertEquals(p.estimateMinimumDemand(), 1);
588	<	p.submit(1);
589	<	s.awaitNext(1);
590	<	assertEquals(p.estimateMinimumDemand(), 0);
579	>	assertEquals(1, p.estimateMinimumDemand());
580		}
581
582		/**
#	Line 595 | Line 584 | public class SubmissionPublisherTest ext
584		*/
585		public void testEmptySubmit() {
586		SubmissionPublisher<Integer> p = basicPublisher();
587	<	assertEquals(p.submit(1), 0);
587	>	assertEquals(0, p.submit(1));
588		}
589
590		/**
#	Line 623 | Line 612 | public class SubmissionPublisherTest ext
612		p.subscribe(s2);
613		s2.awaitSubscribe();
614		s1.awaitSubscribe();
615	<	assertEquals(p.submit(1), 1);
615	>	assertEquals(1, p.submit(1));
616		assertTrue(p.estimateMaximumLag() >= 1);
617		assertTrue(p.submit(2) >= 2);
618		assertTrue(p.estimateMaximumLag() >= 2);
#	Line 634 | Line 623 | public class SubmissionPublisherTest ext
623		p.submit(4);
624		p.close();
625		s2.awaitComplete();
626	<	assertEquals(s2.nexts, 4);
626	>	assertEquals(4, s2.nexts);
627		s1.awaitComplete();
628	<	assertEquals(s2.nexts, 4);
628	>	assertEquals(4, s2.nexts);
629		}
630
631		/**
632		* submit eventually issues requested items when buffer capacity is 1
633		*/
634		public void testCap1Submit() {
635	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
636	<	basicExecutor, 1);
635	>	SubmissionPublisher<Integer> p
636	>	= new SubmissionPublisher<>(basicExecutor, 1);
637		TestSubscriber s1 = new TestSubscriber();
638		TestSubscriber s2 = new TestSubscriber();
639		p.subscribe(s1);
640		p.subscribe(s2);
641		for (int i = 1; i <= 20; ++i) {
653	–	assertTrue(p.estimateMinimumDemand() <= 1);
642		assertTrue(p.submit(i) >= 0);
643		}
644		p.close();
645		s2.awaitComplete();
646		s1.awaitComplete();
647	<	assertEquals(s2.nexts, 20);
648	<	assertEquals(s2.completes, 1);
649	<	assertEquals(s1.nexts, 20);
650	<	assertEquals(s1.completes, 1);
647	>	assertEquals(20, s2.nexts);
648	>	assertEquals(1, s2.completes);
649	>	assertEquals(20, s1.nexts);
650	>	assertEquals(1, s1.completes);
651		}
652
653		static boolean noopHandle(AtomicInteger count) {
#	Line 713 | Line 701 | public class SubmissionPublisherTest ext
701		p.offer(4, null);
702		p.close();
703		s2.awaitComplete();
704	<	assertEquals(s2.nexts, 4);
704	>	assertEquals(4, s2.nexts);
705		s1.awaitComplete();
706	<	assertEquals(s2.nexts, 4);
706	>	assertEquals(4, s2.nexts);
707		}
708
709		/**
710		* offer reports drops if saturated
711		*/
712		public void testDroppedOffer() {
713	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
714	<	basicExecutor, 4);
713	>	SubmissionPublisher<Integer> p
714	>	= new SubmissionPublisher<>(basicExecutor, 4);
715		TestSubscriber s1 = new TestSubscriber();
716		s1.request = false;
717		TestSubscriber s2 = new TestSubscriber();
#	Line 751 | Line 739 | public class SubmissionPublisherTest ext
739		*/
740		public void testHandledDroppedOffer() {
741		AtomicInteger calls = new AtomicInteger();
742	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
743	<	basicExecutor, 4);
742	>	SubmissionPublisher<Integer> p
743	>	= new SubmissionPublisher<>(basicExecutor, 4);
744		TestSubscriber s1 = new TestSubscriber();
745		s1.request = false;
746		TestSubscriber s2 = new TestSubscriber();
#	Line 774 | Line 762 | public class SubmissionPublisherTest ext
762		assertTrue(calls.get() >= 4);
763		}
764
777	–
765		/**
766		* offer succeeds if drop handler forces request
767		*/
768		public void testRecoveredHandledDroppedOffer() {
769		AtomicInteger calls = new AtomicInteger();
770	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
771	<	basicExecutor, 4);
770	>	SubmissionPublisher<Integer> p
771	>	= new SubmissionPublisher<>(basicExecutor, 4);
772		TestSubscriber s1 = new TestSubscriber();
773		s1.request = false;
774		TestSubscriber s2 = new TestSubscriber();
#	Line 798 | Line 785 | public class SubmissionPublisherTest ext
785		p.close();
786		s2.awaitComplete();
787		s1.awaitComplete();
788	<	assertEquals(s1.nexts + s2.nexts, n);
788	>	assertEquals(n, s1.nexts + s2.nexts);
789		assertTrue(calls.get() >= 2);
790		}
791
805	–
792		/**
793		* Timed offer to a publisher with no subscribers returns lag 0
794		*/
795		public void testEmptyTimedOffer() {
796		SubmissionPublisher<Integer> p = basicPublisher();
797	+	long startTime = System.nanoTime();
798		assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
799	+	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
800		}
801
802		/**
#	Line 816 | Line 804 | public class SubmissionPublisherTest ext
804		*/
805		public void testNullTimedOffer() {
806		SubmissionPublisher<Integer> p = basicPublisher();
807	+	long startTime = System.nanoTime();
808		try {
809	<	p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
809	>	p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
810		shouldThrow();
811		} catch (NullPointerException success) {}
812		try {
813	<	p.offer(1, SHORT_DELAY_MS, null, null);
813	>	p.offer(1, LONG_DELAY_MS, null, null);
814		shouldThrow();
815		} catch (NullPointerException success) {}
816	+	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
817		}
818
819		/**
#	Line 839 | Line 829 | public class SubmissionPublisherTest ext
829		p.subscribe(s2);
830		s2.awaitSubscribe();
831		s1.awaitSubscribe();
832	<	assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
833	<	assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
832	>	long startTime = System.nanoTime();
833	>	assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
834	>	assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
835		s1.sn.request(4);
836	<	assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
836	>	assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
837		s2.sn.request(4);
838	<	p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
838	>	p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
839		p.close();
840		s2.awaitComplete();
841	<	assertEquals(s2.nexts, 4);
841	>	assertEquals(4, s2.nexts);
842		s1.awaitComplete();
843	<	assertEquals(s2.nexts, 4);
843	>	assertEquals(4, s2.nexts);
844	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
845		}
846
847		/**
848		* Timed offer reports drops if saturated
849		*/
850		public void testDroppedTimedOffer() {
851	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
852	<	basicExecutor, 4);
851	>	SubmissionPublisher<Integer> p
852	>	= new SubmissionPublisher<>(basicExecutor, 4);
853		TestSubscriber s1 = new TestSubscriber();
854		s1.request = false;
855		TestSubscriber s2 = new TestSubscriber();
#	Line 866 | Line 858 | public class SubmissionPublisherTest ext
858		p.subscribe(s2);
859		s2.awaitSubscribe();
860		s1.awaitSubscribe();
861	+	long delay = timeoutMillis();
862		for (int i = 1; i <= 4; ++i)
863	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
864	<	p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
865	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
863	>	assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
864	>	long startTime = System.nanoTime();
865	>	assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
866		s1.sn.request(64);
867	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
867	>	assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
868	>	// 2 * delay should elapse but check only 1 * delay to allow timer slop
869	>	assertTrue(millisElapsedSince(startTime) >= delay);
870		s2.sn.request(64);
871		p.close();
872		s2.awaitComplete();
#	Line 885 | Line 880 | public class SubmissionPublisherTest ext
880		*/
881		public void testHandledDroppedTimedOffer() {
882		AtomicInteger calls = new AtomicInteger();
883	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
884	<	basicExecutor, 4);
883	>	SubmissionPublisher<Integer> p
884	>	= new SubmissionPublisher<>(basicExecutor, 4);
885		TestSubscriber s1 = new TestSubscriber();
886		s1.request = false;
887		TestSubscriber s2 = new TestSubscriber();
#	Line 895 | Line 890 | public class SubmissionPublisherTest ext
890		p.subscribe(s2);
891		s2.awaitSubscribe();
892		s1.awaitSubscribe();
893	+	long delay = timeoutMillis();
894		for (int i = 1; i <= 4; ++i)
895	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
896	<	p.offer(5, (s, x) -> noopHandle(calls));
897	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
895	>	assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
896	>	long startTime = System.nanoTime();
897	>	assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
898		s1.sn.request(64);
899	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
899	>	assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
900	>	assertTrue(millisElapsedSince(startTime) >= delay);
901		s2.sn.request(64);
902		p.close();
903		s2.awaitComplete();
#	Line 913 | Line 910 | public class SubmissionPublisherTest ext
910		*/
911		public void testRecoveredHandledDroppedTimedOffer() {
912		AtomicInteger calls = new AtomicInteger();
913	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
914	<	basicExecutor, 4);
913	>	SubmissionPublisher<Integer> p
914	>	= new SubmissionPublisher<>(basicExecutor, 4);
915		TestSubscriber s1 = new TestSubscriber();
916		s1.request = false;
917		TestSubscriber s2 = new TestSubscriber();
#	Line 924 | Line 921 | public class SubmissionPublisherTest ext
921		s2.awaitSubscribe();
922		s1.awaitSubscribe();
923		int n = 0;
924	<	for (int i = 1; i <= 8; ++i) {
925	<	int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
924	>	long delay = timeoutMillis();
925	>	long startTime = System.nanoTime();
926	>	for (int i = 1; i <= 6; ++i) {
927	>	int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
928		n = n + 2 + (d < 0 ? d : 0);
929		}
930	+	assertTrue(millisElapsedSince(startTime) >= delay);
931		p.close();
932		s2.awaitComplete();
933		s1.awaitComplete();
934	<	assertEquals(s1.nexts + s2.nexts, n);
934	>	assertEquals(n, s1.nexts + s2.nexts);
935		assertTrue(calls.get() >= 2);
936		}
937
938	+	/**
939	+	* consume returns a CompletableFuture that is done when
940	+	* publisher completes
941	+	*/
942	+	public void testConsume() {
943	+	AtomicInteger sum = new AtomicInteger();
944	+	SubmissionPublisher<Integer> p = basicPublisher();
945	+	CompletableFuture<Void> f =
946	+	p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
947	+	int n = 20;
948	+	for (int i = 1; i <= n; ++i)
949	+	p.submit(i);
950	+	p.close();
951	+	f.join();
952	+	assertEquals((n * (n + 1)) / 2, sum.get());
953	+	}
954
955	+	/**
956	+	* consume(null) throws NPE
957	+	*/
958	+	public void testConsumeNPE() {
959	+	SubmissionPublisher<Integer> p = basicPublisher();
960	+	try {
961	+	CompletableFuture<Void> f = p.consume(null);
962	+	shouldThrow();
963	+	} catch (NullPointerException success) {}
964	+	}
965	+
966	+	/**
967	+	* consume eventually stops processing published items if cancelled
968	+	*/
969	+	public void testCancelledConsume() {
970	+	AtomicInteger count = new AtomicInteger();
971	+	SubmissionPublisher<Integer> p = basicPublisher();
972	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
973	+	f.cancel(true);
974	+	int n = 1000000; // arbitrary limit
975	+	for (int i = 1; i <= n; ++i)
976	+	p.submit(i);
977	+	assertTrue(count.get() < n);
978	+	}
979	+
980	+	/**
981	+	* Tests scenario for
982	+	* JDK-8187947: A race condition in SubmissionPublisher
983	+	* cvs update -D '2017-11-25' src/main/java/util/concurrent/SubmissionPublisher.java && ant -Djsr166.expensiveTests=true -Djsr166.tckTestClass=SubmissionPublisherTest -Djsr166.methodFilter=testMissedSignal tck; cvs update -A src/main/java/util/concurrent/SubmissionPublisher.java
984	+	*/
985	+	public void testMissedSignal_8187947() throws Exception {
986	+	if (!atLeastJava9()) return; // backport to jdk8 too hard
987	+	final int N = expensiveTests ? (1 << 20) : (1 << 10);
988	+	final CountDownLatch finished = new CountDownLatch(1);
989	+	final SubmissionPublisher<Boolean> pub = new SubmissionPublisher<>();
990	+	class Sub implements Subscriber<Boolean> {
991	+	int received;
992	+	public void onSubscribe(Subscription s) {
993	+	s.request(N);
994	+	}
995	+	public void onNext(Boolean item) {
996	+	if (++received == N)
997	+	finished.countDown();
998	+	else
999	+	CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1000	+	}
1001	+	public void onError(Throwable t) { throw new AssertionError(t); }
1002	+	public void onComplete() {}
1003	+	}
1004	+	pub.subscribe(new Sub());
1005	+	CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1006	+	await(finished);
1007	+	}
1008		}

Diff Legend

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