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.2 by jsr166, Mon Sep 7 20:20:40 2015 UTC vs.
Revision 1.24 by jsr166, Mon Nov 27 01:19:51 2017 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 354 | Line 338 | public class SubmissionPublisherTest ext
338		p.subscribe(s);
339		} catch (Exception ok) {}
340		s.awaitError();
341	<	assertEquals(s.nexts, 0);
342	<	assertEquals(s.errors, 1);
343	<	assertEquals(s.completes, 0);
341	>	assertEquals(0, s.nexts);
342	>	assertEquals(1, s.errors);
343	>	assertEquals(0, s.completes);
344		}
345
346		/**
347	<	* subscribe(null) thows NPE
347	>	* subscribe(null) throws NPE
348		*/
349		public void testSubscribe6() {
350		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 385 | Line 369 | public class SubmissionPublisherTest ext
369		assertTrue(p.isClosed());
370		assertNull(p.getClosedException());
371		s1.awaitComplete();
372	<	assertEquals(s1.nexts, 1);
373	<	assertEquals(s1.completes, 1);
372	>	assertEquals(1, s1.nexts);
373	>	assertEquals(1, s1.completes);
374		s2.awaitComplete();
375	<	assertEquals(s2.nexts, 1);
376	<	assertEquals(s2.completes, 1);
375	>	assertEquals(1, s2.nexts);
376	>	assertEquals(1, s2.completes);
377		}
378
379		/**
380		* Closing a publisher exceptionally causes onError to subscribers
381	+	* after they are subscribed
382		*/
383		public void testCloseExceptionallyError() {
384		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 404 | Line 389 | public class SubmissionPublisherTest ext
389		p.submit(1);
390		p.closeExceptionally(new SPException());
391		assertTrue(p.isClosed());
392	+	s1.awaitSubscribe();
393		s1.awaitError();
394		assertTrue(s1.nexts <= 1);
395	<	assertEquals(s1.errors, 1);
395	>	assertEquals(1, s1.errors);
396	>	s2.awaitSubscribe();
397		s2.awaitError();
398		assertTrue(s2.nexts <= 1);
399	<	assertEquals(s2.errors, 1);
399	>	assertEquals(1, s2.errors);
400		}
401
402		/**
403		* Cancelling a subscription eventually causes no more onNexts to be issued
404		*/
405		public void testCancel() {
406	<	SubmissionPublisher<Integer> p = basicPublisher();
406	>	SubmissionPublisher<Integer> p =
407	>	new SubmissionPublisher<Integer>(basicExecutor, 4); // must be < 20
408		TestSubscriber s1 = new TestSubscriber();
409		TestSubscriber s2 = new TestSubscriber();
410		p.subscribe(s1);
#	Line 428 | Line 416 | public class SubmissionPublisherTest ext
416		p.submit(i);
417		p.close();
418		s2.awaitComplete();
419	<	assertEquals(s2.nexts, 20);
420	<	assertEquals(s2.completes, 1);
419	>	assertEquals(20, s2.nexts);
420	>	assertEquals(1, s2.completes);
421		assertTrue(s1.nexts < 20);
422		assertFalse(p.isSubscribed(s1));
423		}
#	Line 449 | Line 437 | public class SubmissionPublisherTest ext
437		p.submit(2);
438		p.close();
439		s2.awaitComplete();
440	<	assertEquals(s2.nexts, 2);
440	>	assertEquals(2, s2.nexts);
441		s1.awaitComplete();
442	<	assertEquals(s1.errors, 1);
442	>	assertEquals(1, s1.errors);
443		}
444
445		/**
446	<	* If a handler is supplied in conctructor, it is invoked when
446	>	* If a handler is supplied in constructor, it is invoked when
447		* subscriber throws an exception in onNext
448		*/
449		public void testThrowOnNextHandler() {
450		AtomicInteger calls = new AtomicInteger();
451	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
452	<	(basicExecutor, 8,
465	<	(s, e) -> calls.getAndIncrement());
451	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
452	>	basicExecutor, 8, (s, e) -> calls.getAndIncrement());
453		TestSubscriber s1 = new TestSubscriber();
454		TestSubscriber s2 = new TestSubscriber();
455		p.subscribe(s1);
#	Line 473 | Line 460 | public class SubmissionPublisherTest ext
460		p.submit(2);
461		p.close();
462		s2.awaitComplete();
463	<	assertEquals(s2.nexts, 2);
464	<	assertEquals(s2.completes, 1);
463	>	assertEquals(2, s2.nexts);
464	>	assertEquals(1, s2.completes);
465		s1.awaitError();
466	<	assertEquals(s1.errors, 1);
467	<	assertEquals(calls.get(), 1);
466	>	assertEquals(1, s1.errors);
467	>	assertEquals(1, calls.get());
468		}
469
470		/**
#	Line 494 | Line 481 | public class SubmissionPublisherTest ext
481		p.close();
482		s2.awaitComplete();
483		s1.awaitComplete();
484	<	assertEquals(s2.nexts, 20);
485	<	assertEquals(s2.completes, 1);
486	<	assertEquals(s1.nexts, 20);
487	<	assertEquals(s1.completes, 1);
484	>	assertEquals(20, s2.nexts);
485	>	assertEquals(1, s2.completes);
486	>	assertEquals(20, s1.nexts);
487	>	assertEquals(1, s1.completes);
488		}
489
490		/**
#	Line 509 | Line 496 | public class SubmissionPublisherTest ext
496		s1.request = false;
497		p.subscribe(s1);
498		s1.awaitSubscribe();
499	<	assertTrue(p.estimateMinimumDemand() == 0);
499	>	assertEquals(0, p.estimateMinimumDemand());
500		TestSubscriber s2 = new TestSubscriber();
501		p.subscribe(s2);
502		p.submit(1);
503		p.submit(2);
504		s2.awaitNext(1);
505	<	assertEquals(s1.nexts, 0);
505	>	assertEquals(0, s1.nexts);
506		s1.sn.request(3);
507		p.submit(3);
508		p.close();
509		s2.awaitComplete();
510	<	assertEquals(s2.nexts, 3);
511	<	assertEquals(s2.completes, 1);
510	>	assertEquals(3, s2.nexts);
511	>	assertEquals(1, s2.completes);
512		s1.awaitComplete();
513		assertTrue(s1.nexts > 0);
514	<	assertEquals(s1.completes, 1);
514	>	assertEquals(1, s1.completes);
515		}
516
517		/**
#	Line 543 | Line 530 | public class SubmissionPublisherTest ext
530		p.submit(2);
531		p.close();
532		s2.awaitComplete();
533	<	assertEquals(s2.nexts, 2);
534	<	assertEquals(s2.completes, 1);
533	>	assertEquals(2, s2.nexts);
534	>	assertEquals(1, s2.completes);
535		s1.awaitNext(1);
536	<	assertEquals(s1.nexts, 1);
536	>	assertEquals(1, s1.nexts);
537		}
538
539		/**
540	<	* Negative request causes error
540	>	* Non-positive request causes error
541		*/
542		public void testRequest3() {
543		SubmissionPublisher<Integer> p = basicPublisher();
544		TestSubscriber s1 = new TestSubscriber();
545		TestSubscriber s2 = new TestSubscriber();
546	+	TestSubscriber s3 = new TestSubscriber();
547		p.subscribe(s1);
548		p.subscribe(s2);
549	+	p.subscribe(s3);
550	+	s3.awaitSubscribe();
551		s2.awaitSubscribe();
552		s1.awaitSubscribe();
553		s1.sn.request(-1L);
554	+	s3.sn.request(0L);
555		p.submit(1);
556		p.submit(2);
557		p.close();
558		s2.awaitComplete();
559	<	assertEquals(s2.nexts, 2);
560	<	assertEquals(s2.completes, 1);
559	>	assertEquals(2, s2.nexts);
560	>	assertEquals(1, s2.completes);
561		s1.awaitError();
562	<	assertEquals(s1.errors, 1);
562	>	assertEquals(1, s1.errors);
563		assertTrue(s1.lastError instanceof IllegalArgumentException);
564	+	s3.awaitError();
565	+	assertEquals(1, s3.errors);
566	+	assertTrue(s3.lastError instanceof IllegalArgumentException);
567		}
568
569		/**
570		* estimateMinimumDemand reports 0 until request, nonzero after
571	<	* request, and zero again after delivery
571	>	* request
572		*/
573		public void testEstimateMinimumDemand() {
574		TestSubscriber s = new TestSubscriber();
#	Line 582 | Line 576 | public class SubmissionPublisherTest ext
576		s.request = false;
577		p.subscribe(s);
578		s.awaitSubscribe();
579	<	assertEquals(p.estimateMinimumDemand(), 0);
579	>	assertEquals(0, p.estimateMinimumDemand());
580		s.sn.request(1);
581	<	assertEquals(p.estimateMinimumDemand(), 1);
588	<	p.submit(1);
589	<	s.awaitNext(1);
590	<	assertEquals(p.estimateMinimumDemand(), 0);
581	>	assertEquals(1, p.estimateMinimumDemand());
582		}
583
584		/**
585	<	* Submit to a publisher with no subscribers returns lag 0
585	>	* submit to a publisher with no subscribers returns lag 0
586		*/
587		public void testEmptySubmit() {
588		SubmissionPublisher<Integer> p = basicPublisher();
589	<	assertEquals(p.submit(1), 0);
589	>	assertEquals(0, p.submit(1));
590		}
591
592		/**
593	<	* Submit(null) throws NPE
593	>	* submit(null) throws NPE
594		*/
595		public void testNullSubmit() {
596		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 610 | Line 601 | public class SubmissionPublisherTest ext
601		}
602
603		/**
604	<	* Submit returns number of lagged items, compatible with result
604	>	* submit returns number of lagged items, compatible with result
605		* of estimateMaximumLag.
606		*/
607		public void testLaggedSubmit() {
#	Line 623 | Line 614 | public class SubmissionPublisherTest ext
614		p.subscribe(s2);
615		s2.awaitSubscribe();
616		s1.awaitSubscribe();
617	<	assertEquals(p.submit(1), 1);
617	>	assertEquals(1, p.submit(1));
618		assertTrue(p.estimateMaximumLag() >= 1);
619		assertTrue(p.submit(2) >= 2);
620		assertTrue(p.estimateMaximumLag() >= 2);
#	Line 634 | Line 625 | public class SubmissionPublisherTest ext
625		p.submit(4);
626		p.close();
627		s2.awaitComplete();
628	<	assertEquals(s2.nexts, 4);
628	>	assertEquals(4, s2.nexts);
629		s1.awaitComplete();
630	<	assertEquals(s2.nexts, 4);
630	>	assertEquals(4, s2.nexts);
631		}
632
633		/**
634		* submit eventually issues requested items when buffer capacity is 1
635		*/
636		public void testCap1Submit() {
637	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
638	<	basicExecutor, 1);
637	>	SubmissionPublisher<Integer> p
638	>	= new SubmissionPublisher<>(basicExecutor, 1);
639		TestSubscriber s1 = new TestSubscriber();
640		TestSubscriber s2 = new TestSubscriber();
641		p.subscribe(s1);
642		p.subscribe(s2);
643		for (int i = 1; i <= 20; ++i) {
653	–	assertTrue(p.estimateMinimumDemand() <= 1);
644		assertTrue(p.submit(i) >= 0);
645		}
646		p.close();
647		s2.awaitComplete();
648		s1.awaitComplete();
649	<	assertEquals(s2.nexts, 20);
650	<	assertEquals(s2.completes, 1);
651	<	assertEquals(s1.nexts, 20);
652	<	assertEquals(s1.completes, 1);
649	>	assertEquals(20, s2.nexts);
650	>	assertEquals(1, s2.completes);
651	>	assertEquals(20, s1.nexts);
652	>	assertEquals(1, s1.completes);
653		}
654
655		static boolean noopHandle(AtomicInteger count) {
#	Line 674 | Line 664 | public class SubmissionPublisherTest ext
664		}
665
666		/**
667	<	* Offer to a publisher with no subscribers returns lag 0
667	>	* offer to a publisher with no subscribers returns lag 0
668		*/
669		public void testEmptyOffer() {
670		SubmissionPublisher<Integer> p = basicPublisher();
671	<	assertEquals(p.offer(1, null), 0);
671	>	assertEquals(0, p.offer(1, null));
672		}
673
674		/**
675	<	* Offer(null) throws NPE
675	>	* offer(null) throws NPE
676		*/
677		public void testNullOffer() {
678		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 693 | Line 683 | public class SubmissionPublisherTest ext
683		}
684
685		/**
686	<	* Offer returns number of lagged items if not saturated
686	>	* offer returns number of lagged items if not saturated
687		*/
688		public void testLaggedOffer() {
689		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 713 | Line 703 | public class SubmissionPublisherTest ext
703		p.offer(4, null);
704		p.close();
705		s2.awaitComplete();
706	<	assertEquals(s2.nexts, 4);
706	>	assertEquals(4, s2.nexts);
707		s1.awaitComplete();
708	<	assertEquals(s2.nexts, 4);
708	>	assertEquals(4, s2.nexts);
709		}
710
711		/**
712	<	* Offer reports drops if saturated
712	>	* offer reports drops if saturated
713		*/
714		public void testDroppedOffer() {
715	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
716	<	basicExecutor, 4);
715	>	SubmissionPublisher<Integer> p
716	>	= new SubmissionPublisher<>(basicExecutor, 4);
717		TestSubscriber s1 = new TestSubscriber();
718		s1.request = false;
719		TestSubscriber s2 = new TestSubscriber();
#	Line 747 | Line 737 | public class SubmissionPublisherTest ext
737		}
738
739		/**
740	<	* Offer invokes drop handler if saturated
740	>	* offer invokes drop handler if saturated
741		*/
742		public void testHandledDroppedOffer() {
743		AtomicInteger calls = new AtomicInteger();
744	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
745	<	basicExecutor, 4);
744	>	SubmissionPublisher<Integer> p
745	>	= new SubmissionPublisher<>(basicExecutor, 4);
746		TestSubscriber s1 = new TestSubscriber();
747		s1.request = false;
748		TestSubscriber s2 = new TestSubscriber();
#	Line 774 | Line 764 | public class SubmissionPublisherTest ext
764		assertTrue(calls.get() >= 4);
765		}
766
777	–
767		/**
768	<	* Offer succeeds if drop handler forces request
768	>	* offer succeeds if drop handler forces request
769		*/
770		public void testRecoveredHandledDroppedOffer() {
771		AtomicInteger calls = new AtomicInteger();
772	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
773	<	basicExecutor, 4);
772	>	SubmissionPublisher<Integer> p
773	>	= new SubmissionPublisher<>(basicExecutor, 4);
774		TestSubscriber s1 = new TestSubscriber();
775		s1.request = false;
776		TestSubscriber s2 = new TestSubscriber();
#	Line 798 | Line 787 | public class SubmissionPublisherTest ext
787		p.close();
788		s2.awaitComplete();
789		s1.awaitComplete();
790	<	assertEquals(s1.nexts + s2.nexts, n);
790	>	assertEquals(n, s1.nexts + s2.nexts);
791		assertTrue(calls.get() >= 2);
792		}
793
805	–
794		/**
795	<	* TimedOffer to a publisher with no subscribers returns lag 0
795	>	* Timed offer to a publisher with no subscribers returns lag 0
796		*/
797		public void testEmptyTimedOffer() {
798		SubmissionPublisher<Integer> p = basicPublisher();
799	<	assertEquals(p.offer(1, null), 0);
799	>	long startTime = System.nanoTime();
800	>	assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
801	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
802		}
803
804		/**
805	<	* Timed Offer with null item or TimeUnit throws NPE
805	>	* Timed offer with null item or TimeUnit throws NPE
806		*/
807		public void testNullTimedOffer() {
808		SubmissionPublisher<Integer> p = basicPublisher();
809	+	long startTime = System.nanoTime();
810		try {
811	<	p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
811	>	p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
812		shouldThrow();
813		} catch (NullPointerException success) {}
814		try {
815	<	p.offer(1, SHORT_DELAY_MS, null, null);
815	>	p.offer(1, LONG_DELAY_MS, null, null);
816		shouldThrow();
817		} catch (NullPointerException success) {}
818	+	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
819		}
820
821		/**
822	<	* Timed Offer returns number of lagged items if not saturated
822	>	* Timed offer returns number of lagged items if not saturated
823		*/
824		public void testLaggedTimedOffer() {
825		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 839 | Line 831 | public class SubmissionPublisherTest ext
831		p.subscribe(s2);
832		s2.awaitSubscribe();
833		s1.awaitSubscribe();
834	<	assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
835	<	assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
834	>	long startTime = System.nanoTime();
835	>	assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
836	>	assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
837		s1.sn.request(4);
838	<	assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
838	>	assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
839		s2.sn.request(4);
840	<	p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
840	>	p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
841		p.close();
842		s2.awaitComplete();
843	<	assertEquals(s2.nexts, 4);
843	>	assertEquals(4, s2.nexts);
844		s1.awaitComplete();
845	<	assertEquals(s2.nexts, 4);
845	>	assertEquals(4, s2.nexts);
846	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
847		}
848
849		/**
850	<	* Timed Offer reports drops if saturated
850	>	* Timed offer reports drops if saturated
851		*/
852		public void testDroppedTimedOffer() {
853	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
854	<	basicExecutor, 4);
853	>	SubmissionPublisher<Integer> p
854	>	= new SubmissionPublisher<>(basicExecutor, 4);
855		TestSubscriber s1 = new TestSubscriber();
856		s1.request = false;
857		TestSubscriber s2 = new TestSubscriber();
#	Line 866 | Line 860 | public class SubmissionPublisherTest ext
860		p.subscribe(s2);
861		s2.awaitSubscribe();
862		s1.awaitSubscribe();
863	+	long delay = timeoutMillis();
864		for (int i = 1; i <= 4; ++i)
865	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
866	<	p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
867	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
865	>	assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
866	>	long startTime = System.nanoTime();
867	>	assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
868		s1.sn.request(64);
869	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
869	>	assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
870	>	// 2 * delay should elapse but check only 1 * delay to allow timer slop
871	>	assertTrue(millisElapsedSince(startTime) >= delay);
872		s2.sn.request(64);
873		p.close();
874		s2.awaitComplete();
#	Line 881 | Line 878 | public class SubmissionPublisherTest ext
878		}
879
880		/**
881	<	* Timed Offer invokes drop handler if saturated
881	>	* Timed offer invokes drop handler if saturated
882		*/
883		public void testHandledDroppedTimedOffer() {
884		AtomicInteger calls = new AtomicInteger();
885	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
886	<	basicExecutor, 4);
885	>	SubmissionPublisher<Integer> p
886	>	= new SubmissionPublisher<>(basicExecutor, 4);
887		TestSubscriber s1 = new TestSubscriber();
888		s1.request = false;
889		TestSubscriber s2 = new TestSubscriber();
#	Line 895 | Line 892 | public class SubmissionPublisherTest ext
892		p.subscribe(s2);
893		s2.awaitSubscribe();
894		s1.awaitSubscribe();
895	+	long delay = timeoutMillis();
896		for (int i = 1; i <= 4; ++i)
897	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
898	<	p.offer(5, (s, x) -> noopHandle(calls));
899	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
897	>	assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
898	>	long startTime = System.nanoTime();
899	>	assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
900		s1.sn.request(64);
901	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
901	>	assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
902	>	assertTrue(millisElapsedSince(startTime) >= delay);
903		s2.sn.request(64);
904		p.close();
905		s2.awaitComplete();
#	Line 909 | Line 908 | public class SubmissionPublisherTest ext
908		}
909
910		/**
911	<	* Timed Offer succeeds if drop handler forces request
911	>	* Timed offer succeeds if drop handler forces request
912		*/
913		public void testRecoveredHandledDroppedTimedOffer() {
914		AtomicInteger calls = new AtomicInteger();
915	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
916	<	basicExecutor, 4);
915	>	SubmissionPublisher<Integer> p
916	>	= new SubmissionPublisher<>(basicExecutor, 4);
917		TestSubscriber s1 = new TestSubscriber();
918		s1.request = false;
919		TestSubscriber s2 = new TestSubscriber();
#	Line 924 | Line 923 | public class SubmissionPublisherTest ext
923		s2.awaitSubscribe();
924		s1.awaitSubscribe();
925		int n = 0;
926	<	for (int i = 1; i <= 8; ++i) {
927	<	int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
926	>	long delay = timeoutMillis();
927	>	long startTime = System.nanoTime();
928	>	for (int i = 1; i <= 6; ++i) {
929	>	int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
930		n = n + 2 + (d < 0 ? d : 0);
931		}
932	+	assertTrue(millisElapsedSince(startTime) >= delay);
933		p.close();
934		s2.awaitComplete();
935		s1.awaitComplete();
936	<	assertEquals(s1.nexts + s2.nexts, n);
936	>	assertEquals(n, s1.nexts + s2.nexts);
937		assertTrue(calls.get() >= 2);
938		}
939
940	+	/**
941	+	* consume returns a CompletableFuture that is done when
942	+	* publisher completes
943	+	*/
944	+	public void testConsume() {
945	+	AtomicInteger sum = new AtomicInteger();
946	+	SubmissionPublisher<Integer> p = basicPublisher();
947	+	CompletableFuture<Void> f =
948	+	p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
949	+	int n = 20;
950	+	for (int i = 1; i <= n; ++i)
951	+	p.submit(i);
952	+	p.close();
953	+	f.join();
954	+	assertEquals((n * (n + 1)) / 2, sum.get());
955	+	}
956	+
957	+	/**
958	+	* consume(null) throws NPE
959	+	*/
960	+	public void testConsumeNPE() {
961	+	SubmissionPublisher<Integer> p = basicPublisher();
962	+	try {
963	+	CompletableFuture<Void> f = p.consume(null);
964	+	shouldThrow();
965	+	} catch (NullPointerException success) {}
966	+	}
967	+
968	+	/**
969	+	* consume eventually stops processing published items if cancelled
970	+	*/
971	+	public void testCancelledConsume() {
972	+	AtomicInteger count = new AtomicInteger();
973	+	SubmissionPublisher<Integer> p = basicPublisher();
974	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
975	+	f.cancel(true);
976	+	int n = 1000000; // arbitrary limit
977	+	for (int i = 1; i <= n; ++i)
978	+	p.submit(i);
979	+	assertTrue(count.get() < n);
980	+	}
981
982	+	/**
983	+	* Tests scenario for
984	+	* JDK-8187947: A race condition in SubmissionPublisher
985	+	* 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
986	+	*/
987	+	public void testMissedSignal_8187947() throws Exception {
988	+	final int N = expensiveTests ? (1 << 20) : (1 << 10);
989	+	final CountDownLatch finished = new CountDownLatch(1);
990	+	final SubmissionPublisher<Boolean> pub = new SubmissionPublisher<>();
991	+	class Sub implements Subscriber<Boolean> {
992	+	int received;
993	+	public void onSubscribe(Subscription s) {
994	+	s.request(N);
995	+	}
996	+	public void onNext(Boolean item) {
997	+	if (++received == N)
998	+	finished.countDown();
999	+	else
1000	+	CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1001	+	}
1002	+	public void onError(Throwable t) { throw new AssertionError(t); }
1003	+	public void onComplete() {}
1004	+	}
1005	+	pub.subscribe(new Sub());
1006	+	CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1007	+	await(finished);
1008	+	}
1009		}

Diff Legend

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