[Reactive] Apply bugfixes from helidon-io#1511

akarnokd · Mar 19, 2020 · b69d89c · b69d89c
1 parent 9b10b52
commit b69d89c
Show file tree

Hide file tree

Showing 11 changed files with 66 additions and 14 deletions.
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/Multi.java b/common/reactive/src/main/java/io/helidon/common/reactive/Multi.java
@@ -461,7 +461,7 @@ static <T> Multi<T> never() {
      * @param <T>         item type
      * @return Multi
      */
-    static <T> Multi<T> concat(Multi<T> firstMulti, Multi<T> secondMulti) {
+    static <T> Multi<T> concat(Flow.Publisher<T> firstMulti, Flow.Publisher<T> secondMulti) {
         return ConcatPublisher.create(firstMulti, secondMulti);
     }
 
@@ -484,23 +484,23 @@ default Multi<T> onTerminate(Runnable onTerminate) {
     /**
      * Executes given {@link java.lang.Runnable} when onComplete signal is received.
      *
-     * @param onTerminate {@link java.lang.Runnable} to be executed.
+     * @param onComplete {@link java.lang.Runnable} to be executed.
      * @return Multi
      */
-    default Multi<T> onComplete(Runnable onTerminate) {
+    default Multi<T> onComplete(Runnable onComplete) {
         return new MultiTappedPublisher<>(this,
                 null,
                 null,
                 null,
-                onTerminate,
+                onComplete,
                 null,
                 null);
     }
 
     /**
      * Executes given {@link java.lang.Runnable} when onError signal is received.
      *
-     * @param onErrorConsumer {@link java.lang.Runnable} to be executed.
+     * @param onErrorConsumer {@link java.util.function.Consumer} to be executed.
      * @return Multi
      */
     default Multi<T> onError(Consumer<Throwable> onErrorConsumer) {

diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/MultiFlatMapIterable.java b/common/reactive/src/main/java/io/helidon/common/reactive/MultiFlatMapIterable.java
@@ -194,6 +194,7 @@ void drain() {
                                 hasNext = iterator.hasNext();
                             } catch (Throwable ex) {
                                 canceled = true;
+                                upstream.cancel();
                                 downstream.onError(ex);
                                 continue;
                             }
@@ -222,6 +223,7 @@ void drain() {
                                         "The iterator returned a null item");
                             } catch (Throwable ex) {
                                 canceled = true;
+                                upstream.cancel();
                                 downstream.onError(ex);
                                 continue outer;
                             }
@@ -242,6 +244,7 @@ void drain() {
                                 hasNext = iterator.hasNext();
                             } catch (Throwable ex) {
                                 canceled = true;
+                                upstream.cancel();
                                 downstream.onError(ex);
                                 continue outer;
                             }

diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/MultiFlatMapPublisher.java b/common/reactive/src/main/java/io/helidon/common/reactive/MultiFlatMapPublisher.java
@@ -407,11 +407,10 @@ void drainLoop() {
                                 continue;
                             }
                         }
-
-                        emitted = e;
                     }
                 }
 
+                emitted = e;
                 missed = addAndGet(-missed);
                 if (missed == 0) {
                     break;

diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/MultiOnErrorResumeWith.java b/common/reactive/src/main/java/io/helidon/common/reactive/MultiOnErrorResumeWith.java
@@ -93,7 +93,9 @@ public void onError(Throwable throwable) {
                 publisher = Objects.requireNonNull(fallbackFunction.apply(throwable),
                         "The fallback function returned a null Flow.Publisher");
             } catch (Throwable ex) {
-                ex.addSuppressed(throwable);
+                if (ex != throwable) {
+                    ex.addSuppressed(throwable);
+                }
                 downstream.onError(ex);
                 return;
             }

diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/Single.java b/common/reactive/src/main/java/io/helidon/common/reactive/Single.java
@@ -313,23 +313,23 @@ default Single<T> onTerminate(Runnable onTerminate) {
     /**
      * Executes given {@link java.lang.Runnable} when onComplete signal is received.
      *
-     * @param onTerminate {@link java.lang.Runnable} to be executed.
+     * @param onComplete  {@link java.lang.Runnable} to be executed.
      * @return Single
      */
-    default Single<T> onComplete(Runnable onTerminate) {
+    default Single<T> onComplete(Runnable onComplete) {
         return new SingleTappedPublisher<>(this,
                 null,
                 null,
                 null,
-                onTerminate,
+                onComplete,
                 null,
                 null);
     }
 
     /**
      * Executes given {@link java.lang.Runnable} when onError signal is received.
      *
-     * @param onErrorConsumer {@link java.lang.Runnable} to be executed.
+     * @param onErrorConsumer {@link java.util.function.Consumer} to be executed.
      * @return Single
      */
     default Single<T> onError(Consumer<Throwable> onErrorConsumer) {

diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/SingleOnErrorResume.java b/common/reactive/src/main/java/io/helidon/common/reactive/SingleOnErrorResume.java
@@ -77,7 +77,9 @@ public void onError(Throwable throwable) {
                 fallback = Objects.requireNonNull(fallbackFunction.apply(throwable),
                         "The fallback function returned a null item");
             } catch (Throwable ex) {
-                ex.addSuppressed(throwable);
+                if (ex != throwable) {
+                    ex.addSuppressed(throwable);
+                }
                 error(ex);
                 return;
             }

diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/SingleOnErrorResumeWith.java b/common/reactive/src/main/java/io/helidon/common/reactive/SingleOnErrorResumeWith.java
@@ -82,7 +82,9 @@ public void onError(Throwable throwable) {
                 fallback = Objects.requireNonNull(fallbackFunction.apply(throwable),
                         "The fallback function returned a null Single");
             } catch (Throwable ex) {
-                ex.addSuppressed(throwable);
+                if (ex != throwable) {
+                    ex.addSuppressed(throwable);
+                }
                 error(ex);
                 return;
             }

diff --git a/common/reactive/src/test/java/io/helidon/common/reactive/MultiOnErrorResumeTest.java b/common/reactive/src/test/java/io/helidon/common/reactive/MultiOnErrorResumeTest.java
@@ -76,4 +76,15 @@ public void failAfterItemsBackpressure() {
                 .request(1)
                 .assertResult(1, 2, 3, 4);
     }
+
+    @Test
+    public void noSelfSuppressionFailure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<>();
+
+        Multi.<Integer>error(new IllegalArgumentException())
+                .onErrorResume(e -> { throw (IllegalArgumentException)e; })
+                .subscribe(ts);
+
+        ts.assertFailure(IllegalArgumentException.class);
+    }
 }
diff --git a/common/reactive/src/test/java/io/helidon/common/reactive/MultiOnErrorResumeWithTest.java b/common/reactive/src/test/java/io/helidon/common/reactive/MultiOnErrorResumeWithTest.java
@@ -99,4 +99,15 @@ public void failAfterItemsBackpressure() {
         .request(2)
         .assertResult(1, 2, 3, 4, 5);
     }
+
+    @Test
+    public void noSelfSuppressionFailure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<>();
+
+        Multi.<Integer>error(new IllegalArgumentException())
+                .onErrorResumeWith(e -> { throw (IllegalArgumentException)e; })
+                .subscribe(ts);
+
+        ts.assertFailure(IllegalArgumentException.class);
+    }
 }
diff --git a/common/reactive/src/test/java/io/helidon/common/reactive/SingleOnErrorResumeTest.java b/common/reactive/src/test/java/io/helidon/common/reactive/SingleOnErrorResumeTest.java
@@ -55,4 +55,15 @@ public void emptySource() {
 
         ts.assertResult();
     }
+
+    @Test
+    public void noSelfSuppressionFailure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<>();
+
+        Single.<Integer>error(new IllegalArgumentException())
+                .onErrorResume(e -> { throw (IllegalArgumentException)e; })
+                .subscribe(ts);
+
+        ts.assertFailure(IllegalArgumentException.class);
+    }
 }
diff --git a/common/reactive/src/test/java/io/helidon/common/reactive/SingleOnErrorResumeWithTest.java b/common/reactive/src/test/java/io/helidon/common/reactive/SingleOnErrorResumeWithTest.java
@@ -73,4 +73,15 @@ public void emptyFallback() {
 
         ts.assertResult();
     }
+
+    @Test
+    public void noSelfSuppressionFailure() {
+        TestSubscriber<Integer> ts = new TestSubscriber<>();
+
+        Single.<Integer>error(new IllegalArgumentException())
+                .onErrorResumeWith(e -> { throw (IllegalArgumentException)e; })
+                .subscribe(ts);
+
+        ts.assertFailure(IllegalArgumentException.class);
+    }
 }