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.19 by dl, Thu Mar 9 00:11:16 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		/**
#	Line 264 | Line 244 | public class SubmissionPublisherTest ext
244		SubmissionPublisher<Integer> p = basicPublisher();
245		p.subscribe(s);
246		assertTrue(p.hasSubscribers());
247	<	assertEquals(p.getNumberOfSubscribers(), 1);
247	>	assertEquals(1, p.getNumberOfSubscribers());
248		assertTrue(p.getSubscribers().contains(s));
249		assertTrue(p.isSubscribed(s));
250		s.awaitSubscribe();
251		assertNotNull(s.sn);
252	<	assertEquals(s.nexts, 0);
253	<	assertEquals(s.errors, 0);
254	<	assertEquals(s.completes, 0);
252	>	assertEquals(0, s.nexts);
253	>	assertEquals(0, s.errors);
254	>	assertEquals(0, s.completes);
255		TestSubscriber s2 = new TestSubscriber();
256		p.subscribe(s2);
257		assertTrue(p.hasSubscribers());
258	<	assertEquals(p.getNumberOfSubscribers(), 2);
258	>	assertEquals(2, p.getNumberOfSubscribers());
259		assertTrue(p.getSubscribers().contains(s));
260		assertTrue(p.getSubscribers().contains(s2));
261		assertTrue(p.isSubscribed(s));
262		assertTrue(p.isSubscribed(s2));
263		s2.awaitSubscribe();
264		assertNotNull(s2.sn);
265	<	assertEquals(s2.nexts, 0);
266	<	assertEquals(s2.errors, 0);
267	<	assertEquals(s2.completes, 0);
265	>	assertEquals(0, s2.nexts);
266	>	assertEquals(0, s2.errors);
267	>	assertEquals(0, s2.completes);
268	>	p.close();
269		}
270
271		/**
#	Line 297 | Line 278 | public class SubmissionPublisherTest ext
278		p.close();
279		p.subscribe(s);
280		s.awaitComplete();
281	<	assertEquals(s.nexts, 0);
282	<	assertEquals(s.errors, 0);
283	<	assertEquals(s.completes, 1);
281	>	assertEquals(0, s.nexts);
282	>	assertEquals(0, s.errors);
283	>	assertEquals(1, s.completes, 1);
284		}
285
286		/**
#	Line 315 | Line 296 | public class SubmissionPublisherTest ext
296		assertSame(p.getClosedException(), ex);
297		p.subscribe(s);
298		s.awaitError();
299	<	assertEquals(s.nexts, 0);
300	<	assertEquals(s.errors, 1);
299	>	assertEquals(0, s.nexts);
300	>	assertEquals(1, s.errors);
301		}
302
303		/**
#	Line 328 | Line 309 | public class SubmissionPublisherTest ext
309		SubmissionPublisher<Integer> p = basicPublisher();
310		p.subscribe(s);
311		assertTrue(p.hasSubscribers());
312	<	assertEquals(p.getNumberOfSubscribers(), 1);
312	>	assertEquals(1, p.getNumberOfSubscribers());
313		assertTrue(p.getSubscribers().contains(s));
314		assertTrue(p.isSubscribed(s));
315		s.awaitSubscribe();
316		assertNotNull(s.sn);
317	<	assertEquals(s.nexts, 0);
318	<	assertEquals(s.errors, 0);
319	<	assertEquals(s.completes, 0);
317	>	assertEquals(0, s.nexts);
318	>	assertEquals(0, s.errors);
319	>	assertEquals(0, s.completes);
320		p.subscribe(s);
321		s.awaitError();
322	<	assertEquals(s.nexts, 0);
323	<	assertEquals(s.errors, 1);
322	>	assertEquals(0, s.nexts);
323	>	assertEquals(1, s.errors);
324		assertFalse(p.isSubscribed(s));
325		}
326
#	Line 354 | Line 335 | public class SubmissionPublisherTest ext
335		p.subscribe(s);
336		} catch (Exception ok) {}
337		s.awaitError();
338	<	assertEquals(s.nexts, 0);
339	<	assertEquals(s.errors, 1);
340	<	assertEquals(s.completes, 0);
338	>	assertEquals(0, s.nexts);
339	>	assertEquals(1, s.errors);
340	>	assertEquals(0, s.completes);
341		}
342
343		/**
344	<	* subscribe(null) thows NPE
344	>	* subscribe(null) throws NPE
345		*/
346		public void testSubscribe6() {
347		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 385 | Line 366 | public class SubmissionPublisherTest ext
366		assertTrue(p.isClosed());
367		assertNull(p.getClosedException());
368		s1.awaitComplete();
369	<	assertEquals(s1.nexts, 1);
370	<	assertEquals(s1.completes, 1);
369	>	assertEquals(1, s1.nexts);
370	>	assertEquals(1, s1.completes);
371		s2.awaitComplete();
372	<	assertEquals(s2.nexts, 1);
373	<	assertEquals(s2.completes, 1);
372	>	assertEquals(1, s2.nexts);
373	>	assertEquals(1, s2.completes);
374		}
375
376		/**
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 404 | 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(s1.errors, 1);
392	>	assertEquals(1, s1.errors);
393	>	s2.awaitSubscribe();
394		s2.awaitError();
395		assertTrue(s2.nexts <= 1);
396	<	assertEquals(s2.errors, 1);
396	>	assertEquals(1, s2.errors);
397		}
398
399		/**
#	Line 428 | Line 412 | public class SubmissionPublisherTest ext
412		p.submit(i);
413		p.close();
414		s2.awaitComplete();
415	<	assertEquals(s2.nexts, 20);
416	<	assertEquals(s2.completes, 1);
415	>	assertEquals(20, s2.nexts);
416	>	assertEquals(1, s2.completes);
417		assertTrue(s1.nexts < 20);
418		assertFalse(p.isSubscribed(s1));
419		}
#	Line 449 | Line 433 | public class SubmissionPublisherTest ext
433		p.submit(2);
434		p.close();
435		s2.awaitComplete();
436	<	assertEquals(s2.nexts, 2);
436	>	assertEquals(2, s2.nexts);
437		s1.awaitComplete();
438	<	assertEquals(s1.errors, 1);
438	>	assertEquals(1, s1.errors);
439		}
440
441		/**
442	<	* If a handler is supplied in conctructor, it is invoked when
442	>	* If a handler is supplied in constructor, it is invoked when
443		* subscriber throws an exception in onNext
444		*/
445		public void testThrowOnNextHandler() {
446		AtomicInteger calls = new AtomicInteger();
447	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
448	<	(basicExecutor, 8,
465	<	(s, e) -> calls.getAndIncrement());
447	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
448	>	basicExecutor, 8, (s, e) -> calls.getAndIncrement());
449		TestSubscriber s1 = new TestSubscriber();
450		TestSubscriber s2 = new TestSubscriber();
451		p.subscribe(s1);
#	Line 473 | Line 456 | public class SubmissionPublisherTest ext
456		p.submit(2);
457		p.close();
458		s2.awaitComplete();
459	<	assertEquals(s2.nexts, 2);
460	<	assertEquals(s2.completes, 1);
459	>	assertEquals(2, s2.nexts);
460	>	assertEquals(1, s2.completes);
461		s1.awaitError();
462	<	assertEquals(s1.errors, 1);
463	<	assertEquals(calls.get(), 1);
462	>	assertEquals(1, s1.errors);
463	>	assertEquals(1, calls.get());
464		}
465
466		/**
#	Line 494 | Line 477 | public class SubmissionPublisherTest ext
477		p.close();
478		s2.awaitComplete();
479		s1.awaitComplete();
480	<	assertEquals(s2.nexts, 20);
481	<	assertEquals(s2.completes, 1);
482	<	assertEquals(s1.nexts, 20);
483	<	assertEquals(s1.completes, 1);
480	>	assertEquals(20, s2.nexts);
481	>	assertEquals(1, s2.completes);
482	>	assertEquals(20, s1.nexts);
483	>	assertEquals(1, s1.completes);
484		}
485
486		/**
#	Line 515 | Line 498 | public class SubmissionPublisherTest ext
498		p.submit(1);
499		p.submit(2);
500		s2.awaitNext(1);
501	<	assertEquals(s1.nexts, 0);
501	>	assertEquals(0, s1.nexts);
502		s1.sn.request(3);
503		p.submit(3);
504		p.close();
505		s2.awaitComplete();
506	<	assertEquals(s2.nexts, 3);
507	<	assertEquals(s2.completes, 1);
506	>	assertEquals(3, s2.nexts);
507	>	assertEquals(1, s2.completes);
508		s1.awaitComplete();
509		assertTrue(s1.nexts > 0);
510	<	assertEquals(s1.completes, 1);
510	>	assertEquals(1, s1.completes);
511		}
512
513		/**
#	Line 543 | Line 526 | public class SubmissionPublisherTest ext
526		p.submit(2);
527		p.close();
528		s2.awaitComplete();
529	<	assertEquals(s2.nexts, 2);
530	<	assertEquals(s2.completes, 1);
529	>	assertEquals(2, s2.nexts);
530	>	assertEquals(1, s2.completes);
531		s1.awaitNext(1);
532	<	assertEquals(s1.nexts, 1);
532	>	assertEquals(1, s1.nexts);
533		}
534
535		/**
536	<	* Negative request causes error
536	>	* Non-positive request causes error
537		*/
538		public void testRequest3() {
539		SubmissionPublisher<Integer> p = basicPublisher();
540		TestSubscriber s1 = new TestSubscriber();
541		TestSubscriber s2 = new TestSubscriber();
542	+	TestSubscriber s3 = new TestSubscriber();
543		p.subscribe(s1);
544		p.subscribe(s2);
545	+	p.subscribe(s3);
546	+	s3.awaitSubscribe();
547		s2.awaitSubscribe();
548		s1.awaitSubscribe();
549		s1.sn.request(-1L);
550	+	s3.sn.request(0L);
551		p.submit(1);
552		p.submit(2);
553		p.close();
554		s2.awaitComplete();
555	<	assertEquals(s2.nexts, 2);
556	<	assertEquals(s2.completes, 1);
555	>	assertEquals(2, s2.nexts);
556	>	assertEquals(1, s2.completes);
557		s1.awaitError();
558	<	assertEquals(s1.errors, 1);
558	>	assertEquals(1, s1.errors);
559		assertTrue(s1.lastError instanceof IllegalArgumentException);
560	+	s3.awaitError();
561	+	assertEquals(1, s3.errors);
562	+	assertTrue(s3.lastError instanceof IllegalArgumentException);
563		}
564
565		/**
#	Line 582 | Line 572 | public class SubmissionPublisherTest ext
572		s.request = false;
573		p.subscribe(s);
574		s.awaitSubscribe();
575	<	assertEquals(p.estimateMinimumDemand(), 0);
575	>	assertEquals(0, p.estimateMinimumDemand());
576		s.sn.request(1);
577	<	assertEquals(p.estimateMinimumDemand(), 1);
577	>	assertEquals(1, p.estimateMinimumDemand());
578		p.submit(1);
579		s.awaitNext(1);
580	<	assertEquals(p.estimateMinimumDemand(), 0);
580	>	assertEquals(0, p.estimateMinimumDemand());
581		}
582
583		/**
584	<	* Submit to a publisher with no subscribers returns lag 0
584	>	* submit to a publisher with no subscribers returns lag 0
585		*/
586		public void testEmptySubmit() {
587		SubmissionPublisher<Integer> p = basicPublisher();
588	<	assertEquals(p.submit(1), 0);
588	>	assertEquals(0, p.submit(1));
589		}
590
591		/**
592	<	* Submit(null) throws NPE
592	>	* submit(null) throws NPE
593		*/
594		public void testNullSubmit() {
595		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 610 | Line 600 | public class SubmissionPublisherTest ext
600		}
601
602		/**
603	<	* Submit returns number of lagged items, compatible with result
603	>	* submit returns number of lagged items, compatible with result
604		* of estimateMaximumLag.
605		*/
606		public void testLaggedSubmit() {
#	Line 623 | Line 613 | public class SubmissionPublisherTest ext
613		p.subscribe(s2);
614		s2.awaitSubscribe();
615		s1.awaitSubscribe();
616	<	assertEquals(p.submit(1), 1);
616	>	assertEquals(1, p.submit(1));
617		assertTrue(p.estimateMaximumLag() >= 1);
618		assertTrue(p.submit(2) >= 2);
619		assertTrue(p.estimateMaximumLag() >= 2);
#	Line 634 | Line 624 | public class SubmissionPublisherTest ext
624		p.submit(4);
625		p.close();
626		s2.awaitComplete();
627	<	assertEquals(s2.nexts, 4);
627	>	assertEquals(4, s2.nexts);
628		s1.awaitComplete();
629	<	assertEquals(s2.nexts, 4);
629	>	assertEquals(4, s2.nexts);
630		}
631
632		/**
633		* submit eventually issues requested items when buffer capacity is 1
634		*/
635		public void testCap1Submit() {
636	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
637	<	basicExecutor, 1);
636	>	SubmissionPublisher<Integer> p
637	>	= new SubmissionPublisher<>(basicExecutor, 1);
638		TestSubscriber s1 = new TestSubscriber();
639		TestSubscriber s2 = new TestSubscriber();
640		p.subscribe(s1);
#	Line 656 | Line 646 | public class SubmissionPublisherTest ext
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		}

Diff Legend

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