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

Author: ktoso

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

@@ -1,21 +1,20 @@
 package org.reactivestreams.example.unicast;
 
-import java.util.Collections;
-import java.util.Iterator;
 import org.reactivestreams.Publisher;
 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> {
@@ -31,7 +30,6 @@ public AsyncSubscriberTest() {
 
   @Override public Subscriber<Integer> createSubscriber() {
     return new AsyncSubscriber<Integer>(e) {
-      private long acc;
       @Override protected boolean whenNext(final Integer element) {
         return true;
       }

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

@@ -238,6 +238,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

@@ -135,7 +135,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
@@ -339,6 +339,25 @@ 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 {
+    final boolean publisherIsEmptyStream = maxElementsFromPublisher() == 0;
+    if (publisherIsEmptyStream) {
+      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();
+        }
+      });
+    } else {
+      notVerified("This test is specific for 0-lenght streams");
+    }
+  }
+
   // 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,45 @@
+package org.reactivestreams.tck;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.testng.annotations.Test;
+
+@Test
+public class EmptyLazyPublisherTest extends PublisherVerification<Integer> {
+
+  public EmptyLazyPublisherTest() {
+    // TODO these params will be removed once https://github.com/reactive-streams/reactive-streams/pull/193 is merged
+    super(new TestEnvironment(300), 500);
+  }
+
+  @Override
+  public Publisher<Integer> createPublisher(long elements) {
+    return new Publisher<Integer>() {
+      @Override
+      public void subscribe(final Subscriber<? super Integer> s) {
+        s.onSubscribe(new Subscription() {
+          @Override
+          public void request(long n) {
+            s.onComplete();
+          }
+
+          @Override
+          public void cancel() {
+            // noop
+          }
+        });
+      }
+    };
+  }
+
+  @Override
+  public Publisher<Integer> createErrorStatePublisher() {
+    return null;
+  }
+
+  @Override
+  public long maxElementsFromPublisher() {
+    return 0;
+  }
+}