Massive overhaul of streams TCK

* Introduced two different types of TCK tests, tests that test the API
  which don't need an SPI to run (for testing clean room implementations
  that don't depend on the API artifact that we produce), as well as the
  existing SPI tests, which test the engines implementation of each
  stage.
* Added a lot more tests to each stage, testing things like cancel
  propagation, error propagation, etc.
* Made a lot more things a lot stricter, for example, there's now a lot
  of null checking done, the of stage must ensure all errors are carried
  by the returned completion stage, even if an error is thrown when
  creating the iterator.

Author: jroper

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/InternalStages.java

@@ -23,7 +23,10 @@
 
 /**
  * Internal stages, used to capture the graph while being built, but never passed to a
- * {@link org.eclipse.microprofile.reactive.streams.spi.ReactiveStreamsEngine}.
+ * {@link org.eclipse.microprofile.reactive.streams.spi.ReactiveStreamsEngine}. These exist for performance reasons,
+ * allowing the builder to hold the graph as an immutable linked tree where multiple stages can be appended in constant
+ * time, rather than needing to copy an array each time. However, when it comes to building the graph, it is first
+ * flattened out to an array, removing any of the internal stages that held nested stages, etc.
  */
 class InternalStages {
 

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/ProcessorBuilder.java

@@ -27,10 +27,10 @@
 import org.reactivestreams.Subscriber;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
 import java.util.function.BinaryOperator;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -242,6 +242,7 @@ public ProcessorBuilder<T, R> dropWhile(Predicate<? super R> predicate) {
      * @return A new subscriber builder.
      */
     public SubscriberBuilder<T, Void> forEach(Consumer<? super R> action) {
+        Objects.requireNonNull(action, "Action must not be null");
         return collect(Collector.<R, Void, Void>of(
             () -> null,
             (n, r) -> action.accept(r),
@@ -303,24 +304,6 @@ public SubscriberBuilder<T, Optional<R>> reduce(BinaryOperator<R> accumulator) {
         return addTerminalStage(new Stage.Collect(Reductions.reduce(accumulator)));
     }
 
-    /**
-     * Perform a reduction on the elements of this stream, using the provided identity value, accumulation function and
-     * combiner function.
-     * <p>
-     * The result of the reduction is returned in the {@link CompletionSubscriber}.
-     *
-     * @param identity    The identity value.
-     * @param accumulator The accumulator function.
-     * @param combiner    The combiner function.
-     * @return A new subscriber builder.
-     */
-    public <S> SubscriberBuilder<T, S> reduce(S identity,
-                                              BiFunction<S, ? super R, S> accumulator,
-                                              BinaryOperator<S> combiner) {
-
-        return addTerminalStage(new Stage.Collect(Reductions.reduce(identity, accumulator, combiner)));
-    }
-
     /**
      * Collect the elements emitted by this processor builder using the given {@link Collector}.
      * <p>
@@ -536,6 +519,7 @@ public Processor<T, R> buildRs() {
      * @return A {@link Processor} that will run this stream.
      */
     public Processor<T, R> buildRs(ReactiveStreamsEngine engine) {
+        Objects.requireNonNull(engine, "Engine must not be null");
         return engine.buildProcessor(toGraph());
     }
 

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/PublisherBuilder.java

@@ -27,10 +27,10 @@
 import org.reactivestreams.Subscriber;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
-import java.util.function.BiFunction;
 import java.util.function.BinaryOperator;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -241,6 +241,7 @@ public PublisherBuilder<T> dropWhile(Predicate<? super T> predicate) {
      * @return A new completion builder.
      */
     public CompletionRunner<Void> forEach(Consumer<? super T> action) {
+        Objects.requireNonNull(action, "Action must not be null");
         return collect(Collector.<T, Void, Void>of(
             () -> null,
             (n, t) -> action.accept(t),
@@ -301,24 +302,6 @@ public CompletionRunner<Optional<T>> reduce(BinaryOperator<T> accumulator) {
         return addTerminalStage(new Stage.Collect(Reductions.reduce(accumulator)));
     }
 
-    /**
-     * Perform a reduction on the elements of this stream, using the provided identity value, accumulation function and
-     * combiner function.
-     * <p>
-     * The result of the reduction is returned in the {@link CompletionStage}.
-     *
-     * @param identity    The identity value.
-     * @param accumulator The accumulator function.
-     * @param combiner    The combiner function.
-     * @return A new completion builder.
-     */
-    public <S> CompletionRunner<S> reduce(S identity,
-                                          BiFunction<S, ? super T, S> accumulator,
-                                          BinaryOperator<S> combiner) {
-
-        return addTerminalStage(new Stage.Collect(Reductions.reduce(identity, accumulator, combiner)));
-    }
-
     /**
      * Find the first element emitted by the {@link Publisher}, and return it in a
      * {@link CompletionStage}.
@@ -390,6 +373,7 @@ public CompletionRunner<Void> to(Subscriber<T> subscriber) {
      * @return A {@link CompletionRunner} that completes when the stream completes.
      */
     public <R> CompletionRunner<R> to(SubscriberBuilder<T, R> subscriber) {
+        Objects.requireNonNull(subscriber, "Subscriber must not be null");
         return addTerminalStage(new InternalStages.Nested(subscriber.getGraphBuilder()));
     }
 
@@ -536,6 +520,7 @@ public Publisher<T> buildRs() {
      * @return A {@link Publisher} that will run this stream.
      */
     public Publisher<T> buildRs(ReactiveStreamsEngine engine) {
+        Objects.requireNonNull(engine, "Engine must not be null");
         return engine.buildPublisher(toGraph());
     }
 

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/ReactiveStreams.java

@@ -26,6 +26,7 @@
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Objects;
 import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 import java.util.stream.Stream;
@@ -163,6 +164,7 @@ public static <T> SubscriberBuilder<T, Void> fromSubscriber(Subscriber<? extends
      * @return A publisher builder.
      */
     public static <T> PublisherBuilder<T> iterate(T seed, UnaryOperator<T> f) {
+        Objects.requireNonNull(f, "Operator must not be null");
         return fromIterable(() -> Stream.iterate(seed, f).iterator());
     }
 
@@ -174,6 +176,7 @@ public static <T> PublisherBuilder<T> iterate(T seed, UnaryOperator<T> f) {
      * @return A publisher builder.
      */
     public static <T> PublisherBuilder<T> generate(Supplier<? extends T> s) {
+        Objects.requireNonNull(s, "Supplier must not be null");
         return fromIterable(() -> Stream.<T>generate((Supplier) s).iterator());
     }
 

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/Reductions.java

@@ -19,8 +19,8 @@
 
 package org.eclipse.microprofile.reactive.streams;
 
+import java.util.Objects;
 import java.util.Optional;
-import java.util.function.BiFunction;
 import java.util.function.BinaryOperator;
 import java.util.stream.Collector;
 
@@ -33,7 +33,7 @@ private Reductions() {
     }
 
     static <T> Collector<T, ?, Optional<T>> reduce(BinaryOperator<T> reducer) {
-
+        Objects.requireNonNull(reducer, "Reduction function must not be null");
         return Collector.of(Reduction<T>::new,
             (r, t) -> {
                 if (r.value == null) {
@@ -59,25 +59,14 @@ else if (s.value != null) {
     }
 
     static <T> Collector<T, ?, T> reduce(T identity, BinaryOperator<T> reducer) {
-
+        Objects.requireNonNull(reducer, "Reduction function must not be null");
         return Collector.of(() -> new Reduction<>(identity),
             (r, t) -> r.value = reducer.apply(r.value, t),
             (r, s) -> r.replace(reducer.apply(r.value, s.value)),
             r -> r.value
         );
     }
 
-    static <T, S> Collector<T, ?, S> reduce(S identity,
-                                            BiFunction<S, ? super T, S> accumulator,
-                                            BinaryOperator<S> combiner) {
-
-        return Collector.of(() -> new Reduction<>(identity),
-            (r, t) -> r.value = accumulator.apply(r.value, t),
-            (r, s) -> r.replace(combiner.apply(r.value, s.value)),
-            r -> r.value
-        );
-    }
-
     private static class Reduction<T> {
         private T value;
 

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/SubscriberBuilder.java

@@ -23,6 +23,8 @@
 import org.eclipse.microprofile.reactive.streams.spi.ReactiveStreamsEngine;
 import org.eclipse.microprofile.reactive.streams.spi.Stage;
 
+import java.util.Objects;
+
 /**
  * A builder for a {@link org.reactivestreams.Subscriber} and its result.
  * <p>
@@ -63,6 +65,7 @@ public CompletionSubscriber<T, R> build() {
      * @return A {@link CompletionSubscriber} that will run this stream.
      */
     public CompletionSubscriber<T, R> build(ReactiveStreamsEngine engine) {
+        Objects.requireNonNull(engine, "Engine must not be null");
         return engine.buildSubscriber(toGraph());
     }