Merge pull request #4 from jroper/rework-zerodep

Significant rework of zerodep implementation

Author: jroper

akka/src/main/java/com/lightbend/microprofile/reactive/streams/akka/AkkaEngine.java

@@ -4,6 +4,7 @@
 
 package com.lightbend.microprofile.reactive.streams.akka;
 
+import akka.Done;
 import akka.NotUsed;
 import akka.stream.Attributes;
 import akka.stream.Materializer;
@@ -256,6 +257,9 @@ else if (size == 1) {
       Predicate<Object> predicate = (Predicate) stage.getPredicate();
       return flow.dropWhile(predicate::test);
     });
+    addFlowStage(Stage.OnComplete.class, (flow, stage) -> flow.via(TerminationPeeker.onComplete(stage.getAction())));
+    addFlowStage(Stage.OnError.class, (flow, stage) -> flow.via(TerminationPeeker.onError(stage.getConsumer())));
+    addFlowStage(Stage.OnTerminate.class, (flow, stage) -> flow.via(TerminationPeeker.onTerminate(stage.getAction())));
 
     // Sinks
     addSinkStage(Stage.FindFirst.class, stage -> Sink.headOption());

akka/src/main/java/com/lightbend/microprofile/reactive/streams/akka/TerminationPeeker.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package com.lightbend.microprofile.reactive.streams.akka;
+
+import akka.stream.Attributes;
+import akka.stream.FlowShape;
+import akka.stream.Inlet;
+import akka.stream.Outlet;
+import akka.stream.stage.AbstractInHandler;
+import akka.stream.stage.AbstractOutHandler;
+import akka.stream.stage.GraphStage;
+import akka.stream.stage.GraphStageLogic;
+import akka.stream.stage.GraphStageWithMaterializedValue;
+import scala.Tuple2;
+
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
+
+/**
+ * Differs from Flow.watchTermination in that termination callbacks are run in the stream, so if the callback experiences an error,
+ * the stream fails.
+ */
+class TerminationPeeker<T> extends GraphStage<FlowShape<T, T>> {
+  private final Inlet<T> in = Inlet.create("TerminationPeeker.in");
+  private final Outlet<T> out = Outlet.create("TerminationPeeker.out");
+
+  private final FlowShape<T, T> shape = FlowShape.of(in, out);
+
+  private final Runnable onCompleteCallback;
+  private final Consumer<Throwable> onErrorCallback;
+
+  private TerminationPeeker(Runnable onCompleteCallback, Consumer<Throwable> onErrorCallback) {
+    this.onCompleteCallback = onCompleteCallback;
+    this.onErrorCallback = onErrorCallback;
+  }
+
+  static <T> GraphStage<FlowShape<T, T>> onComplete(Runnable action) {
+    return new TerminationPeeker<>(action, t -> {});
+  }
+
+  static <T> GraphStage<FlowShape<T, T>> onTerminate(Runnable action) {
+    return new TerminationPeeker<>(action, t -> action.run());
+  }
+
+  static <T> GraphStage<FlowShape<T, T>> onError(Consumer<Throwable> consumer) {
+    return new TerminationPeeker<>(() -> {}, consumer);
+  }
+
+  @Override
+  public FlowShape<T, T> shape() {
+    return shape;
+  }
+
+  @Override
+  public GraphStageLogic createLogic(Attributes inheritedAttributes) {
+    return new GraphStageLogic(shape()) {
+      {
+        setHandler(in, new AbstractInHandler() {
+          @Override
+          public void onPush() throws Exception {
+            push(out, grab(in));
+          }
+
+          @Override
+          public void onUpstreamFinish() throws Exception {
+            onCompleteCallback.run();
+            complete(out);
+          }
+
+          @Override
+          public void onUpstreamFailure(Throwable ex) throws Exception {
+            onErrorCallback.accept(ex);
+            fail(out, ex);
+          }
+        });
+        setHandler(out, new AbstractOutHandler() {
+          @Override
+          public void onPull() throws Exception {
+            pull(in);
+          }
+
+          @Override
+          public void onDownstreamFinish() throws Exception {
+            onCompleteCallback.run();
+            cancel(in);
+          }
+        });
+      }
+
+    };
+  }
+}

akka/src/main/java/com/lightbend/microprofile/reactive/streams/akka/TerminationWatcher.java

@@ -12,6 +12,9 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
+/**
+ * Differs from Flow.watchTermination in that cancellation results in the completion stage failing with a cancellation exception.
+ */
 class TerminationWatcher<T> extends GraphStageWithMaterializedValue<FlowShape<T, T>, CompletionStage<Void>> {
   private final Inlet<T> in = Inlet.create("TerminationWatcher.in");
   private final Outlet<T> out = Outlet.create("TerminationWatcher.out");

bin/buildDeps.sh

@@ -1,9 +1,9 @@
 #!/bin/bash
 
 # This should be set to the commit hash that is being tracked. Needed even if TRACKING_PR is set.
-TRACKING_COMMIT="60f67ef"
+TRACKING_COMMIT="41f41e2"
 # To track a particular pull request, put it's number here, otherwise comment it out.
-TRACKING_PR="64"
+# TRACKING_PR="64"
 
 set -e
 
@@ -25,6 +25,4 @@ fi
 
 git checkout "${TRACKING_COMMIT}"
 
-cd streams
-
-mvn clean install -Dmaven.test.skip -Drat.skip=true -Dcheckstyle.skip=true -Dmaven.javadoc.skip=true -Dasciidoctor.skip=true
+mvn -am -pl streams/api,streams/tck clean install -Dmaven.test.skip -Drat.skip=true -Dcheckstyle.skip=true -Dmaven.javadoc.skip=true -Dasciidoctor.skip=true

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/BuiltGraph.java

@@ -43,7 +43,7 @@ class BuiltGraph implements Executor {
   private static final int DEFAULT_BUFFER_LOW_WATERMARK = 4;
 
   private final Executor mutex;
-  private final Deque<UnrolledSignal> unrolledSignals = new ArrayDeque<>();
+  private final Deque<Signal> signals = new ArrayDeque<>();
   private final Set<Port> ports = new LinkedHashSet<>();
   private final Set<GraphStage> stages = new LinkedHashSet<>();
 
@@ -197,10 +197,10 @@ private Builder buildGraph(Graph graph, Shape shape) {
             if (isSubscriber(stage)) {
               // We're connecting a publisher to a subscriber, don't create any port, just record what the current
               // publisher is.
-              if (previousStage instanceof Stage.PublisherStage) {
-                currentPublisher = ((Stage.PublisherStage) previousStage).getRsPublisher();
+              if (stage instanceof Stage.SubscriberStage) {
+                currentSubscriber = ((Stage.SubscriberStage) stage).getRsSubscriber();
               } else {
-                currentPublisher = ((Stage.ProcessorStage) previousStage).getRsProcessor();
+                currentSubscriber = ((Stage.ProcessorStage) stage).getRsProcessor();
               }
             } else {
               // We're connecting a publisher to an inlet, create a subscriber inlet for that.
@@ -373,7 +373,7 @@ private void addStage(Stage stage, StageInlet inlet, Publisher publisher, StageO
 
           addStage(new ConcatStage(BuiltGraph.this, firstInlet, secondInlet, outlet));
         } else if (stage instanceof Stage.PublisherStage) {
-          addStage(new ConnectorStage<>(BuiltGraph.this, ((Stage.PublisherStage) stage).getRsPublisher(), subscriber));
+          addStage(new ConnectorStage(BuiltGraph.this, ((Stage.PublisherStage) stage).getRsPublisher(), subscriber));
         } else if (stage instanceof Stage.Failed) {
           addStage(new FailedStage(BuiltGraph.this, outlet, ((Stage.Failed) stage).getError()));
         } else {
@@ -410,6 +410,12 @@ private void addStage(Stage stage, StageInlet inlet, Publisher publisher, StageO
           addStage(new DropWhileStage(BuiltGraph.this, inlet, outlet, ((Stage.DropWhile) stage).getPredicate()));
         } else if (stage instanceof Stage.Peek) {
           addStage(new PeekStage<>(BuiltGraph.this, inlet, outlet, ((Stage.Peek) stage).getConsumer()));
+        } else if (stage instanceof Stage.OnComplete) {
+          addStage(new OnCompleteStage<>(BuiltGraph.this, inlet, outlet, ((Stage.OnComplete) stage).getAction()));
+        } else if (stage instanceof Stage.OnError) {
+          addStage(new OnErrorStage<>(BuiltGraph.this, inlet, outlet, ((Stage.OnError) stage).getConsumer()));
+        } else if (stage instanceof Stage.OnTerminate) {
+          addStage(new OnTerminateStage<>(BuiltGraph.this, inlet, outlet, ((Stage.OnTerminate) stage).getAction()));
         } else {
           throw new UnsupportedStageException(stage);
         }
@@ -429,8 +435,6 @@ private void addStage(Stage stage, StageInlet inlet, Publisher publisher, StageO
             SubscriberInlet subscriberInlet = addPort(createSubscriberInlet());
             if (publisher != null) {
               addStage(new ConnectorStage(BuiltGraph.this, publisher, subscriberInlet));
-            } else {
-              firstSubscriber = subscriberInlet;
             }
             inlet = subscriberInlet;
           }
@@ -473,7 +477,7 @@ private boolean isPublisher(Stage stage) {
    * from all other signals on this graph. Any exceptions thrown by the command will cause the graph to be terminated
    * with a failure.
    * <p>
-   * Commands are also allowed to (synchronously) emit unrolled signals, by adding them to the unrolledSignals queue.
+   * Commands are also allowed to (synchronously) emit unrolled signals, by adding them to the signals queue.
    * Unrolled signals are used for breaking infinite recursion scenarios. This method will drain all unrolled signals
    * (including subsequent signals emitted by the unrolled signals themselves) after invocation of the command.
    *
@@ -488,14 +492,19 @@ public void execute(Runnable command) {
 
         // Now drain a maximum of 32 signals from the queue
         int signalsDrained = 0;
-        while (!unrolledSignals.isEmpty() && signalsDrained < 32) {
+        while (!signals.isEmpty() && signalsDrained < 32) {
           signalsDrained++;
-          unrolledSignals.poll().signal();
+          Signal signal = signals.removeFirst();
+          try {
+            signal.signal();
+          } catch (RuntimeException e) {
+            signal.signalFailed(e);
+          }
         }
 
         // If there were more than 32 unrolled signals, we resubmit
         // to the executor to allow us to receive external signals
-        if (!unrolledSignals.isEmpty()) {
+        if (!signals.isEmpty()) {
           execute(() -> {
           });
         }
@@ -504,7 +513,7 @@ public void execute(Runnable command) {
         // shut down the stream
         streamFailure(e);
         // Clear remaining signals
-        unrolledSignals.clear();
+        signals.clear();
       }
     });
   }
@@ -524,20 +533,16 @@ private void streamFailure(Throwable error) {
     // todo handle better
     error.printStackTrace();
     for (Port port : ports) {
-      try {
-        port.onStreamFailure(error);
-      } catch (RuntimeException e) {
-        // Ignore
-      }
+      port.onStreamFailure(error);
     }
     ports.clear();
   }
 
   /**
    * Enqueue a signal to be executed serially after the current signal processing finishes.
    */
-  void enqueueSignal(UnrolledSignal signal) {
-    unrolledSignals.add(signal);
+  void enqueueSignal(Signal signal) {
+    signals.add(signal);
   }
 
   /**
@@ -633,32 +638,3 @@ public void onUpstreamFailure(Throwable error) {
 
 }
 
-
-/**
- * An unrolled signal.
- * <p>
- * It is possible for stages to get into an infinite recursion, doing push/pulls between each other. This interface
- * allows them to unroll the recursion, by adding the signal to the unrolledSignals queue in this class, which then
- * gets executed after the first callback is executed.
- */
-interface UnrolledSignal {
-  void signal();
-}
-
-/**
- * A port, which may sit between two stages of this graph.
- */
-interface Port {
-  /**
-   * If an exception is thrown by the graph, or otherwise encountered, each port will be shut down in the order they
-   * were created, by invoking this. This method should implement any clean up associated with a port, if the port
-   * isn't already shut down.
-   */
-  void onStreamFailure(Throwable reason);
-
-  /**
-   * Verify that this port is ready to start receiving signals.
-   */
-  void verifyReady();
-}
-

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/CancelStage.java

@@ -23,9 +23,7 @@ class CancelStage extends GraphStage implements InletListener {
 
   @Override
   protected void postStart() {
-    if (!inlet.isClosed()) {
-      inlet.cancel();
-    }
+    inlet.cancel();
     result.complete(null);
   }
 

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/CollectStage.java

@@ -29,10 +29,7 @@ public CollectStage(BuiltGraph builtGraph, StageInlet<T> inlet,
 
   @Override
   protected void postStart() {
-    // It's possible that an earlier stage finished immediately, so check first
-    if (!inlet.isClosed()) {
-      inlet.pull();
-    }
+    inlet.pull();
   }
 
   @Override
@@ -43,7 +40,11 @@ public void onPush() {
 
   @Override
   public void onUpstreamFinish() {
-    result.complete(collector.finisher().apply(container));
+    try {
+      result.complete(collector.finisher().apply(container));
+    } catch (RuntimeException e) {
+      result.completeExceptionally(e);
+    }
     container = null;
   }
 

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/ConnectorStage.java

@@ -7,6 +7,8 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
+import java.util.Objects;
+
 /**
  * Connector stage. Does nothing but connects a publisher to a subscriber when the graph starts.
  */
@@ -16,8 +18,8 @@ public class ConnectorStage<T> extends GraphStage {
 
   public ConnectorStage(BuiltGraph builtGraph, Publisher<T> publisher, Subscriber<T> subscriber) {
     super(builtGraph);
-    this.publisher = publisher;
-    this.subscriber = subscriber;
+    this.publisher = Objects.requireNonNull(publisher);
+    this.subscriber = Objects.requireNonNull(subscriber);
   }
 
   @Override

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/FailedStage.java

@@ -21,9 +21,7 @@ public FailedStage(BuiltGraph builtGraph, StageOutlet<?> outlet, Throwable error
 
   @Override
   protected void postStart() {
-    if (!outlet.isClosed()) {
-      outlet.fail(error);
-    }
+    outlet.fail(error);
   }
 
   @Override

zerodep/src/main/java/com/lightbend/microprofile/reactive/streams/zerodep/FindFirstStage.java

@@ -22,9 +22,7 @@ class FindFirstStage<T> extends GraphStage implements InletListener {
 
   @Override
   protected void postStart() {
-    if (!inlet.isClosed()) {
-      inlet.pull();
-    }
+    inlet.pull();
   }
 
   @Override