Merge pull request #5 from jroper/zerodep-improvements

Error handling improvements in zerodep impl

Author: jroper

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

@@ -494,12 +494,7 @@ public void execute(Runnable command) {
         int signalsDrained = 0;
         while (!signals.isEmpty() && signalsDrained < 32) {
           signalsDrained++;
-          Signal signal = signals.removeFirst();
-          try {
-            signal.signal();
-          } catch (RuntimeException e) {
-            signal.signalFailed(e);
-          }
+          signals.removeFirst().signal();
         }
 
         // If there were more than 32 unrolled signals, we resubmit
@@ -509,9 +504,9 @@ public void execute(Runnable command) {
           });
         }
 
-      } catch (RuntimeException e) {
+      } catch (Throwable t) {
         // shut down the stream
-        streamFailure(e);
+        streamFailure(t);
         // Clear remaining signals
         signals.clear();
       }
@@ -533,7 +528,11 @@ private void streamFailure(Throwable error) {
     // todo handle better
     error.printStackTrace();
     for (Port port : ports) {
-      port.onStreamFailure(error);
+      try {
+        port.onStreamFailure(error);
+      } catch (Exception e) {
+        // Ignore
+      }
     }
     ports.clear();
   }

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

@@ -40,11 +40,7 @@ public void onPush() {
 
   @Override
   public void onUpstreamFinish() {
-    try {
-      result.complete(collector.finisher().apply(container));
-    } catch (RuntimeException e) {
-      result.completeExceptionally(e);
-    }
+    result.complete(collector.finisher().apply(container));
     container = null;
   }
 

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

@@ -25,12 +25,8 @@ public void onPush() {
 
   @Override
   public void onUpstreamFinish() {
-    try {
-      action.run();
-      outlet.complete();
-    } catch (RuntimeException e) {
-      outlet.fail(e);
-    }
+    action.run();
+    outlet.complete();
   }
 
   @Override

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

@@ -25,12 +25,8 @@ public void onPush() {
 
   @Override
   public void onUpstreamFinish() {
-    try {
-      action.run();
-      outlet.complete();
-    } catch (RuntimeException e) {
-      outlet.fail(e);
-    }
+    action.run();
+    outlet.complete();
   }
 
   @Override

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

@@ -41,10 +41,11 @@ public void onStreamFailure(Throwable reason) {
       }
     }
     if (!downstreamFinished) {
-      failure = reason;
       if (subscriber != null) {
         downstreamFinished = true;
         subscriber.onError(reason);
+      } else {
+        failure = reason;
       }
     }
   }
@@ -91,6 +92,7 @@ public void cancel() {
 
   @Override
   public void request(long n) {
+    // possible optimization: place n in a queue, and dispatch singleton runnable to executor, to save an allocation
     builtGraph.execute(() -> {
       if (!upstreamFinished) {
         if (n <= 0) {
@@ -192,30 +194,21 @@ public void setListener(OutletListener listener) {
     this.listener = Objects.requireNonNull(listener, "Listener must not be null");
   }
 
-  private abstract class AbstractSignal implements Signal {
-    @Override
-    public void signalFailed(Throwable error) {
-      onStreamFailure(error);
-    }
-  }
-
-  private final Signal onPullSignal = new AbstractSignal() {
-    @Override
-    public void signal() {
-      if (!upstreamFinished) {
-        pulled = true;
+  private final Signal onPullSignal = () -> {
+    if (!upstreamFinished) {
+      pulled = true;
+      try {
         listener.onPull();
+      } catch (Exception e) {
+        onStreamFailure(e);
       }
     }
   };
 
-  private final Signal onDownstreamFinishSignal = new AbstractSignal() {
-    @Override
-    public void signal() {
-      if (!upstreamFinished) {
-        upstreamFinished = true;
-        listener.onDownstreamFinish();
-      }
+  private final Signal onDownstreamFinishSignal = () -> {
+    if (!upstreamFinished) {
+      upstreamFinished = true;
+      listener.onDownstreamFinish();
     }
   };
 }

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

@@ -5,24 +5,20 @@
 
 /**
  * A signal, used by inlets and outlets.
- *
- * A signal represents a callback to an inlet or outlet listener. All callbacks should be wrapped in a signal to
- * allow signals to be enqueued, so that they can be executed in an unrolled fashion, rather than recursively.
- *
- * Implementations of signal are only allocated when a stream or sub stream starts. They should carry no state, any
- * state associated with the signal should be held by the inlet or outlet. This ensures that no allocations are done
- * of signals while running the stream.
+ * <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.
+ * <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 {
   /**
    * Invoke the signal.
    */
   void signal();
-
-  /**
-   * This will be invoked if {@link #signal()} throws an exception. Implementations of this should then terminate both
-   * upstream and downstream. This mechanism means that {@code signal} doesn't have to do its own exception handling,
-   * but rather exceptions can be handled generically by a stage.
-   */
-  void signalFailed(Throwable t);
 }

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

@@ -90,23 +90,30 @@ 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.
    */
   void onPush();
 
   /**
-   * Indicates that upstream has completed the stream. No signals may be sent to the inlet after this has been invoked.
-   *
-   * This must be very careful not to throw an exception. If it does, then the signal to complete will not reach
-   * downstream.
+   * 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}.
    */
   void onUpstreamFinish();
 
   /**
-   * Indicates that upstream has completed the stream with a failure. No signals may be sent to the inlet after this has
-   * been invoked.
-   *
-   * This must be very careful not to throw an exception. If it does, then the signal to complete will not reach
-   * downstream.
+   * Indicates that upstream has completed the stream with a failure. Once this has been invoked, no other signals
+   * 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).
    */
   void onUpstreamFailure(Throwable error);
 }

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

@@ -78,15 +78,20 @@ public void onDownstreamFinish() {
 interface OutletListener {
   /**
    * A pull signal, indicates that downstream is ready to be pushed to.
+   * <p>
+   * If this throws an exception, the stage will be cancelled using {@link #onDownstreamFinish()} and the error will
+   * be propagated downstream.
    */
   void onPull();
 
   /**
    * A completion signal, indicates that downstream has completed. No further signals may be sent to this outlet after
    * this signal is received.
-   *
-   * This must be very careful not to throw an exception. If it does, then the signal to cancel will not reach
-   * upstream.
+   * <p>
+   * If this throws an exception, the entire stream will be shut down, since there's no other way to guarantee that
+   * the cancel signal will be propagated upstream. 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).
    */
   void onDownstreamFinish();
 }

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

@@ -166,69 +166,67 @@ public void setListener(InletListener listener) {
     }
   }
 
-  private abstract class AbstractSignal implements Signal {
+  private abstract class RecoverableSignal implements Signal {
     @Override
-    public void signalFailed(Throwable error) {
-      if (!outletFinished) {
-        outletFinished = true;
-        outletListener.onDownstreamFinish();
-      }
-      if (!inletFinished) {
-        inletFinished = true;
-        inletListener.onUpstreamFailure(error);
+    public final void signal() {
+      try {
+        doSignal();
+      } catch (Exception e) {
+        onStreamFailure(e);
       }
     }
+
+    protected abstract void doSignal();
   }
 
-  private final Signal onPullSignal = new AbstractSignal() {
+  private final Signal onPullSignal = new RecoverableSignal() {
     @Override
-    public void signal() {
+    protected void doSignal() {
       if (!outletFinished) {
         outletPulled = true;
         outletListener.onPull();
       }
     }
   };
 
-  private final Signal onDownstreamFinishSignal = new AbstractSignal() {
-    @Override
-    public void signal() {
-      if (!outletFinished) {
-        outletFinished = true;
-        outletListener.onDownstreamFinish();
-      }
+  private final Signal onDownstreamFinishSignal = () -> {
+    if (!outletFinished) {
+      outletFinished = true;
+      outletListener.onDownstreamFinish();
     }
   };
 
-  private final Signal onPushSignal = new AbstractSignal() {
+  private final Signal onPushSignal = new RecoverableSignal() {
     @Override
-    public void signal() {
+    protected void doSignal() {
       if (!inletFinished) {
         inletPushed = true;
         inletListener.onPush();
       }
     }
   };
 
-  private final Signal onUpstreamFinishSignal = new AbstractSignal() {
-    @Override
-    public void signal() {
-      if (!inletFinished) {
-        inletFinished = true;
+  private final Signal onUpstreamFinishSignal = () -> {
+    if (!inletFinished) {
+      inletFinished = true;
+      try {
         inletListener.onUpstreamFinish();
+      } catch (Exception e) {
+        inletListener.onUpstreamFailure(e);
+        if (!outletFinished) {
+          outletFinished = true;
+          outletListener.onDownstreamFinish();
+        }
       }
     }
   };
 
-  private final Signal onUpstreamErrorSignal = new AbstractSignal() {
-    @Override
-    public void signal() {
-      if (!inletFinished) {
-        inletFinished = true;
-        Throwable theFailure = failure;
-        failure = null;
-        inletListener.onUpstreamFailure(theFailure);
-      }
+  private final Signal onUpstreamErrorSignal = () -> {
+    if (!inletFinished) {
+      inletFinished = true;
+      Throwable theFailure = failure;
+      failure = null;
+      inletListener.onUpstreamFailure(theFailure);
     }
   };
 }