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.1 by dl, Mon Sep 7 17:14:06 2015 UTC vs.
Revision 1.21 by jsr166, Mon May 29 22:44:27 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
32	<	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());
55	<
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	<
36	>
37		static class SPException extends RuntimeException {}
38
39		class TestSubscriber implements Subscriber<Integer> {
#	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 {
177	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(null, 8);
177	>	new SubmissionPublisher<Integer>(null, 8);
178		shouldThrow();
179	<	} catch (NullPointerException success) {
199	<	}
179	>	} catch (NullPointerException success) {}
180		}
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);
188		try {
189	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, -1);
189	>	new SubmissionPublisher<Integer>(e, -1);
190		shouldThrow();
191	<	} catch (IllegalArgumentException success) {
212	<	}
191	>	} catch (IllegalArgumentException success) {}
192		}
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 226 | Line 206 | public class SubmissionPublisherTest ext
206		try {
207		p.submit(1);
208		shouldThrow();
209	<	}
230	<	catch(IllegalStateException success) {
231	<	}
209	>	} catch (IllegalStateException success) {}
210		Throwable ex = new SPException();
211		p.closeExceptionally(ex);
212		assertTrue(p.isClosed());
#	Line 237 | 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 251 | Line 229 | public class SubmissionPublisherTest ext
229		try {
230		p.submit(1);
231		shouldThrow();
232	<	}
255	<	catch(IllegalStateException success) {
256	<	}
232	>	} catch (IllegalStateException success) {}
233		p.close();
234		assertTrue(p.isClosed());
235		assertSame(p.getClosedException(), ex);
#	Line 270 | 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 303 | 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 321 | 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 334 | 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 358 | Line 335 | public class SubmissionPublisherTest ext
335		s.throwOnCall = true;
336		try {
337		p.subscribe(s);
338	<	} catch(Exception ok) {
362	<	}
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();
350		try {
351		p.subscribe(null);
352	<	} catch(NullPointerException success) {
353	<	}
352	>	shouldThrow();
353	>	} catch (NullPointerException success) {}
354		checkInitialState(p);
355		}
356
#	Line 392 | 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 411 | 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 435 | 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 456 | 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,
472	<	(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 480 | 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 501 | 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 516 | 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 550 | 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		/**
#	Line 589 | 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);
579	>	assertEquals(1, p.estimateMinimumDemand());
580		p.submit(1);
581		s.awaitNext(1);
582	<	assertEquals(p.estimateMinimumDemand(), 0);
582	>	assertEquals(0, p.estimateMinimumDemand());
583		}
584
585		/**
586	<	* Submit to a publisher with no subscribers returns lag 0
586	>	* submit to a publisher with no subscribers returns lag 0
587		*/
588		public void testEmptySubmit() {
589		SubmissionPublisher<Integer> p = basicPublisher();
590	<	assertEquals(p.submit(1), 0);
590	>	assertEquals(0, p.submit(1));
591		}
592
593		/**
594	<	* Submit(null) throws NPE
594	>	* submit(null) throws NPE
595		*/
596		public void testNullSubmit() {
597		SubmissionPublisher<Integer> p = basicPublisher();
598		try {
599		p.submit(null);
600	<	} catch (NullPointerException success) {
601	<	}
600	>	shouldThrow();
601	>	} catch (NullPointerException success) {}
602		}
603
604		/**
605	<	* Submit returns number of lagged items, compatible with result
605	>	* submit returns number of lagged items, compatible with result
606		* of estimateMaximumLag.
607		*/
608		public void testLaggedSubmit() {
#	Line 630 | Line 615 | public class SubmissionPublisherTest ext
615		p.subscribe(s2);
616		s2.awaitSubscribe();
617		s1.awaitSubscribe();
618	<	assertEquals(p.submit(1), 1);
618	>	assertEquals(1, p.submit(1));
619		assertTrue(p.estimateMaximumLag() >= 1);
620		assertTrue(p.submit(2) >= 2);
621		assertTrue(p.estimateMaximumLag() >= 2);
#	Line 641 | Line 626 | public class SubmissionPublisherTest ext
626		p.submit(4);
627		p.close();
628		s2.awaitComplete();
629	<	assertEquals(s2.nexts, 4);
629	>	assertEquals(4, s2.nexts);
630		s1.awaitComplete();
631	<	assertEquals(s2.nexts, 4);
631	>	assertEquals(4, s2.nexts);
632		}
633
634		/**
635		* submit eventually issues requested items when buffer capacity is 1
636		*/
637		public void testCap1Submit() {
638	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
639	<	basicExecutor, 1);
638	>	SubmissionPublisher<Integer> p
639	>	= new SubmissionPublisher<>(basicExecutor, 1);
640		TestSubscriber s1 = new TestSubscriber();
641		TestSubscriber s2 = new TestSubscriber();
642		p.subscribe(s1);
#	Line 663 | Line 648 | public class SubmissionPublisherTest ext
648		p.close();
649		s2.awaitComplete();
650		s1.awaitComplete();
651	<	assertEquals(s2.nexts, 20);
652	<	assertEquals(s2.completes, 1);
653	<	assertEquals(s1.nexts, 20);
654	<	assertEquals(s1.completes, 1);
651	>	assertEquals(20, s2.nexts);
652	>	assertEquals(1, s2.completes);
653	>	assertEquals(20, s1.nexts);
654	>	assertEquals(1, s1.completes);
655		}
656	<
656	>
657		static boolean noopHandle(AtomicInteger count) {
658		count.getAndIncrement();
659		return false;
#	Line 681 | Line 666 | public class SubmissionPublisherTest ext
666		}
667
668		/**
669	<	* Offer to a publisher with no subscribers returns lag 0
669	>	* offer to a publisher with no subscribers returns lag 0
670		*/
671		public void testEmptyOffer() {
672		SubmissionPublisher<Integer> p = basicPublisher();
673	<	assertEquals(p.offer(1, null), 0);
673	>	assertEquals(0, p.offer(1, null));
674		}
675
676		/**
677	<	* Offer(null) throws NPE
677	>	* offer(null) throws NPE
678		*/
679		public void testNullOffer() {
680		SubmissionPublisher<Integer> p = basicPublisher();
681		try {
682		p.offer(null, null);
683	<	} catch (NullPointerException success) {
684	<	}
683	>	shouldThrow();
684	>	} catch (NullPointerException success) {}
685		}
686
687		/**
688	<	* Offer returns number of lagged items if not saturated
688	>	* offer returns number of lagged items if not saturated
689		*/
690		public void testLaggedOffer() {
691		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 720 | Line 705 | public class SubmissionPublisherTest ext
705		p.offer(4, null);
706		p.close();
707		s2.awaitComplete();
708	<	assertEquals(s2.nexts, 4);
708	>	assertEquals(4, s2.nexts);
709		s1.awaitComplete();
710	<	assertEquals(s2.nexts, 4);
710	>	assertEquals(4, s2.nexts);
711		}
712
713		/**
714	<	* Offer reports drops if saturated
714	>	* offer reports drops if saturated
715		*/
716		public void testDroppedOffer() {
717	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
718	<	basicExecutor, 4);
717	>	SubmissionPublisher<Integer> p
718	>	= new SubmissionPublisher<>(basicExecutor, 4);
719		TestSubscriber s1 = new TestSubscriber();
720		s1.request = false;
721		TestSubscriber s2 = new TestSubscriber();
#	Line 754 | Line 739 | public class SubmissionPublisherTest ext
739		}
740
741		/**
742	<	* Offer invokes drop handler if saturated
742	>	* offer invokes drop handler if saturated
743		*/
744		public void testHandledDroppedOffer() {
745		AtomicInteger calls = new AtomicInteger();
746	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
747	<	basicExecutor, 4);
746	>	SubmissionPublisher<Integer> p
747	>	= new SubmissionPublisher<>(basicExecutor, 4);
748		TestSubscriber s1 = new TestSubscriber();
749		s1.request = false;
750		TestSubscriber s2 = new TestSubscriber();
#	Line 781 | Line 766 | public class SubmissionPublisherTest ext
766		assertTrue(calls.get() >= 4);
767		}
768
784	–
769		/**
770	<	* Offer succeeds if drop handler forces request
770	>	* offer succeeds if drop handler forces request
771		*/
772		public void testRecoveredHandledDroppedOffer() {
773		AtomicInteger calls = new AtomicInteger();
774	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
775	<	basicExecutor, 4);
774	>	SubmissionPublisher<Integer> p
775	>	= new SubmissionPublisher<>(basicExecutor, 4);
776		TestSubscriber s1 = new TestSubscriber();
777		s1.request = false;
778		TestSubscriber s2 = new TestSubscriber();
#	Line 805 | Line 789 | public class SubmissionPublisherTest ext
789		p.close();
790		s2.awaitComplete();
791		s1.awaitComplete();
792	<	assertEquals(s1.nexts + s2.nexts, n);
792	>	assertEquals(n, s1.nexts + s2.nexts);
793		assertTrue(calls.get() >= 2);
794		}
795
812	–
796		/**
797	<	* TimedOffer to a publisher with no subscribers returns lag 0
797	>	* Timed offer to a publisher with no subscribers returns lag 0
798		*/
799		public void testEmptyTimedOffer() {
800		SubmissionPublisher<Integer> p = basicPublisher();
801	<	assertEquals(p.offer(1, null), 0);
801	>	long startTime = System.nanoTime();
802	>	assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
803	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
804		}
805
806		/**
807	<	* Timed Offer with null item or TimeUnit throws NPE
807	>	* Timed offer with null item or TimeUnit throws NPE
808		*/
809		public void testNullTimedOffer() {
810		SubmissionPublisher<Integer> p = basicPublisher();
811	+	long startTime = System.nanoTime();
812		try {
813	<	p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
814	<	} catch (NullPointerException success) {
815	<	}
813	>	p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
814	>	shouldThrow();
815	>	} catch (NullPointerException success) {}
816		try {
817	<	p.offer(1, SHORT_DELAY_MS, null, null);
818	<	} catch (NullPointerException success) {
819	<	}
817	>	p.offer(1, LONG_DELAY_MS, null, null);
818	>	shouldThrow();
819	>	} catch (NullPointerException success) {}
820	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
821		}
822
823		/**
824	<	* Timed Offer returns number of lagged items if not saturated
824	>	* Timed offer returns number of lagged items if not saturated
825		*/
826		public void testLaggedTimedOffer() {
827		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 846 | Line 833 | public class SubmissionPublisherTest ext
833		p.subscribe(s2);
834		s2.awaitSubscribe();
835		s1.awaitSubscribe();
836	<	assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
837	<	assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
836	>	long startTime = System.nanoTime();
837	>	assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
838	>	assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
839		s1.sn.request(4);
840	<	assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
840	>	assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
841		s2.sn.request(4);
842	<	p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
842	>	p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
843		p.close();
844		s2.awaitComplete();
845	<	assertEquals(s2.nexts, 4);
845	>	assertEquals(4, s2.nexts);
846		s1.awaitComplete();
847	<	assertEquals(s2.nexts, 4);
847	>	assertEquals(4, s2.nexts);
848	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
849		}
850
851		/**
852	<	* Timed Offer reports drops if saturated
852	>	* Timed offer reports drops if saturated
853		*/
854		public void testDroppedTimedOffer() {
855	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
856	<	basicExecutor, 4);
855	>	SubmissionPublisher<Integer> p
856	>	= new SubmissionPublisher<>(basicExecutor, 4);
857		TestSubscriber s1 = new TestSubscriber();
858		s1.request = false;
859		TestSubscriber s2 = new TestSubscriber();
#	Line 873 | Line 862 | public class SubmissionPublisherTest ext
862		p.subscribe(s2);
863		s2.awaitSubscribe();
864		s1.awaitSubscribe();
865	+	long delay = timeoutMillis();
866		for (int i = 1; i <= 4; ++i)
867	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
868	<	p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
869	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
867	>	assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
868	>	long startTime = System.nanoTime();
869	>	assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
870		s1.sn.request(64);
871	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
871	>	assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
872	>	// 2 * delay should elapse but check only 1 * delay to allow timer slop
873	>	assertTrue(millisElapsedSince(startTime) >= delay);
874		s2.sn.request(64);
875		p.close();
876		s2.awaitComplete();
#	Line 888 | Line 880 | public class SubmissionPublisherTest ext
880		}
881
882		/**
883	<	* Timed Offer invokes drop handler if saturated
883	>	* Timed offer invokes drop handler if saturated
884		*/
885		public void testHandledDroppedTimedOffer() {
886		AtomicInteger calls = new AtomicInteger();
887	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
888	<	basicExecutor, 4);
887	>	SubmissionPublisher<Integer> p
888	>	= new SubmissionPublisher<>(basicExecutor, 4);
889		TestSubscriber s1 = new TestSubscriber();
890		s1.request = false;
891		TestSubscriber s2 = new TestSubscriber();
#	Line 902 | Line 894 | public class SubmissionPublisherTest ext
894		p.subscribe(s2);
895		s2.awaitSubscribe();
896		s1.awaitSubscribe();
897	+	long delay = timeoutMillis();
898		for (int i = 1; i <= 4; ++i)
899	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
900	<	p.offer(5, (s, x) -> noopHandle(calls));
901	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
899	>	assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
900	>	long startTime = System.nanoTime();
901	>	assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
902		s1.sn.request(64);
903	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
903	>	assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
904	>	assertTrue(millisElapsedSince(startTime) >= delay);
905		s2.sn.request(64);
906		p.close();
907		s2.awaitComplete();
#	Line 916 | Line 910 | public class SubmissionPublisherTest ext
910		}
911
912		/**
913	<	* Timed Offer succeeds if drop handler forces request
913	>	* Timed offer succeeds if drop handler forces request
914		*/
915		public void testRecoveredHandledDroppedTimedOffer() {
916		AtomicInteger calls = new AtomicInteger();
917	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
918	<	basicExecutor, 4);
917	>	SubmissionPublisher<Integer> p
918	>	= new SubmissionPublisher<>(basicExecutor, 4);
919		TestSubscriber s1 = new TestSubscriber();
920		s1.request = false;
921		TestSubscriber s2 = new TestSubscriber();
#	Line 931 | Line 925 | public class SubmissionPublisherTest ext
925		s2.awaitSubscribe();
926		s1.awaitSubscribe();
927		int n = 0;
928	<	for (int i = 1; i <= 8; ++i) {
929	<	int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
928	>	long delay = timeoutMillis();
929	>	long startTime = System.nanoTime();
930	>	for (int i = 1; i <= 6; ++i) {
931	>	int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
932		n = n + 2 + (d < 0 ? d : 0);
933		}
934	+	assertTrue(millisElapsedSince(startTime) >= delay);
935		p.close();
936		s2.awaitComplete();
937		s1.awaitComplete();
938	<	assertEquals(s1.nexts + s2.nexts, n);
938	>	assertEquals(n, s1.nexts + s2.nexts);
939		assertTrue(calls.get() >= 2);
940		}
941
942	+	/**
943	+	* consume returns a CompletableFuture that is done when
944	+	* publisher completes
945	+	*/
946	+	public void testConsume() {
947	+	AtomicInteger sum = new AtomicInteger();
948	+	SubmissionPublisher<Integer> p = basicPublisher();
949	+	CompletableFuture<Void> f =
950	+	p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
951	+	int n = 20;
952	+	for (int i = 1; i <= n; ++i)
953	+	p.submit(i);
954	+	p.close();
955	+	f.join();
956	+	assertEquals((n * (n + 1)) / 2, sum.get());
957	+	}
958	+
959	+	/**
960	+	* consume(null) throws NPE
961	+	*/
962	+	public void testConsumeNPE() {
963	+	SubmissionPublisher<Integer> p = basicPublisher();
964	+	try {
965	+	CompletableFuture<Void> f = p.consume(null);
966	+	shouldThrow();
967	+	} catch (NullPointerException success) {}
968	+	}
969	+
970	+	/**
971	+	* consume eventually stops processing published items if cancelled
972	+	*/
973	+	public void testCancelledConsume() {
974	+	AtomicInteger count = new AtomicInteger();
975	+	SubmissionPublisher<Integer> p = basicPublisher();
976	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
977	+	f.cancel(true);
978	+	int n = 1000000; // arbitrary limit
979	+	for (int i = 1; i <= n; ++i)
980	+	p.submit(i);
981	+	assertTrue(count.get() < n);
982	+	}
983
984		}

Diff Legend

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