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.3 by jsr166, Mon Sep 7 20:28:47 2015 UTC vs.
Revision 1.23 by dl, Sun Nov 26 21:37:56 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		/**
402		* Cancelling a subscription eventually causes no more onNexts to be issued
403		*/
404		public void testCancel() {
405	<	SubmissionPublisher<Integer> p = basicPublisher();
405	>	SubmissionPublisher<Integer> p =
406	>	new SubmissionPublisher<Integer>(basicExecutor, 4); // must be < 20
407		TestSubscriber s1 = new TestSubscriber();
408		TestSubscriber s2 = new TestSubscriber();
409		p.subscribe(s1);
#	Line 428 | Line 415 | public class SubmissionPublisherTest ext
415		p.submit(i);
416		p.close();
417		s2.awaitComplete();
418	<	assertEquals(s2.nexts, 20);
419	<	assertEquals(s2.completes, 1);
418	>	assertEquals(20, s2.nexts);
419	>	assertEquals(1, s2.completes);
420		assertTrue(s1.nexts < 20);
421		assertFalse(p.isSubscribed(s1));
422		}
#	Line 449 | Line 436 | public class SubmissionPublisherTest ext
436		p.submit(2);
437		p.close();
438		s2.awaitComplete();
439	<	assertEquals(s2.nexts, 2);
439	>	assertEquals(2, s2.nexts);
440		s1.awaitComplete();
441	<	assertEquals(s1.errors, 1);
441	>	assertEquals(1, s1.errors);
442		}
443
444		/**
445	<	* If a handler is supplied in conctructor, it is invoked when
445	>	* If a handler is supplied in constructor, it is invoked when
446		* subscriber throws an exception in onNext
447		*/
448		public void testThrowOnNextHandler() {
449		AtomicInteger calls = new AtomicInteger();
450	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
451	<	(basicExecutor, 8,
465	<	(s, e) -> calls.getAndIncrement());
450	>	SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
451	>	basicExecutor, 8, (s, e) -> calls.getAndIncrement());
452		TestSubscriber s1 = new TestSubscriber();
453		TestSubscriber s2 = new TestSubscriber();
454		p.subscribe(s1);
#	Line 473 | Line 459 | public class SubmissionPublisherTest ext
459		p.submit(2);
460		p.close();
461		s2.awaitComplete();
462	<	assertEquals(s2.nexts, 2);
463	<	assertEquals(s2.completes, 1);
462	>	assertEquals(2, s2.nexts);
463	>	assertEquals(1, s2.completes);
464		s1.awaitError();
465	<	assertEquals(s1.errors, 1);
466	<	assertEquals(calls.get(), 1);
465	>	assertEquals(1, s1.errors);
466	>	assertEquals(1, calls.get());
467		}
468
469		/**
#	Line 494 | Line 480 | public class SubmissionPublisherTest ext
480		p.close();
481		s2.awaitComplete();
482		s1.awaitComplete();
483	<	assertEquals(s2.nexts, 20);
484	<	assertEquals(s2.completes, 1);
485	<	assertEquals(s1.nexts, 20);
486	<	assertEquals(s1.completes, 1);
483	>	assertEquals(20, s2.nexts);
484	>	assertEquals(1, s2.completes);
485	>	assertEquals(20, s1.nexts);
486	>	assertEquals(1, s1.completes);
487		}
488
489		/**
#	Line 509 | Line 495 | public class SubmissionPublisherTest ext
495		s1.request = false;
496		p.subscribe(s1);
497		s1.awaitSubscribe();
498	<	assertTrue(p.estimateMinimumDemand() == 0);
498	>	assertEquals(0, p.estimateMinimumDemand());
499		TestSubscriber s2 = new TestSubscriber();
500		p.subscribe(s2);
501		p.submit(1);
502		p.submit(2);
503		s2.awaitNext(1);
504	<	assertEquals(s1.nexts, 0);
504	>	assertEquals(0, s1.nexts);
505		s1.sn.request(3);
506		p.submit(3);
507		p.close();
508		s2.awaitComplete();
509	<	assertEquals(s2.nexts, 3);
510	<	assertEquals(s2.completes, 1);
509	>	assertEquals(3, s2.nexts);
510	>	assertEquals(1, s2.completes);
511		s1.awaitComplete();
512		assertTrue(s1.nexts > 0);
513	<	assertEquals(s1.completes, 1);
513	>	assertEquals(1, s1.completes);
514		}
515
516		/**
#	Line 543 | Line 529 | public class SubmissionPublisherTest ext
529		p.submit(2);
530		p.close();
531		s2.awaitComplete();
532	<	assertEquals(s2.nexts, 2);
533	<	assertEquals(s2.completes, 1);
532	>	assertEquals(2, s2.nexts);
533	>	assertEquals(1, s2.completes);
534		s1.awaitNext(1);
535	<	assertEquals(s1.nexts, 1);
535	>	assertEquals(1, s1.nexts);
536		}
537
538		/**
539	<	* Negative request causes error
539	>	* Non-positive request causes error
540		*/
541		public void testRequest3() {
542		SubmissionPublisher<Integer> p = basicPublisher();
543		TestSubscriber s1 = new TestSubscriber();
544		TestSubscriber s2 = new TestSubscriber();
545	+	TestSubscriber s3 = new TestSubscriber();
546		p.subscribe(s1);
547		p.subscribe(s2);
548	+	p.subscribe(s3);
549	+	s3.awaitSubscribe();
550		s2.awaitSubscribe();
551		s1.awaitSubscribe();
552		s1.sn.request(-1L);
553	+	s3.sn.request(0L);
554		p.submit(1);
555		p.submit(2);
556		p.close();
557		s2.awaitComplete();
558	<	assertEquals(s2.nexts, 2);
559	<	assertEquals(s2.completes, 1);
558	>	assertEquals(2, s2.nexts);
559	>	assertEquals(1, s2.completes);
560		s1.awaitError();
561	<	assertEquals(s1.errors, 1);
561	>	assertEquals(1, s1.errors);
562		assertTrue(s1.lastError instanceof IllegalArgumentException);
563	+	s3.awaitError();
564	+	assertEquals(1, s3.errors);
565	+	assertTrue(s3.lastError instanceof IllegalArgumentException);
566		}
567
568		/**
569		* estimateMinimumDemand reports 0 until request, nonzero after
570	<	* request, and zero again after delivery
570	>	* request
571		*/
572		public void testEstimateMinimumDemand() {
573		TestSubscriber s = new TestSubscriber();
#	Line 582 | Line 575 | public class SubmissionPublisherTest ext
575		s.request = false;
576		p.subscribe(s);
577		s.awaitSubscribe();
578	<	assertEquals(p.estimateMinimumDemand(), 0);
578	>	assertEquals(0, p.estimateMinimumDemand());
579		s.sn.request(1);
580	<	assertEquals(p.estimateMinimumDemand(), 1);
588	<	p.submit(1);
589	<	s.awaitNext(1);
590	<	assertEquals(p.estimateMinimumDemand(), 0);
580	>	assertEquals(1, p.estimateMinimumDemand());
581		}
582
583		/**
#	Line 595 | Line 585 | public class SubmissionPublisherTest ext
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		/**
#	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);
641		p.subscribe(s2);
642		for (int i = 1; i <= 20; ++i) {
653	–	assertTrue(p.estimateMinimumDemand() <= 1);
643		assertTrue(p.submit(i) >= 0);
644		}
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 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
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 751 | Line 740 | public class SubmissionPublisherTest ext
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
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		/**
#	Line 816 | Line 805 | public class SubmissionPublisherTest ext
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		/**
#	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
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 885 | Line 881 | public class SubmissionPublisherTest ext
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 913 | Line 911 | public class SubmissionPublisherTest ext
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