Merge pull request #67 from cescoffier/define-recover-stages

Define the onErrorResume stage.

Author: jroper

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

@@ -407,6 +407,71 @@ public ProcessorBuilder<T, R> onError(Consumer<Throwable> errorHandler) {
         return addStage(new Stage.OnError(errorHandler));
     }
 
+    /**
+     * Returns a stream containing all the elements from this stream. Additionally, in the case of failure, rather than
+     * invoking {@link #onError(Consumer)}, it invokes the given method and emits the result as final event of the stream.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead emit the return value of
+     * the passed function. This operator prevents errors from propagating or to supply fallback data should errors be
+     * encountered.
+     *
+     * @param errorHandler the function returning the value that need to be emitting instead of the error.
+     *                     The function must not return {@code null}
+     * @return The new processor
+     */
+    public ProcessorBuilder<T, R> onErrorResume(Function<Throwable, R> errorHandler) {
+        return addStage(new Stage.OnErrorResume(errorHandler));
+    }
+
+    /**
+     * Returns a stream containing all the elements from this stream. Additionally, in the case of failure, rather than
+     * invoking {@link #onError(Consumer)}, it invokes the given method and emits the returned {@link PublisherBuilder}
+     * instead.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead relinquish control to the
+     * {@link PublisherBuilder} returned from given function, which invoke the subscriber's <code>onNext</code> method if
+     * it is able to do so. In such a case, because no publisher necessarily invokes <code>onError</code>, the subscriber
+     * may never know that an error happened.
+     *
+     * @param errorHandler the function returning the stream that need to be emitting instead of the error.
+     *                     The function must not return {@code null}
+     * @return The new processor
+     */
+    public ProcessorBuilder<T, R> onErrorResumeWith(Function<Throwable, PublisherBuilder<R>> errorHandler) {
+        return addStage(new Stage.OnErrorResumeWith(errorHandler.andThen(PublisherBuilder::toGraph)));
+    }
+
+    /**
+     * Returns a stream containing all the elements from this stream. Additionally, in the case of failure, rather than
+     * invoking {@link #onError(Consumer)}, it invokes the given method and emits the returned {@link PublisherBuilder}
+     * instead.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead relinquish control to the
+     * {@link PublisherBuilder} returned from given function, which invoke the subscriber's <code>onNext</code> method if
+     * it is able to do so. In such a case, because no publisher necessarily invokes <code>onError</code>, the subscriber
+     * may never know that an error happened.
+     *
+     * @param errorHandler the function returning the stream that need to be emitting instead of the error.
+     *                     The function must not return {@code null}
+     * @return The new processor
+     */
+    public ProcessorBuilder<T, R> onErrorResumeWithPublisher(Function<Throwable, Publisher<R>> errorHandler) {
+        return addStage(new Stage.OnErrorResumeWith(
+            errorHandler
+                .andThen(ReactiveStreams::fromPublisher)
+                .andThen(PublisherBuilder::toGraph))
+        );
+    }
+
     /**
      * Returns a stream containing all the elements from this stream, additionally performing the provided action when this
      * stream completes or failed. The given action does not know if the stream failed or completed. If you need to

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

@@ -425,6 +425,71 @@ public PublisherBuilder<T> onError(Consumer<Throwable> errorHandler) {
         return addStage(new Stage.OnError(errorHandler));
     }
 
+    /**
+     * Returns a stream containing all the elements from this stream. Additionally, in the case of failure, rather than
+     * invoking {@link #onError(Consumer)}, it invokes the given method and emits the result as final event of the stream.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead emit the return value of
+     * the passed function. This operator prevents errors from propagating or to supply fallback data should errors be
+     * encountered.
+     *
+     * @param errorHandler the function returning the value that need to be emitting instead of the error.
+     *                     The function must not return {@code null}
+     * @return The new publisher
+     */
+    public PublisherBuilder<T> onErrorResume(Function<Throwable, T> errorHandler) {
+        return addStage(new Stage.OnErrorResume(errorHandler));
+    }
+
+    /**
+     * Returns a stream containing all the elements from this stream. Additionally, in the case of failure, rather than
+     * invoking {@link #onError(Consumer)}, it invokes the given method and emits the returned {@link PublisherBuilder}
+     * instead.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead relinquish control to the
+     * {@link PublisherBuilder} returned from given function, which invoke the subscriber's <code>onNext</code> method if
+     * it is able to do so. In such a case, because no publisher necessarily invokes <code>onError</code>, the subscriber
+     * may never know that an error happened.
+     *
+     * @param errorHandler the function returning the stream that need to be emitting instead of the error.
+     *                     The function must not return {@code null}
+     * @return The new publisher
+     */
+    public PublisherBuilder<T> onErrorResumeWith(Function<Throwable, PublisherBuilder<T>> errorHandler) {
+        return addStage(new Stage.OnErrorResumeWith(errorHandler.andThen(PublisherBuilder::toGraph)));
+    }
+
+    /**
+     * Returns a stream containing all the elements from this stream. Additionally, in the case of failure, rather than
+     * invoking {@link #onError(Consumer)}, it invokes the given method and emits the returned {@link PublisherBuilder}
+     * instead.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead relinquish control to the
+     * {@link PublisherBuilder} returned from given function, which invoke the subscriber's <code>onNext</code> method if
+     * it is able to do so. In such a case, because no publisher necessarily invokes <code>onError</code>, the subscriber
+     * may never know that an error happened.
+     *
+     * @param errorHandler the function returning the stream that need to be emitting instead of the error.
+     *                     The function must not return {@code null}
+     * @return The new publisher
+     */
+    public <S> PublisherBuilder<S> onErrorResumeWithPublisher(Function<Throwable, Publisher<? extends S>> errorHandler) {
+        return addStage(new Stage.OnErrorResumeWith(
+            errorHandler
+                .andThen(ReactiveStreams::fromPublisher)
+                .andThen(PublisherBuilder::toGraph))
+        );
+    }
+
     /**
      * Returns a stream containing all the elements from this stream, additionally performing the provided action when this
      * stream completes or failed. The given action does not know if the stream failed or completed. If you need to

streams/api/src/main/java/org/eclipse/microprofile/reactive/streams/spi/Stage.java

@@ -559,6 +559,73 @@ public Runnable getAction() {
         }
     }
 
+    /**
+     * A stage to handle error from upstream. It builds a stream containing all the elements from  upstream. Additionally,
+     * in the case of failure, rather than invoking {@link Subscriber#onError(Throwable)}, it invokes a given method and
+     * emits the result as final event of the stream.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead emit the return value of
+     * the passed function. This operator prevents errors from propagating or to supply fallback data should errors be
+     * encountered.
+     *
+     * Any {@link RuntimeException} thrown by the function should be propagated down the stream as an error.
+     *
+     */
+    final class OnErrorResume implements Inlet, Outlet {
+        private final Function<Throwable, ?> function;
+
+
+        public OnErrorResume(Function<Throwable, ?>  function) {
+            this.function = function;
+        }
+
+        /**
+         * The error handler.
+         *
+         * @return  the error handler.
+         */
+        public Function<Throwable, ?> getFunction() {
+            return function;
+        }
+    }
+
+    /**
+     * A stage to handle error from upstream. It builds a stream containing all the elements from  upstream. Additionally,
+     * in the case of failure, rather than invoking {@link Subscriber#onError(Throwable)}, it invokes a given method and
+     * switch the control to the returned stream.
+     *
+     * By default, when a stream encounters an error that prevents it from emitting the expected item to its subscriber,
+     * the stream (publisher) invokes its subscriber's <code>onError</code> method, and then terminate without invoking
+     * any more of its subscriber's methods. This operator changes this behavior. If the current stream encounters an
+     * error, instead of invoking its subscriber's <code>onError</code> method, it will instead relinquish control to the
+     * {@link org.eclipse.microprofile.reactive.streams.PublisherBuilder} returned from given function, which invoke the
+     * subscriber's <code>onNext</code> method if it is able to do so. In such a case, because no publisher necessarily
+     * invokes <code>onError</code>, the subscriber may never know that an error happened.
+     *
+     * Any {@link RuntimeException} thrown by the function should be propagated down the stream as an error.
+     *
+     */
+    final class OnErrorResumeWith implements Inlet, Outlet {
+        private final Function<Throwable, Graph> function;
+
+
+        public OnErrorResumeWith(Function<Throwable, Graph>  function) {
+            this.function = function;
+        }
+
+        /**
+         * The error handler.
+         *
+         * @return  the error handler.
+         */
+        public Function<Throwable, Graph> getFunction() {
+            return function;
+        }
+    }
+
     /**
      * A failed publisher.
      * <p>
@@ -616,7 +683,7 @@ public Graph getSecond() {
     }
 
     final class Cancel implements Inlet {
-        public final static Cancel INSTANCE = new Cancel();
+        public static final Cancel INSTANCE = new Cancel();
 
         private Cancel() {
         }