+tck reactive-streams#198 allow 0-length publishers to be tested

Author: ktoso

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

@@ -4,16 +4,17 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.tck.SubscriberBlackboxVerification;
 import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
-import static org.testng.Assert.assertEquals;
 
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.AfterClass;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.testng.Assert.assertEquals;
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class AsyncSubscriberTest extends SubscriberBlackboxVerification<Integer> {

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

@@ -1,16 +1,14 @@
 package org.reactivestreams.example.unicast;
 
-import java.util.Collections;
-import java.util.Iterator;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.tck.PublisherVerification;
 import org.reactivestreams.tck.TestEnvironment;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.AfterClass;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 @Test // Must be here for TestNG to find and run this, do not remove
 public class IterablePublisherTest extends PublisherVerification<Integer> {

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

@@ -241,6 +241,11 @@ public void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates() th
     publisherVerification.required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates();
   }
 
+  @Override @Test
+  public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable {
+    publisherVerification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete();
+  }
+
   @Override @Test
   public void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable {
     publisherVerification.untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled();

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

@@ -179,7 +179,7 @@ Optional<T> requestNextElementOrEndOfStream(Publisher<T> pub, ManualSubscriber<T
 
   @Override @Test
   public void required_validate_maxElementsFromPublisher() throws Exception {
-    assertTrue(maxElementsFromPublisher() > 0, "maxElementsFromPublisher MUST return a number > 0");
+    assertTrue(maxElementsFromPublisher() >= 0, "maxElementsFromPublisher MUST return a number >= 0");
   }
 
   @Override @Test
@@ -383,6 +383,20 @@ public void run(Publisher<T> pub) throws Throwable {
     });
   }
 
+  // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.5
+  @Override @Test
+  public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable {
+    activePublisherTest(0, true, new PublisherTestRun<T>() {
+      @Override
+      public void run(Publisher<T> pub) throws Throwable {
+        ManualSubscriber<T> sub = env.newManualSubscriber(pub);
+        sub.request(1);
+        sub.expectCompletion();
+        sub.expectNone();
+      }
+    });
+  }
+
   // Verifies rule: https://github.com/reactive-streams/reactive-streams#1.6
   @Override @Test
   public void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable {

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

@@ -1,7 +1,6 @@
 package org.reactivestreams.tck.support;
 
 
-
 /**
  * Internal TCK use only.
  * 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 {
   void stochastic_spec103_mustSignalOnMethodsSequentially() throws Throwable;
   void optional_spec104_mustSignalOnErrorWhenFails() throws Throwable;
   void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates() throws Throwable;
+  void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete() throws Throwable;
   void untested_spec106_mustConsiderSubscriptionCancelledAfterOnErrorOrOnCompleteHasBeenCalled() throws Throwable;
   void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled() throws Throwable;
   void untested_spec107_mustNotEmitFurtherSignalsOnceOnErrorHasBeenSignalled() throws Throwable;

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

@@ -0,0 +1,42 @@
+package org.reactivestreams.tck;
+
+import org.reactivestreams.example.unicast.AsyncIterablePublisher;
+import org.reactivestreams.Publisher;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Test
+public class EmptyLazyPublisherTest extends PublisherVerification<Integer> {
+
+  private ExecutorService ex;
+
+  public EmptyLazyPublisherTest() {
+    super(new TestEnvironment());
+  }
+
+  @BeforeClass
+  void before() { ex = Executors.newFixedThreadPool(4); }
+
+  @AfterClass
+  void after() { if (ex != null) ex.shutdown(); }
+
+  @Override
+  public Publisher<Integer> createPublisher(long elements) {
+    return new AsyncIterablePublisher<Integer>(Collections.<Integer>emptyList(), ex);
+  }
+
+  @Override
+  public Publisher<Integer> createErrorStatePublisher() {
+    return null;
+  }
+
+  @Override
+  public long maxElementsFromPublisher() {
+    return 0;
+  }
+}

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

@@ -184,6 +184,35 @@ public void required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates_shou
     }, "Expected end-of-stream but got element [3]");
   }
 
+  @Test
+  public void optional_spec105_emptyStreamMustTerminateBySignallingOnComplete_shouldNotAllowEagerOnComplete() throws Throwable {
+    final Publisher<Integer> illegalEmptyEagerOnCompletePublisher = new Publisher<Integer>() {
+      @Override public void subscribe(final Subscriber<? super Integer> s) {
+        s.onComplete();
+      }
+    };
+
+    requireTestFailure(new ThrowingRunnable() {
+      @Override public void run() throws Throwable {
+        PublisherVerification<Integer> verification = new PublisherVerification<Integer>(newTestEnvironment()) {
+          @Override public Publisher<Integer> createPublisher(long elements) {
+            return illegalEmptyEagerOnCompletePublisher;
+          }
+
+          @Override public long maxElementsFromPublisher() {
+            return 0; // it is an "empty" Publisher
+          }
+
+          @Override public Publisher<Integer> createErrorStatePublisher() {
+            return null;
+          }
+        };
+
+        verification.optional_spec105_emptyStreamMustTerminateBySignallingOnComplete();
+      }
+    }, "Subscriber::onComplete() called before Subscriber::onSubscribe");
+  }
+
   @Test
   public void required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled_shouldFailForNotCompletingPublisher() throws Throwable {
     requireTestFailure(new ThrowingRunnable() {