Skip to content

Commit aa446de

Browse files
ktosoktoso
committed
+tck reactive-streams#198 allow 0-lenght publishers to be tested
1 parent 93415d1 commit aa446de

File tree

7 files changed

+126
-16
lines changed

7 files changed

+126
-16
lines changed

‎examples/src/test/java/org/reactivestreams/example/unicast/AsyncSubscriberTest.java

+7-9
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,20 @@
11
package org.reactivestreams.example.unicast;
22

3-
import java.util.Collections;
4-
import java.util.Iterator;
53
import org.reactivestreams.Publisher;
64
import org.reactivestreams.Subscriber;
75
import org.reactivestreams.tck.SubscriberBlackboxVerification;
86
import org.reactivestreams.tck.TestEnvironment;
7+
import org.testng.annotations.AfterClass;
8+
import org.testng.annotations.BeforeClass;
99
import org.testng.annotations.Test;
10-
import static org.testng.Assert.assertEquals;
1110

12-
import org.testng.annotations.BeforeClass;
13-
import org.testng.annotations.AfterClass;
14-
import java.util.concurrent.Executors;
15-
import java.util.concurrent.ExecutorService;
1611
import java.util.concurrent.CountDownLatch;
17-
import java.util.concurrent.atomic.AtomicLong;
12+
import java.util.concurrent.ExecutorService;
13+
import java.util.concurrent.Executors;
1814
import java.util.concurrent.TimeUnit;
15+
import java.util.concurrent.atomic.AtomicLong;
16+
17+
import static org.testng.Assert.assertEquals;
1918

2019
@Test // Must be here for TestNG to find and run this, do not remove
2120
public class AsyncSubscriberTest extends SubscriberBlackboxVerification<Integer> {
@@ -31,7 +30,6 @@ public AsyncSubscriberTest() {
3130

3231
@Override public Subscriber<Integer> createSubscriber() {
3332
return new AsyncSubscriber<Integer>(e) {
34-
private long acc;
3533
@Override protected boolean whenNext(final Integer element) {
3634
return true;
3735
}

‎examples/src/test/java/org/reactivestreams/example/unicast/IterablePublisherTest.java

+3-5
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,14 @@
11
package org.reactivestreams.example.unicast;
22

3-
import java.util.Collections;
4-
import java.util.Iterator;
53
import org.reactivestreams.Publisher;
64
import org.reactivestreams.tck.PublisherVerification;
75
import org.reactivestreams.tck.TestEnvironment;
6+
import org.testng.annotations.AfterClass;
7+
import org.testng.annotations.BeforeClass;
88
import org.testng.annotations.Test;
99

10-
import org.testng.annotations.BeforeClass;
11-
import org.testng.annotations.AfterClass;
12-
import java.util.concurrent.Executors;
1310
import java.util.concurrent.ExecutorService;
11+
import java.util.concurrent.Executors;
1412

1513
@Test // Must be here for TestNG to find and run this, do not remove
1614
public class IterablePublisherTest extends PublisherVerification<Integer> {

‎tck/src/main/java/org/reactivestreams/tck/IdentityProcessorVerification.java

+5
Original file line numberDiff line numberDiff line change
@@ -238,6 +238,11 @@ public void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates() th
238238
publisherVerification.required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates();
239239
}
240240

241+
@Override @Test
242+
public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable {
243+
publisherVerification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete();
244+
}
245+
241246
@Override @Test
242247
public void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable {
243248
publisherVerification.untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled();

‎tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java

+20-1
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ Optional<T> requestNextElementOrEndOfStream(Publisher<T> pub, ManualSubscriber<T
135135

136136
@Override @Test
137137
public void required_validate_maxElementsFromPublisher() throws Exception {
138-
assertTrue(maxElementsFromPublisher() > 0, "maxElementsFromPublisher MUST return a number > 0");
138+
assertTrue(maxElementsFromPublisher() >= 0, "maxElementsFromPublisher MUST return a number >= 0");
139139
}
140140

141141
@Override @Test
@@ -339,6 +339,25 @@ public void run(Publisher<T> pub) throws Throwable {
339339
});
340340
}
341341

342+
// Verifies rule: https://github.com/reactive-streams/reactive-streams#1.5
343+
@Override @Test
344+
public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable {
345+
final boolean publisherIsEmptyStream = maxElementsFromPublisher() == 0;
346+
if (publisherIsEmptyStream) {
347+
activePublisherTest(0, true, new PublisherTestRun<T>() {
348+
@Override
349+
public void run(Publisher<T> pub) throws Throwable {
350+
ManualSubscriber<T> sub = env.newManualSubscriber(pub);
351+
sub.request(1);
352+
sub.expectCompletion();
353+
sub.expectNone();
354+
}
355+
});
356+
} else {
357+
notVerified("This test is specific for 0-lenght streams");
358+
}
359+
}
360+
342361
// Verifies rule: https://github.com/reactive-streams/reactive-streams#1.6
343362
@Override @Test
344363
public void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable {

‎tck/src/main/java/org/reactivestreams/tck/support/PublisherVerificationRules.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
package org.reactivestreams.tck.support;
22

33

4-
54
/**
65
* Internal TCK use only.
76
* Add / Remove tests for PublisherVerification here to make sure that they arre added/removed in the other places.
@@ -16,6 +15,7 @@ public interface PublisherVerificationRules {
1615
void stochastic_spec103_mustSignalOnMethodsSequentially() throws Throwable;
1716
void optional_spec104_mustSignalOnErrorWhenFails() throws Throwable;
1817
void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates() throws Throwable;
18+
void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable;
1919
void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable;
2020
void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled() throws Throwable;
2121
void untested_spec107_mustNotEmitFurtherSignalsOnceOnErrorHasBeenSignalled() throws Throwable;

‎tck/src/test/java/org/reactivestreams/tck/EmptyLazyPublisherTest.java

+45
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
package org.reactivestreams.tck;
2+
3+
import org.reactivestreams.Publisher;
4+
import org.reactivestreams.Subscriber;
5+
import org.reactivestreams.Subscription;
6+
import org.testng.annotations.Test;
7+
8+
@Test
9+
public class EmptyLazyPublisherTest extends PublisherVerification<Integer> {
10+
11+
public EmptyLazyPublisherTest() {
12+
// TODO these params will be removed once https://github.com/reactive-streams/reactive-streams/pull/193 is merged
13+
super(new TestEnvironment(300), 500);
14+
}
15+
16+
@Override
17+
public Publisher<Integer> createPublisher(long elements) {
18+
return new Publisher<Integer>() {
19+
@Override
20+
public void subscribe(final Subscriber<? super Integer> s) {
21+
s.onSubscribe(new Subscription() {
22+
@Override
23+
public void request(long n) {
24+
s.onComplete();
25+
}
26+
27+
@Override
28+
public void cancel() {
29+
// noop
30+
}
31+
});
32+
}
33+
};
34+
}
35+
36+
@Override
37+
public Publisher<Integer> createErrorStatePublisher() {
38+
return null;
39+
}
40+
41+
@Override
42+
public long maxElementsFromPublisher() {
43+
return 0;
44+
}
45+
}

‎tck/src/test/java/org/reactivestreams/tck/PublisherVerificationTest.java

+45
Original file line numberDiff line numberDiff line change
@@ -187,6 +187,51 @@ public void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates_shou
187187
}, "Expected end-of-stream but got element [3]");
188188
}
189189

190+
@Test
191+
public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete_shouldNotAllowEagerOnComplete() throws Throwable {
192+
final Publisher<Integer> illegalEmptyEagerOnCompletePublisher = new Publisher<Integer>() {
193+
@Override public void subscribe(final Subscriber<? super Integer> s) {
194+
s.onComplete();
195+
}
196+
};
197+
198+
requireTestFailure(new ThrowingRunnable() {
199+
@Override public void run() throws Throwable {
200+
PublisherVerification<Integer> verification = new PublisherVerification<Integer>(newTestEnvironment(), GC_TIMEOUT_MILLIS) {
201+
@Override public Publisher<Integer> createPublisher(long elements) {
202+
return illegalEmptyEagerOnCompletePublisher;
203+
}
204+
205+
@Override public long maxElementsFromPublisher() {
206+
return 0;
207+
}
208+
209+
@Override public Publisher<Integer> createErrorStatePublisher() {
210+
return null;
211+
}
212+
};
213+
214+
verification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete();
215+
}
216+
}, "Subscriber::onComplete() called before Subscriber::onSubscribe");
217+
}
218+
219+
@Test
220+
public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete_shouldBeSkippedForNonEmptyPublisher() throws Throwable {
221+
final Publisher<Integer> failingPublisher = new Publisher<Integer>() {
222+
@Override public void subscribe(final Subscriber<? super Integer> s) {
223+
throw new TestException();
224+
}
225+
};
226+
227+
requireTestSkip(new ThrowingRunnable() {
228+
@Override
229+
public void run() throws Throwable {
230+
customPublisherVerification(failingPublisher).optional_spec105_emptyStreamMustTerminateBySignallingOnComplete();
231+
}
232+
}, "This test is specific for 0-lenght streams");
233+
}
234+
190235
@Test
191236
public void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled_shouldFailForNotCompletingPublisher() throws Throwable {
192237
requireTestFailure(new ThrowingRunnable() {

0 commit comments

Comments
 (0)