Added support for the recover stages

Author: jroper

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

@@ -4,10 +4,11 @@
 
 package com.lightbend.microprofile.reactive.streams.akka;
 
-import akka.Done;
 import akka.NotUsed;
+import akka.japi.JavaPartialFunction;
 import akka.stream.Attributes;
 import akka.stream.Materializer;
+import akka.stream.SourceShape;
 import akka.stream.javadsl.*;
 import org.eclipse.microprofile.reactive.streams.CompletionRunner;
 import org.eclipse.microprofile.reactive.streams.CompletionSubscriber;
@@ -22,11 +23,7 @@
 import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
 
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
@@ -260,6 +257,26 @@ else if (size == 1) {
     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())));
+    addFlowStage(Stage.OnErrorResume.class, (flow, stage) -> {
+      Function<Throwable, Object> function = (Function) stage.getFunction();
+      return flow.recover(new JavaPartialFunction<Throwable, Object>() {
+        @Override
+        public Object apply(Throwable x, boolean isCheck) throws Exception {
+          if (isCheck) return null;
+          else return function.apply(x);
+        }
+      });
+    });
+    addFlowStage(Stage.OnErrorResumeWith.class, (flow, stage) -> {
+      Function<Throwable, Graph> function = (Function) stage.getFunction();
+      return flow.recoverWithRetries(1, new JavaPartialFunction<Throwable, akka.stream.Graph<SourceShape<Object>, NotUsed>>() {
+        @Override
+        public akka.stream.Graph<SourceShape<Object>, NotUsed> apply(Throwable x, boolean isCheck) throws Exception {
+          if (isCheck) return null;
+          else return buildSource(function.apply(x));
+        }
+      });
+    });
 
     // Sinks
     addSinkStage(Stage.FindFirst.class, stage -> Sink.headOption());

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="41f41e2"
+TRACKING_COMMIT="886a91d5"
 # To track a particular pull request, put it's number here, otherwise comment it out.
-# TRACKING_PR="64"
+# TRACKING_PR="67"
 
 set -e
 

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

@@ -416,6 +416,10 @@ private void addStage(Stage stage, StageInlet inlet, Publisher publisher, StageO
           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 if (stage instanceof Stage.OnErrorResume) {
+          addStage(new OnErrorResumeStage(BuiltGraph.this, inlet, outlet, ((Stage.OnErrorResume) stage).getFunction()));
+        } else if (stage instanceof Stage.OnErrorResumeWith) {
+          addStage(new OnErrorResumeWithStage(BuiltGraph.this, inlet, outlet, ((Stage.OnErrorResumeWith) stage).getFunction()));
         } else {
           throw new UnsupportedStageException(stage);
         }

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

@@ -23,8 +23,12 @@ public OfStage(BuiltGraph builtGraph, StageOutlet<T> outlet, Iterable<T> element
 
   @Override
   protected void postStart() {
-    if (!elements.hasNext()) {
-      outlet.complete();
+    try {
+      if (!elements.hasNext()) {
+        outlet.complete();
+      }
+    } catch (Exception e) {
+      outlet.fail(e);
     }
   }
 

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

@@ -0,0 +1,65 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+ */
+package com.lightbend.microprofile.reactive.streams.zerodep;
+
+import java.util.function.Function;
+
+class OnErrorResumeStage<T> extends GraphStage implements InletListener, OutletListener {
+
+  private final StageInlet<T> inlet;
+  private final StageOutlet<T> outlet;
+  private final Function<Throwable, T> function;
+
+  private T recoveredElement;
+
+  public OnErrorResumeStage(BuiltGraph builtGraph, StageInlet<T> inlet, StageOutlet<T> outlet, Function<Throwable, T> function) {
+    super(builtGraph);
+    this.inlet = inlet;
+    this.outlet = outlet;
+    this.function = function;
+
+    inlet.setListener(this);
+    outlet.setListener(this);
+  }
+
+  @Override
+  public void onPush() {
+    outlet.push(inlet.grab());
+  }
+
+  @Override
+  public void onUpstreamFinish() {
+    outlet.complete();
+  }
+
+  @Override
+  public void onUpstreamFailure(Throwable error) {
+    try {
+      T element = function.apply(error);
+      if (outlet.isAvailable()) {
+        outlet.push(element);
+        outlet.complete();
+      } else {
+        recoveredElement = element;
+      }
+    } catch (Exception e) {
+      outlet.fail(e);
+    }
+  }
+
+  @Override
+  public void onPull() {
+    if (recoveredElement != null) {
+      outlet.push(recoveredElement);
+      outlet.complete();
+    } else {
+      inlet.pull();
+    }
+  }
+
+  @Override
+  public void onDownstreamFinish() {
+    inlet.cancel();
+  }
+}

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+ */
+package com.lightbend.microprofile.reactive.streams.zerodep;
+
+import org.eclipse.microprofile.reactive.streams.spi.Graph;
+
+import java.util.function.Function;
+
+class OnErrorResumeWithStage<T> extends GraphStage implements InletListener {
+
+  private final StageInlet<T> inlet;
+  private final StageOutlet<T> outlet;
+  private final Function<Throwable, Graph> function;
+
+  public OnErrorResumeWithStage(BuiltGraph builtGraph, StageInlet<T> inlet, StageOutlet<T> outlet, Function<Throwable, Graph> function) {
+    super(builtGraph);
+    this.inlet = inlet;
+    this.outlet = outlet;
+    this.function = function;
+
+    inlet.setListener(this);
+    outlet.forwardTo(inlet);
+  }
+
+  @Override
+  public void onPush() {
+    outlet.push(inlet.grab());
+  }
+
+  @Override
+  public void onUpstreamFinish() {
+    outlet.complete();
+  }
+
+  @Override
+  public void onUpstreamFailure(Throwable error) {
+    try {
+      Graph graph = function.apply(error);
+
+      BuiltGraph.SubStageInlet<T> newInlet = createSubInlet(graph);
+
+      // Wire the new inlet directly to/from the outlet, this stage no longer has any involvement in the stream.
+      newInlet.forwardTo(outlet);
+      outlet.forwardTo(newInlet);
+
+      newInlet.start();
+      if (outlet.isAvailable()) {
+        newInlet.pull();
+      }
+    } catch (Exception e) {
+      outlet.fail(e);
+    }
+  }
+}

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

@@ -8,15 +8,15 @@
  * <p>
  * A signal is used to wrap callbacks to inlet and outlet listeners. Rather than invoking these directly, they are
  * enqueued to be invoked in an unrolled fashion, after the current callback is finished. This ensures that stages do
- * not have to handle reentrant invocations, when the stage invokes something on a inlet our outlet, it can be sure
- * that its state before invocation hasn't changed after the invocation returns. This also solves blowing the stack
- * as it ensures there is no unbounded recursion.
+ * not have to handle reentrant invocations, when the stage invokes something on an inlet or outlet, it can be sure
+ * that its state before invocation hasn't changed after the invocation returns. This also prevents any unbounded
+ * recursion due to stages synchronously responding back and forth to each others signals.
  * <p>
  * Generally speaking, the implementation of inlets and outlets will ensure that a new signal is not allocated each
  * time it is needed for pulls and pushes, rather the signal is instantiated once at start up and reused each time.
  * This is safe to do since those signals don't carry state.
  */
-public interface Signal {
+interface Signal {
   /**
    * Invoke the signal.
    */

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

@@ -91,18 +91,19 @@ interface InletListener {
   /**
    * Indicates that an element has been pushed. The element can be received using {@link StageInlet#grab()}.
    * <p>
-   * If this throws an exception, the error will be passed to {@link #onUpstreamFailure(Throwable)}, upstream will
-   * be cancelled, and the stage will not receive any further signals.
+   * If this throws an exception, the error will be passed to {@link #onUpstreamFailure(Throwable)}, anything upstream
+   * from this inlet will be cancelled, and the stage listening will not receive any further signals.
    */
   void onPush();
 
   /**
    * Indicates that upstream has completed the stream. Unless this throws an exception, no signals may be sent to the
    * inlet after this has been invoked.
    * <p>
-   * If this throws an exception, the error will be passed to {@link #onUpstreamFailure(Throwable)}, upstream will
-   * be cancelled, and the stage will not receive any further signals. Stages should be careful to ensure that if they
-   * do throw from this method, that they are ready to receive that exception from {@code onUpstreamFailure}.
+   * If this throws an exception, the error will be passed to {@link #onUpstreamFailure(Throwable)}, anything upstream
+   * from this inlet will be cancelled, and the stage will not receive any further signals. Stages should be careful
+   * to ensure that if they do throw from this method, that they are ready to receive that exception from
+   * {@code onUpstreamFailure}.
    */
   void onUpstreamFinish();
 
@@ -111,9 +112,9 @@ interface InletListener {
    * will be sent to this listener.
    * <p>
    * If this throws an exception, the entire stream will be shut down, since there's no other way to guarantee that
-   * the failure signal will be propagated downstream. Hence, stages should generally not throw exceptions from this
-   * method, particularly exceptions from user supplied callbacks, as such errors will not be recoverable (eg, a
-   * recover stage won't be able to resume the stream).
+   * the failure signal will be propagated downstream. Hence, stages should not throw exceptions from this method,
+   * particularly exceptions from user supplied callbacks, as such errors will not be recoverable (eg, a recover stage
+   * won't be able to resume the stream).
    */
   void onUpstreamFailure(Throwable error);
 }