Merge pull request #63 from jroper/finer-grained-stages

Extracted more stages out

Author: jroper

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

@@ -90,7 +90,7 @@ public ProcessorBuilder<T, R> peek(Consumer<? super R> consumer) {
    * @return A new processor builder.
    */
   public ProcessorBuilder<T, R> filter(Predicate<? super R> predicate) {
-    return addStage(new Stage.Filter(() -> predicate));
+    return addStage(new Stage.Filter(predicate));
   }
 
   /**
@@ -178,20 +178,14 @@ public <S> ProcessorBuilder<T, S> flatMapIterable(Function<? super R, ? extends
    *
    * @param maxSize The maximum size of the returned stream.
    * @return A new processor builder.
+   * @throws IllegalArgumentException If {@code maxSize} is less than zero.
    */
   public ProcessorBuilder<T, R> limit(long maxSize) {
     if (maxSize < 0) {
       throw new IllegalArgumentException("Cannot limit a stream to less than zero elements.");
     }
-    else if (maxSize == 0) {
-      // todo this is perhaps not the desired behaviour - it means an element must be received before the stream will
-      // be completed. but then again, limiting a stream to have zero size is a strange thing to do, as running the
-      // stream in theory will then achieve nothing, so this edge case behavior probably isn't important to worry too
-      // much about.
-      return takeWhile(e -> false);
-    }
     else {
-      return addStage(new Stage.TakeWhile(() -> new Predicates.LimitPredicate<>(maxSize), true));
+      return addStage(new Stage.Limit(maxSize));
     }
   }
 
@@ -201,9 +195,13 @@ else if (maxSize == 0) {
    *
    * @param n The number of elements to discard.
    * @return A new processor builder.
+   * @throws IllegalArgumentException If {@code n} is less than zero.
    */
   public ProcessorBuilder<T, R> skip(long n) {
-    return addStage(new Stage.Filter(() -> new Predicates.SkipPredicate<>(n)));
+    if (n < 0) {
+      throw new IllegalArgumentException("Cannot skip less than zero elements");
+    }
+    return addStage(new Stage.Skip(n));
   }
 
   /**
@@ -215,7 +213,7 @@ public ProcessorBuilder<T, R> skip(long n) {
    * @return A new publisher builder.
    */
   public ProcessorBuilder<T, R> takeWhile(Predicate<? super R> predicate) {
-    return addStage(new Stage.TakeWhile(() -> predicate, false));
+    return addStage(new Stage.TakeWhile(predicate));
   }
 
   /**
@@ -229,7 +227,7 @@ public ProcessorBuilder<T, R> takeWhile(Predicate<? super R> predicate) {
    * @return A new processor builder.
    */
   public ProcessorBuilder<T, R> dropWhile(Predicate<? super R> predicate) {
-    return addStage(new Stage.Filter(() -> new Predicates.DropWhilePredicate<>(predicate)));
+    return addStage(new Stage.DropWhile(predicate));
   }
 
   /**

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

@@ -90,7 +90,7 @@ public PublisherBuilder<T> peek(Consumer<? super T> consumer) {
    * @return A new publisher builder.
    */
   public PublisherBuilder<T> filter(Predicate<? super T> predicate) {
-    return addStage(new Stage.Filter(() -> predicate));
+    return addStage(new Stage.Filter(predicate));
   }
 
   /**
@@ -179,16 +179,14 @@ public <S> PublisherBuilder<S> flatMapIterable(Function<? super T, ? extends Ite
    *
    * @param maxSize The maximum size of the returned stream.
    * @return A new publisher builder.
+   * @throws IllegalArgumentException If {@code maxSize} is less than zero.
    */
   public PublisherBuilder<T> limit(long maxSize) {
     if (maxSize < 0) {
       throw new IllegalArgumentException("Cannot limit a stream to less than zero elements.");
     }
-    else if (maxSize == 0) {
-      return takeWhile(e -> false);
-    }
     else {
-      return addStage(new Stage.TakeWhile(() -> new Predicates.LimitPredicate<>(maxSize), true));
+      return addStage(new Stage.Limit(maxSize));
     }
   }
 
@@ -198,9 +196,13 @@ else if (maxSize == 0) {
    *
    * @param n The number of elements to discard.
    * @return A new publisher builder.
+   * @throws IllegalArgumentException If {@code n} is less than zero.
    */
   public PublisherBuilder<T> skip(long n) {
-    return addStage(new Stage.Filter(() -> new Predicates.SkipPredicate<>(n)));
+    if (n < 0) {
+      throw new IllegalArgumentException("Cannot skip less than zero elements");
+    }
+    return addStage(new Stage.Skip(n));
   }
 
   /**
@@ -212,7 +214,7 @@ public PublisherBuilder<T> skip(long n) {
    * @return A new publisher builder.
    */
   public PublisherBuilder<T> takeWhile(Predicate<? super T> predicate) {
-    return addStage(new Stage.TakeWhile(() -> predicate, false));
+    return addStage(new Stage.TakeWhile(predicate));
   }
 
   /**
@@ -226,7 +228,7 @@ public PublisherBuilder<T> takeWhile(Predicate<? super T> predicate) {
    * @return A new publisher builder.
    */
   public PublisherBuilder<T> dropWhile(Predicate<? super T> predicate) {
-    return addStage(new Stage.Filter(() -> new Predicates.DropWhilePredicate<>(predicate)));
+    return addStage(new Stage.DropWhile(predicate));
   }
 
   /**

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

@@ -28,7 +28,6 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
-import java.util.function.Supplier;
 import java.util.stream.Collector;
 
 /**
@@ -133,15 +132,40 @@ public Consumer<?> getConsumer() {
   /**
    * A filter stage.
    * <p>
-   * The given predicate should be supplied when the stream is first run, and then invoked on each element consumed.
-   * If it returns true, the element should be emitted.
+   * The given predicate should be invoked on each element consumed. If it returns true, the element should be
+   * emitted.
    * <p>
    * Any {@link RuntimeException} thrown by the predicate should be propagated down the stream as an error.
    */
   final class Filter implements Inlet, Outlet {
-    private final Supplier<Predicate<?>> predicate;
+    private final Predicate<?> predicate;
+
+    public Filter(Predicate<?> predicate) {
+      this.predicate = predicate;
+    }
+
+    /**
+     * The predicate.
+     *
+     * @return The predicate.
+     */
+    public Predicate<?> getPredicate() {
+      return predicate;
+    }
+  }
+
+  /**
+   * A drop while stage.
+   * <p>
+   * The given predicate should be invoked on each element consumed, until it returns true. Each element that it
+   * returns true for should be dropped, once it returns false, then all elements should be emitted.
+   * <p>
+   * Any {@link RuntimeException} thrown by the predicate should be propagated down the stream as an error.
+   */
+  final class DropWhile implements Inlet, Outlet {
+    private final Predicate<?> predicate;
 
-    public Filter(Supplier<Predicate<?>> predicate) {
+    public DropWhile(Predicate<?> predicate) {
       this.predicate = predicate;
     }
 
@@ -150,11 +174,55 @@ public Filter(Supplier<Predicate<?>> predicate) {
      *
      * @return The predicate.
      */
-    public Supplier<Predicate<?>> getPredicate() {
+    public Predicate<?> getPredicate() {
       return predicate;
     }
   }
 
+  /**
+   * A skip stage.
+   * <p>
+   * The first {@code skip} elements should be skipped, after that all elements should be emitted.
+   */
+  final class Skip implements Inlet, Outlet {
+    private final long skip;
+
+    public Skip(long skip) {
+      this.skip = skip;
+    }
+
+    /**
+     * The number of elements to skip.
+     *
+     * @return The number of elements to skip.
+     */
+    public long getSkip() {
+      return skip;
+    }
+  }
+
+  /**
+   * A limit stage.
+   * <p>
+   * Only {@code limit} elements should be emitted, once that many elements are emitted, the stream should be completed.
+   */
+  final class Limit implements Inlet, Outlet {
+    private final long limit;
+
+    public Limit(long limit) {
+      this.limit = limit;
+    }
+
+    /**
+     * The limit.
+     *
+     * @return The limit.
+     */
+    public long getLimit() {
+      return limit;
+    }
+  }
+
   /**
    * A stage returning a stream consisting of the distinct elements (according to {@link Object#equals(Object)}) of this
    * stream.
@@ -173,35 +241,25 @@ private Distinct() {
    * A take while stage.
    * <p>
    * The given predicate should be supplied when the stream is first run, and then invoked on each element consumed.
-   * When it returns true, the element should be emitted, when it returns false, the element should only be emitted if
-   * inclusive is true, and then the stream should be completed.
+   * When it returns true, the element should be emitted, when it returns false the stream should be completed.
    * <p>
    * Any {@link RuntimeException} thrown by the predicate should be propagated down the stream as an error.
    */
   final class TakeWhile implements Inlet, Outlet {
-    private final Supplier<Predicate<?>> predicate;
-    private final boolean inclusive;
+    private final Predicate<?> predicate;
 
-    public TakeWhile(Supplier<Predicate<?>> predicate, boolean inclusive) {
+    public TakeWhile(Predicate<?> predicate) {
       this.predicate = predicate;
-      this.inclusive = inclusive;
     }
 
     /**
      * The predicate.
      *
      * @return The predicate.
      */
-    public Supplier<Predicate<?>> getPredicate() {
+    public Predicate<?> getPredicate() {
       return predicate;
     }
-
-    /**
-     * Whether the element that this returns false on should be emitted or not.
-     */
-    public boolean isInclusive() {
-      return inclusive;
-    }
   }
 
   /**