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.22 by dl, Sat Nov 4 12:49:58 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.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>();
149	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>();
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 182 | Line 162 | public class SubmissionPublisherTest ext
162		*/
163		public void testConstructor2() {
164		Executor e = Executors.newFixedThreadPool(1);
165	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, 8);
165	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>(e, 8);
166		checkInitialState(p);
167		assertSame(p.getExecutor(), e);
168	<	assertEquals(p.getMaxBufferCapacity(), 8);
168	>	assertEquals(8, p.getMaxBufferCapacity());
169		}
170
171		/**
172	<	* A null Executor argument to SubmissionPublisher constructor throws NPE
172	>	* A null Executor argument to SubmissionPublisher constructor
173	>	* throws NullPointerException
174		*/
175		public void testConstructor3() {
176		try {
#	Line 200 | Line 181 | public class SubmissionPublisherTest ext
181
182		/**
183		* A negative capacity argument to SubmissionPublisher constructor
184	<	* throws IAE
184	>	* throws IllegalArgumentException
185		*/
186		public void testConstructor4() {
187		Executor e = Executors.newFixedThreadPool(1);
#	Line 212 | Line 193 | public class SubmissionPublisherTest ext
193
194		/**
195		* A closed publisher reports isClosed with no closedException and
196	<	* throws ISE upon attempted submission; a subsequent close or
197	<	* closeExceptionally has no additional effect.
196	>	* throws IllegalStateException upon attempted submission; a
197	>	* subsequent close or closeExceptionally has no additional
198	>	* effect.
199		*/
200		public void testClose() {
201		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 233 | Line 215 | public class SubmissionPublisherTest ext
215
216		/**
217		* A publisher closedExceptionally reports isClosed with the
218	<	* closedException and throws ISE upon attempted submission; a
219	<	* subsequent close or closeExceptionally has no additional
220	<	* effect.
218	>	* closedException and throws IllegalStateException upon attempted
219	>	* submission; a subsequent close or closeExceptionally has no
220	>	* additional effect.
221		*/
222		public void testCloseExceptionally() {
223		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 264 | Line 246 | public class SubmissionPublisherTest ext
246		SubmissionPublisher<Integer> p = basicPublisher();
247		p.subscribe(s);
248		assertTrue(p.hasSubscribers());
249	<	assertEquals(p.getNumberOfSubscribers(), 1);
249	>	assertEquals(1, p.getNumberOfSubscribers());
250		assertTrue(p.getSubscribers().contains(s));
251		assertTrue(p.isSubscribed(s));
252		s.awaitSubscribe();
253		assertNotNull(s.sn);
254	<	assertEquals(s.nexts, 0);
255	<	assertEquals(s.errors, 0);
256	<	assertEquals(s.completes, 0);
254	>	assertEquals(0, s.nexts);
255	>	assertEquals(0, s.errors);
256	>	assertEquals(0, s.completes);
257		TestSubscriber s2 = new TestSubscriber();
258		p.subscribe(s2);
259		assertTrue(p.hasSubscribers());
260	<	assertEquals(p.getNumberOfSubscribers(), 2);
260	>	assertEquals(2, p.getNumberOfSubscribers());
261		assertTrue(p.getSubscribers().contains(s));
262		assertTrue(p.getSubscribers().contains(s2));
263		assertTrue(p.isSubscribed(s));
264		assertTrue(p.isSubscribed(s2));
265		s2.awaitSubscribe();
266		assertNotNull(s2.sn);
267	<	assertEquals(s2.nexts, 0);
268	<	assertEquals(s2.errors, 0);
269	<	assertEquals(s2.completes, 0);
267	>	assertEquals(0, s2.nexts);
268	>	assertEquals(0, s2.errors);
269	>	assertEquals(0, s2.completes);
270	>	p.close();
271		}
272
273		/**
#	Line 297 | Line 280 | public class SubmissionPublisherTest ext
280		p.close();
281		p.subscribe(s);
282		s.awaitComplete();
283	<	assertEquals(s.nexts, 0);
284	<	assertEquals(s.errors, 0);
285	<	assertEquals(s.completes, 1);
283	>	assertEquals(0, s.nexts);
284	>	assertEquals(0, s.errors);
285	>	assertEquals(1, s.completes, 1);
286		}
287
288		/**
#	Line 315 | Line 298 | public class SubmissionPublisherTest ext
298		assertSame(p.getClosedException(), ex);
299		p.subscribe(s);
300		s.awaitError();
301	<	assertEquals(s.nexts, 0);
302	<	assertEquals(s.errors, 1);
301	>	assertEquals(0, s.nexts);
302	>	assertEquals(1, s.errors);
303		}
304
305		/**
#	Line 328 | Line 311 | public class SubmissionPublisherTest ext
311		SubmissionPublisher<Integer> p = basicPublisher();
312		p.subscribe(s);
313		assertTrue(p.hasSubscribers());
314	<	assertEquals(p.getNumberOfSubscribers(), 1);
314	>	assertEquals(1, p.getNumberOfSubscribers());
315		assertTrue(p.getSubscribers().contains(s));
316		assertTrue(p.isSubscribed(s));
317		s.awaitSubscribe();
318		assertNotNull(s.sn);
319	<	assertEquals(s.nexts, 0);
320	<	assertEquals(s.errors, 0);
321	<	assertEquals(s.completes, 0);
319	>	assertEquals(0, s.nexts);
320	>	assertEquals(0, s.errors);
321	>	assertEquals(0, s.completes);
322		p.subscribe(s);
323		s.awaitError();
324	<	assertEquals(s.nexts, 0);
325	<	assertEquals(s.errors, 1);
324	>	assertEquals(0, s.nexts);
325	>	assertEquals(1, s.errors);
326		assertFalse(p.isSubscribed(s));
327		}
328
#	Line 354 | Line 337 | public class SubmissionPublisherTest ext
337		p.subscribe(s);
338		} catch (Exception ok) {}
339		s.awaitError();
340	<	assertEquals(s.nexts, 0);
341	<	assertEquals(s.errors, 1);
342	<	assertEquals(s.completes, 0);
340	>	assertEquals(0, s.nexts);
341	>	assertEquals(1, s.errors);
342	>	assertEquals(0, s.completes);
343		}
344
345		/**
346	<	* subscribe(null) thows NPE
346	>	* subscribe(null) throws NPE
347		*/
348		public void testSubscribe6() {
349		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 385 | Line 368 | public class SubmissionPublisherTest ext
368		assertTrue(p.isClosed());
369		assertNull(p.getClosedException());
370		s1.awaitComplete();
371	<	assertEquals(s1.nexts, 1);
372	<	assertEquals(s1.completes, 1);
371	>	assertEquals(1, s1.nexts);
372	>	assertEquals(1, s1.completes);
373		s2.awaitComplete();
374	<	assertEquals(s2.nexts, 1);
375	<	assertEquals(s2.completes, 1);
374	>	assertEquals(1, s2.nexts);
375	>	assertEquals(1, s2.completes);
376		}
377
378		/**
379		* Closing a publisher exceptionally causes onError to subscribers
380	+	* after they are subscribed
381		*/
382		public void testCloseExceptionallyError() {
383		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 404 | Line 388 | public class SubmissionPublisherTest ext
388		p.submit(1);
389		p.closeExceptionally(new SPException());
390		assertTrue(p.isClosed());
391	+	s1.awaitSubscribe();
392		s1.awaitError();
393		assertTrue(s1.nexts <= 1);
394	<	assertEquals(s1.errors, 1);
394	>	assertEquals(1, s1.errors);
395	>	s2.awaitSubscribe();
396		s2.awaitError();
397		assertTrue(s2.nexts <= 1);
398	<	assertEquals(s2.errors, 1);
398	>	assertEquals(1, s2.errors);
399		}
400
401		/**
#	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		/**
583	<	* Submit to a publisher with no subscribers returns lag 0
583	>	* submit to a publisher with no subscribers returns lag 0
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		/**
591	<	* Submit(null) throws NPE
591	>	* submit(null) throws NPE
592		*/
593		public void testNullSubmit() {
594		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 610 | Line 599 | public class SubmissionPublisherTest ext
599		}
600
601		/**
602	<	* Submit returns number of lagged items, compatible with result
602	>	* submit returns number of lagged items, compatible with result
603		* of estimateMaximumLag.
604		*/
605		public void testLaggedSubmit() {
#	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);
#	Line 656 | Line 645 | public class SubmissionPublisherTest ext
645		p.close();
646		s2.awaitComplete();
647		s1.awaitComplete();
648	<	assertEquals(s2.nexts, 20);
649	<	assertEquals(s2.completes, 1);
650	<	assertEquals(s1.nexts, 20);
651	<	assertEquals(s1.completes, 1);
648	>	assertEquals(20, s2.nexts);
649	>	assertEquals(1, s2.completes);
650	>	assertEquals(20, s1.nexts);
651	>	assertEquals(1, s1.completes);
652		}
653
654		static boolean noopHandle(AtomicInteger count) {
#	Line 674 | Line 663 | public class SubmissionPublisherTest ext
663		}
664
665		/**
666	<	* Offer to a publisher with no subscribers returns lag 0
666	>	* offer to a publisher with no subscribers returns lag 0
667		*/
668		public void testEmptyOffer() {
669		SubmissionPublisher<Integer> p = basicPublisher();
670	<	assertEquals(p.offer(1, null), 0);
670	>	assertEquals(0, p.offer(1, null));
671		}
672
673		/**
674	<	* Offer(null) throws NPE
674	>	* offer(null) throws NPE
675		*/
676		public void testNullOffer() {
677		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 693 | Line 682 | public class SubmissionPublisherTest ext
682		}
683
684		/**
685	<	* Offer returns number of lagged items if not saturated
685	>	* offer returns number of lagged items if not saturated
686		*/
687		public void testLaggedOffer() {
688		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 713 | Line 702 | public class SubmissionPublisherTest ext
702		p.offer(4, null);
703		p.close();
704		s2.awaitComplete();
705	<	assertEquals(s2.nexts, 4);
705	>	assertEquals(4, s2.nexts);
706		s1.awaitComplete();
707	<	assertEquals(s2.nexts, 4);
707	>	assertEquals(4, s2.nexts);
708		}
709
710		/**
711	<	* Offer reports drops if saturated
711	>	* offer reports drops if saturated
712		*/
713		public void testDroppedOffer() {
714	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
715	<	basicExecutor, 4);
714	>	SubmissionPublisher<Integer> p
715	>	= new SubmissionPublisher<>(basicExecutor, 4);
716		TestSubscriber s1 = new TestSubscriber();
717		s1.request = false;
718		TestSubscriber s2 = new TestSubscriber();
#	Line 747 | Line 736 | public class SubmissionPublisherTest ext
736		}
737
738		/**
739	<	* Offer invokes drop handler if saturated
739	>	* offer invokes drop handler if saturated
740		*/
741		public void testHandledDroppedOffer() {
742		AtomicInteger calls = new AtomicInteger();
743	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
744	<	basicExecutor, 4);
743	>	SubmissionPublisher<Integer> p
744	>	= new SubmissionPublisher<>(basicExecutor, 4);
745		TestSubscriber s1 = new TestSubscriber();
746		s1.request = false;
747		TestSubscriber s2 = new TestSubscriber();
#	Line 774 | Line 763 | public class SubmissionPublisherTest ext
763		assertTrue(calls.get() >= 4);
764		}
765
777	–
766		/**
767	<	* Offer succeeds if drop handler forces request
767	>	* offer succeeds if drop handler forces request
768		*/
769		public void testRecoveredHandledDroppedOffer() {
770		AtomicInteger calls = new AtomicInteger();
771	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
772	<	basicExecutor, 4);
771	>	SubmissionPublisher<Integer> p
772	>	= new SubmissionPublisher<>(basicExecutor, 4);
773		TestSubscriber s1 = new TestSubscriber();
774		s1.request = false;
775		TestSubscriber s2 = new TestSubscriber();
#	Line 798 | Line 786 | public class SubmissionPublisherTest ext
786		p.close();
787		s2.awaitComplete();
788		s1.awaitComplete();
789	<	assertEquals(s1.nexts + s2.nexts, n);
789	>	assertEquals(n, s1.nexts + s2.nexts);
790		assertTrue(calls.get() >= 2);
791		}
792
805	–
793		/**
794	<	* TimedOffer to a publisher with no subscribers returns lag 0
794	>	* Timed offer to a publisher with no subscribers returns lag 0
795		*/
796		public void testEmptyTimedOffer() {
797		SubmissionPublisher<Integer> p = basicPublisher();
798	<	assertEquals(p.offer(1, null), 0);
798	>	long startTime = System.nanoTime();
799	>	assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
800	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
801		}
802
803		/**
804	<	* Timed Offer with null item or TimeUnit throws NPE
804	>	* Timed offer with null item or TimeUnit throws NPE
805		*/
806		public void testNullTimedOffer() {
807		SubmissionPublisher<Integer> p = basicPublisher();
808	+	long startTime = System.nanoTime();
809		try {
810	<	p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
810	>	p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
811		shouldThrow();
812		} catch (NullPointerException success) {}
813		try {
814	<	p.offer(1, SHORT_DELAY_MS, null, null);
814	>	p.offer(1, LONG_DELAY_MS, null, null);
815		shouldThrow();
816		} catch (NullPointerException success) {}
817	+	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
818		}
819
820		/**
821	<	* Timed Offer returns number of lagged items if not saturated
821	>	* Timed offer returns number of lagged items if not saturated
822		*/
823		public void testLaggedTimedOffer() {
824		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 839 | Line 830 | public class SubmissionPublisherTest ext
830		p.subscribe(s2);
831		s2.awaitSubscribe();
832		s1.awaitSubscribe();
833	<	assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
834	<	assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
833	>	long startTime = System.nanoTime();
834	>	assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
835	>	assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
836		s1.sn.request(4);
837	<	assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
837	>	assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
838		s2.sn.request(4);
839	<	p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
839	>	p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
840		p.close();
841		s2.awaitComplete();
842	<	assertEquals(s2.nexts, 4);
842	>	assertEquals(4, s2.nexts);
843		s1.awaitComplete();
844	<	assertEquals(s2.nexts, 4);
844	>	assertEquals(4, s2.nexts);
845	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
846		}
847
848		/**
849	<	* Timed Offer reports drops if saturated
849	>	* Timed offer reports drops if saturated
850		*/
851		public void testDroppedTimedOffer() {
852	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
853	<	basicExecutor, 4);
852	>	SubmissionPublisher<Integer> p
853	>	= new SubmissionPublisher<>(basicExecutor, 4);
854		TestSubscriber s1 = new TestSubscriber();
855		s1.request = false;
856		TestSubscriber s2 = new TestSubscriber();
#	Line 866 | Line 859 | public class SubmissionPublisherTest ext
859		p.subscribe(s2);
860		s2.awaitSubscribe();
861		s1.awaitSubscribe();
862	+	long delay = timeoutMillis();
863		for (int i = 1; i <= 4; ++i)
864	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
865	<	p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
866	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
864	>	assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
865	>	long startTime = System.nanoTime();
866	>	assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
867		s1.sn.request(64);
868	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
868	>	assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
869	>	// 2 * delay should elapse but check only 1 * delay to allow timer slop
870	>	assertTrue(millisElapsedSince(startTime) >= delay);
871		s2.sn.request(64);
872		p.close();
873		s2.awaitComplete();
#	Line 881 | Line 877 | public class SubmissionPublisherTest ext
877		}
878
879		/**
880	<	* Timed Offer invokes drop handler if saturated
880	>	* Timed offer invokes drop handler if saturated
881		*/
882		public void testHandledDroppedTimedOffer() {
883		AtomicInteger calls = new AtomicInteger();
884	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
885	<	basicExecutor, 4);
884	>	SubmissionPublisher<Integer> p
885	>	= new SubmissionPublisher<>(basicExecutor, 4);
886		TestSubscriber s1 = new TestSubscriber();
887		s1.request = false;
888		TestSubscriber s2 = new TestSubscriber();
#	Line 895 | Line 891 | public class SubmissionPublisherTest ext
891		p.subscribe(s2);
892		s2.awaitSubscribe();
893		s1.awaitSubscribe();
894	+	long delay = timeoutMillis();
895		for (int i = 1; i <= 4; ++i)
896	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
897	<	p.offer(5, (s, x) -> noopHandle(calls));
898	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
896	>	assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
897	>	long startTime = System.nanoTime();
898	>	assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
899		s1.sn.request(64);
900	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
900	>	assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
901	>	assertTrue(millisElapsedSince(startTime) >= delay);
902		s2.sn.request(64);
903		p.close();
904		s2.awaitComplete();
#	Line 909 | Line 907 | public class SubmissionPublisherTest ext
907		}
908
909		/**
910	<	* Timed Offer succeeds if drop handler forces request
910	>	* Timed offer succeeds if drop handler forces request
911		*/
912		public void testRecoveredHandledDroppedTimedOffer() {
913		AtomicInteger calls = new AtomicInteger();
914	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
915	<	basicExecutor, 4);
914	>	SubmissionPublisher<Integer> p
915	>	= new SubmissionPublisher<>(basicExecutor, 4);
916		TestSubscriber s1 = new TestSubscriber();
917		s1.request = false;
918		TestSubscriber s2 = new TestSubscriber();
#	Line 924 | Line 922 | public class SubmissionPublisherTest ext
922		s2.awaitSubscribe();
923		s1.awaitSubscribe();
924		int n = 0;
925	<	for (int i = 1; i <= 8; ++i) {
926	<	int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
925	>	long delay = timeoutMillis();
926	>	long startTime = System.nanoTime();
927	>	for (int i = 1; i <= 6; ++i) {
928	>	int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
929		n = n + 2 + (d < 0 ? d : 0);
930		}
931	+	assertTrue(millisElapsedSince(startTime) >= delay);
932		p.close();
933		s2.awaitComplete();
934		s1.awaitComplete();
935	<	assertEquals(s1.nexts + s2.nexts, n);
935	>	assertEquals(n, s1.nexts + s2.nexts);
936		assertTrue(calls.get() >= 2);
937		}
938
939	+	/**
940	+	* consume returns a CompletableFuture that is done when
941	+	* publisher completes
942	+	*/
943	+	public void testConsume() {
944	+	AtomicInteger sum = new AtomicInteger();
945	+	SubmissionPublisher<Integer> p = basicPublisher();
946	+	CompletableFuture<Void> f =
947	+	p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
948	+	int n = 20;
949	+	for (int i = 1; i <= n; ++i)
950	+	p.submit(i);
951	+	p.close();
952	+	f.join();
953	+	assertEquals((n * (n + 1)) / 2, sum.get());
954	+	}
955	+
956	+	/**
957	+	* consume(null) throws NPE
958	+	*/
959	+	public void testConsumeNPE() {
960	+	SubmissionPublisher<Integer> p = basicPublisher();
961	+	try {
962	+	CompletableFuture<Void> f = p.consume(null);
963	+	shouldThrow();
964	+	} catch (NullPointerException success) {}
965	+	}
966	+
967	+	/**
968	+	* consume eventually stops processing published items if cancelled
969	+	*/
970	+	public void testCancelledConsume() {
971	+	AtomicInteger count = new AtomicInteger();
972	+	SubmissionPublisher<Integer> p = basicPublisher();
973	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
974	+	f.cancel(true);
975	+	int n = 1000000; // arbitrary limit
976	+	for (int i = 1; i <= n; ++i)
977	+	p.submit(i);
978	+	assertTrue(count.get() < n);
979	+	}
980
981		}

Diff Legend

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