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.17 by dl, Thu Dec 15 17:34:51 2016 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>();
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 185 | Line 165 | public class SubmissionPublisherTest ext
165		SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, 8);
166		checkInitialState(p);
167		assertSame(p.getExecutor(), e);
168	<	assertEquals(p.getMaxBufferCapacity(), 8);
168	>	assertEquals(8, p.getMaxBufferCapacity());
169		}
170
171		/**
#	Line 193 | Line 173 | public class SubmissionPublisherTest ext
173		*/
174		public void testConstructor3() {
175		try {
176	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(null, 8);
176	>	new SubmissionPublisher<Integer>(null, 8);
177		shouldThrow();
178	<	} catch (NullPointerException success) {
199	<	}
178	>	} catch (NullPointerException success) {}
179		}
180
181		/**
#	Line 206 | Line 185 | public class SubmissionPublisherTest ext
185		public void testConstructor4() {
186		Executor e = Executors.newFixedThreadPool(1);
187		try {
188	<	SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, -1);
188	>	new SubmissionPublisher<Integer>(e, -1);
189		shouldThrow();
190	<	} catch (IllegalArgumentException success) {
212	<	}
190	>	} catch (IllegalArgumentException success) {}
191		}
192
193		/**
#	Line 226 | Line 204 | public class SubmissionPublisherTest ext
204		try {
205		p.submit(1);
206		shouldThrow();
207	<	}
230	<	catch(IllegalStateException success) {
231	<	}
207	>	} catch (IllegalStateException success) {}
208		Throwable ex = new SPException();
209		p.closeExceptionally(ex);
210		assertTrue(p.isClosed());
#	Line 251 | Line 227 | public class SubmissionPublisherTest ext
227		try {
228		p.submit(1);
229		shouldThrow();
230	<	}
255	<	catch(IllegalStateException success) {
256	<	}
230	>	} catch (IllegalStateException success) {}
231		p.close();
232		assertTrue(p.isClosed());
233		assertSame(p.getClosedException(), ex);
#	Line 270 | 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 303 | 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 321 | 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 334 | 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 358 | Line 333 | public class SubmissionPublisherTest ext
333		s.throwOnCall = true;
334		try {
335		p.subscribe(s);
336	<	} catch(Exception ok) {
362	<	}
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();
348		try {
349		p.subscribe(null);
350	<	} catch(NullPointerException success) {
351	<	}
350	>	shouldThrow();
351	>	} catch (NullPointerException success) {}
352		checkInitialState(p);
353		}
354
#	Line 392 | 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 411 | 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 435 | 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 456 | 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() {
#	Line 480 | Line 457 | public class SubmissionPublisherTest ext
457		p.submit(2);
458		p.close();
459		s2.awaitComplete();
460	<	assertEquals(s2.nexts, 2);
461	<	assertEquals(s2.completes, 1);
460	>	assertEquals(2, s2.nexts);
461	>	assertEquals(1, s2.completes);
462		s1.awaitError();
463	<	assertEquals(s1.errors, 1);
464	<	assertEquals(calls.get(), 1);
463	>	assertEquals(1, s1.errors);
464	>	assertEquals(1, calls.get());
465		}
466
467		/**
#	Line 501 | Line 478 | public class SubmissionPublisherTest ext
478		p.close();
479		s2.awaitComplete();
480		s1.awaitComplete();
481	<	assertEquals(s2.nexts, 20);
482	<	assertEquals(s2.completes, 1);
483	<	assertEquals(s1.nexts, 20);
484	<	assertEquals(s1.completes, 1);
481	>	assertEquals(20, s2.nexts);
482	>	assertEquals(1, s2.completes);
483	>	assertEquals(20, s1.nexts);
484	>	assertEquals(1, s1.completes);
485		}
486
487		/**
#	Line 522 | Line 499 | public class SubmissionPublisherTest ext
499		p.submit(1);
500		p.submit(2);
501		s2.awaitNext(1);
502	<	assertEquals(s1.nexts, 0);
502	>	assertEquals(0, s1.nexts);
503		s1.sn.request(3);
504		p.submit(3);
505		p.close();
506		s2.awaitComplete();
507	<	assertEquals(s2.nexts, 3);
508	<	assertEquals(s2.completes, 1);
507	>	assertEquals(3, s2.nexts);
508	>	assertEquals(1, s2.completes);
509		s1.awaitComplete();
510		assertTrue(s1.nexts > 0);
511	<	assertEquals(s1.completes, 1);
511	>	assertEquals(1, s1.completes);
512		}
513
514		/**
#	Line 550 | Line 527 | public class SubmissionPublisherTest ext
527		p.submit(2);
528		p.close();
529		s2.awaitComplete();
530	<	assertEquals(s2.nexts, 2);
531	<	assertEquals(s2.completes, 1);
530	>	assertEquals(2, s2.nexts);
531	>	assertEquals(1, s2.completes);
532		s1.awaitNext(1);
533	<	assertEquals(s1.nexts, 1);
533	>	assertEquals(1, s1.nexts);
534		}
535
536		/**
#	Line 572 | Line 549 | public class SubmissionPublisherTest ext
549		p.submit(2);
550		p.close();
551		s2.awaitComplete();
552	<	assertEquals(s2.nexts, 2);
553	<	assertEquals(s2.completes, 1);
552	>	assertEquals(2, s2.nexts);
553	>	assertEquals(1, s2.completes);
554		s1.awaitError();
555	<	assertEquals(s1.errors, 1);
555	>	assertEquals(1, s1.errors);
556		assertTrue(s1.lastError instanceof IllegalArgumentException);
557		}
558
#	Line 589 | Line 566 | public class SubmissionPublisherTest ext
566		s.request = false;
567		p.subscribe(s);
568		s.awaitSubscribe();
569	<	assertEquals(p.estimateMinimumDemand(), 0);
569	>	assertEquals(0, p.estimateMinimumDemand());
570		s.sn.request(1);
571	<	assertEquals(p.estimateMinimumDemand(), 1);
571	>	assertEquals(1, p.estimateMinimumDemand());
572		p.submit(1);
573		s.awaitNext(1);
574	<	assertEquals(p.estimateMinimumDemand(), 0);
574	>	assertEquals(0, p.estimateMinimumDemand());
575		}
576
577		/**
578	<	* Submit to a publisher with no subscribers returns lag 0
578	>	* submit to a publisher with no subscribers returns lag 0
579		*/
580		public void testEmptySubmit() {
581		SubmissionPublisher<Integer> p = basicPublisher();
582	<	assertEquals(p.submit(1), 0);
582	>	assertEquals(0, p.submit(1));
583		}
584
585		/**
586	<	* Submit(null) throws NPE
586	>	* submit(null) throws NPE
587		*/
588		public void testNullSubmit() {
589		SubmissionPublisher<Integer> p = basicPublisher();
590		try {
591		p.submit(null);
592	<	} catch (NullPointerException success) {
593	<	}
592	>	shouldThrow();
593	>	} catch (NullPointerException success) {}
594		}
595
596		/**
597	<	* Submit returns number of lagged items, compatible with result
597	>	* submit returns number of lagged items, compatible with result
598		* of estimateMaximumLag.
599		*/
600		public void testLaggedSubmit() {
#	Line 630 | Line 607 | public class SubmissionPublisherTest ext
607		p.subscribe(s2);
608		s2.awaitSubscribe();
609		s1.awaitSubscribe();
610	<	assertEquals(p.submit(1), 1);
610	>	assertEquals(1, p.submit(1));
611		assertTrue(p.estimateMaximumLag() >= 1);
612		assertTrue(p.submit(2) >= 2);
613		assertTrue(p.estimateMaximumLag() >= 2);
#	Line 641 | Line 618 | public class SubmissionPublisherTest ext
618		p.submit(4);
619		p.close();
620		s2.awaitComplete();
621	<	assertEquals(s2.nexts, 4);
621	>	assertEquals(4, s2.nexts);
622		s1.awaitComplete();
623	<	assertEquals(s2.nexts, 4);
623	>	assertEquals(4, s2.nexts);
624		}
625
626		/**
#	Line 663 | Line 640 | public class SubmissionPublisherTest ext
640		p.close();
641		s2.awaitComplete();
642		s1.awaitComplete();
643	<	assertEquals(s2.nexts, 20);
644	<	assertEquals(s2.completes, 1);
645	<	assertEquals(s1.nexts, 20);
646	<	assertEquals(s1.completes, 1);
643	>	assertEquals(20, s2.nexts);
644	>	assertEquals(1, s2.completes);
645	>	assertEquals(20, s1.nexts);
646	>	assertEquals(1, s1.completes);
647		}
648	<
648	>
649		static boolean noopHandle(AtomicInteger count) {
650		count.getAndIncrement();
651		return false;
#	Line 681 | Line 658 | public class SubmissionPublisherTest ext
658		}
659
660		/**
661	<	* Offer to a publisher with no subscribers returns lag 0
661	>	* offer to a publisher with no subscribers returns lag 0
662		*/
663		public void testEmptyOffer() {
664		SubmissionPublisher<Integer> p = basicPublisher();
665	<	assertEquals(p.offer(1, null), 0);
665	>	assertEquals(0, p.offer(1, null));
666		}
667
668		/**
669	<	* Offer(null) throws NPE
669	>	* offer(null) throws NPE
670		*/
671		public void testNullOffer() {
672		SubmissionPublisher<Integer> p = basicPublisher();
673		try {
674		p.offer(null, null);
675	<	} catch (NullPointerException success) {
676	<	}
675	>	shouldThrow();
676	>	} catch (NullPointerException success) {}
677		}
678
679		/**
680	<	* Offer returns number of lagged items if not saturated
680	>	* offer returns number of lagged items if not saturated
681		*/
682		public void testLaggedOffer() {
683		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 720 | Line 697 | public class SubmissionPublisherTest ext
697		p.offer(4, null);
698		p.close();
699		s2.awaitComplete();
700	<	assertEquals(s2.nexts, 4);
700	>	assertEquals(4, s2.nexts);
701		s1.awaitComplete();
702	<	assertEquals(s2.nexts, 4);
702	>	assertEquals(4, s2.nexts);
703		}
704
705		/**
706	<	* Offer reports drops if saturated
706	>	* offer reports drops if saturated
707		*/
708		public void testDroppedOffer() {
709		SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
#	Line 754 | Line 731 | public class SubmissionPublisherTest ext
731		}
732
733		/**
734	<	* Offer invokes drop handler if saturated
734	>	* offer invokes drop handler if saturated
735		*/
736		public void testHandledDroppedOffer() {
737		AtomicInteger calls = new AtomicInteger();
#	Line 781 | Line 758 | public class SubmissionPublisherTest ext
758		assertTrue(calls.get() >= 4);
759		}
760
784	–
761		/**
762	<	* Offer succeeds if drop handler forces request
762	>	* offer succeeds if drop handler forces request
763		*/
764		public void testRecoveredHandledDroppedOffer() {
765		AtomicInteger calls = new AtomicInteger();
#	Line 805 | Line 781 | public class SubmissionPublisherTest ext
781		p.close();
782		s2.awaitComplete();
783		s1.awaitComplete();
784	<	assertEquals(s1.nexts + s2.nexts, n);
784	>	assertEquals(n, s1.nexts + s2.nexts);
785		assertTrue(calls.get() >= 2);
786		}
787
812	–
788		/**
789	<	* TimedOffer to a publisher with no subscribers returns lag 0
789	>	* Timed offer to a publisher with no subscribers returns lag 0
790		*/
791		public void testEmptyTimedOffer() {
792		SubmissionPublisher<Integer> p = basicPublisher();
793	<	assertEquals(p.offer(1, null), 0);
793	>	long startTime = System.nanoTime();
794	>	assertEquals(0, p.offer(1, LONG_DELAY_MS, MILLISECONDS, null));
795	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
796		}
797
798		/**
799	<	* Timed Offer with null item or TimeUnit throws NPE
799	>	* Timed offer with null item or TimeUnit throws NPE
800		*/
801		public void testNullTimedOffer() {
802		SubmissionPublisher<Integer> p = basicPublisher();
803	+	long startTime = System.nanoTime();
804		try {
805	<	p.offer(null, SHORT_DELAY_MS, MILLISECONDS, null);
806	<	} catch (NullPointerException success) {
807	<	}
805	>	p.offer(null, LONG_DELAY_MS, MILLISECONDS, null);
806	>	shouldThrow();
807	>	} catch (NullPointerException success) {}
808		try {
809	<	p.offer(1, SHORT_DELAY_MS, null, null);
810	<	} catch (NullPointerException success) {
811	<	}
809	>	p.offer(1, LONG_DELAY_MS, null, null);
810	>	shouldThrow();
811	>	} catch (NullPointerException success) {}
812	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
813		}
814
815		/**
816	<	* Timed Offer returns number of lagged items if not saturated
816	>	* Timed offer returns number of lagged items if not saturated
817		*/
818		public void testLaggedTimedOffer() {
819		SubmissionPublisher<Integer> p = basicPublisher();
#	Line 846 | Line 825 | public class SubmissionPublisherTest ext
825		p.subscribe(s2);
826		s2.awaitSubscribe();
827		s1.awaitSubscribe();
828	<	assertTrue(p.offer(1, SHORT_DELAY_MS, MILLISECONDS, null) >= 1);
829	<	assertTrue(p.offer(2, SHORT_DELAY_MS, MILLISECONDS, null) >= 2);
828	>	long startTime = System.nanoTime();
829	>	assertTrue(p.offer(1, LONG_DELAY_MS, MILLISECONDS, null) >= 1);
830	>	assertTrue(p.offer(2, LONG_DELAY_MS, MILLISECONDS, null) >= 2);
831		s1.sn.request(4);
832	<	assertTrue(p.offer(3, SHORT_DELAY_MS, MILLISECONDS, null) >= 3);
832	>	assertTrue(p.offer(3, LONG_DELAY_MS, MILLISECONDS, null) >= 3);
833		s2.sn.request(4);
834	<	p.offer(4, SHORT_DELAY_MS, MILLISECONDS, null);
834	>	p.offer(4, LONG_DELAY_MS, MILLISECONDS, null);
835		p.close();
836		s2.awaitComplete();
837	<	assertEquals(s2.nexts, 4);
837	>	assertEquals(4, s2.nexts);
838		s1.awaitComplete();
839	<	assertEquals(s2.nexts, 4);
839	>	assertEquals(4, s2.nexts);
840	>	assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS / 2);
841		}
842
843		/**
844	<	* Timed Offer reports drops if saturated
844	>	* Timed offer reports drops if saturated
845		*/
846		public void testDroppedTimedOffer() {
847		SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
#	Line 873 | Line 854 | public class SubmissionPublisherTest ext
854		p.subscribe(s2);
855		s2.awaitSubscribe();
856		s1.awaitSubscribe();
857	+	long delay = timeoutMillis();
858		for (int i = 1; i <= 4; ++i)
859	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
860	<	p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
861	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
859	>	assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
860	>	long startTime = System.nanoTime();
861	>	assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
862		s1.sn.request(64);
863	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
863	>	assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
864	>	// 2 * delay should elapse but check only 1 * delay to allow timer slop
865	>	assertTrue(millisElapsedSince(startTime) >= delay);
866		s2.sn.request(64);
867		p.close();
868		s2.awaitComplete();
#	Line 888 | Line 872 | public class SubmissionPublisherTest ext
872		}
873
874		/**
875	<	* Timed Offer invokes drop handler if saturated
875	>	* Timed offer invokes drop handler if saturated
876		*/
877		public void testHandledDroppedTimedOffer() {
878		AtomicInteger calls = new AtomicInteger();
#	Line 902 | Line 886 | public class SubmissionPublisherTest ext
886		p.subscribe(s2);
887		s2.awaitSubscribe();
888		s1.awaitSubscribe();
889	+	long delay = timeoutMillis();
890		for (int i = 1; i <= 4; ++i)
891	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
892	<	p.offer(5, (s, x) -> noopHandle(calls));
893	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
891	>	assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
892	>	long startTime = System.nanoTime();
893	>	assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
894		s1.sn.request(64);
895	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
895	>	assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
896	>	assertTrue(millisElapsedSince(startTime) >= delay);
897		s2.sn.request(64);
898		p.close();
899		s2.awaitComplete();
#	Line 916 | Line 902 | public class SubmissionPublisherTest ext
902		}
903
904		/**
905	<	* Timed Offer succeeds if drop handler forces request
905	>	* Timed offer succeeds if drop handler forces request
906		*/
907		public void testRecoveredHandledDroppedTimedOffer() {
908		AtomicInteger calls = new AtomicInteger();
#	Line 931 | Line 917 | public class SubmissionPublisherTest ext
917		s2.awaitSubscribe();
918		s1.awaitSubscribe();
919		int n = 0;
920	<	for (int i = 1; i <= 8; ++i) {
921	<	int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
920	>	long delay = timeoutMillis();
921	>	long startTime = System.nanoTime();
922	>	for (int i = 1; i <= 6; ++i) {
923	>	int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
924		n = n + 2 + (d < 0 ? d : 0);
925		}
926	+	assertTrue(millisElapsedSince(startTime) >= delay);
927		p.close();
928		s2.awaitComplete();
929		s1.awaitComplete();
930	<	assertEquals(s1.nexts + s2.nexts, n);
930	>	assertEquals(n, s1.nexts + s2.nexts);
931		assertTrue(calls.get() >= 2);
932		}
933
934	+	/**
935	+	* consume returns a CompletableFuture that is done when
936	+	* publisher completes
937	+	*/
938	+	public void testConsume() {
939	+	AtomicInteger sum = new AtomicInteger();
940	+	SubmissionPublisher<Integer> p = basicPublisher();
941	+	CompletableFuture<Void> f =
942	+	p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
943	+	int n = 20;
944	+	for (int i = 1; i <= n; ++i)
945	+	p.submit(i);
946	+	p.close();
947	+	f.join();
948	+	assertEquals((n * (n + 1)) / 2, sum.get());
949	+	}
950	+
951	+	/**
952	+	* consume(null) throws NPE
953	+	*/
954	+	public void testConsumeNPE() {
955	+	SubmissionPublisher<Integer> p = basicPublisher();
956	+	try {
957	+	CompletableFuture<Void> f = p.consume(null);
958	+	shouldThrow();
959	+	} catch (NullPointerException success) {}
960	+	}
961	+
962	+	/**
963	+	* consume eventually stops processing published items if cancelled
964	+	*/
965	+	public void testCancelledConsume() {
966	+	AtomicInteger count = new AtomicInteger();
967	+	SubmissionPublisher<Integer> p = basicPublisher();
968	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
969	+	f.cancel(true);
970	+	int n = 1000000; // arbitrary limit
971	+	for (int i = 1; i <= n; ++i)
972	+	p.submit(i);
973	+	assertTrue(count.get() < n);
974	+	}
975
976		}

Diff Legend

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