Coupled cancel from onComplete or onError workaround

danielkec · danielkec · commit 33fab3d32f3e · 2020-02-21T16:06:07.000+01:00
* microprofile/microprofile-reactive-streams-operators#131 Signed-off-by: Daniel Kec <daniel.kec@oracle.com>
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/MultiCoupledProcessor.java b/common/reactive/src/main/java/io/helidon/common/reactive/MultiCoupledProcessor.java
@@ -17,8 +17,10 @@
 
 package io.helidon.common.reactive;
 
+import java.util.LinkedList;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Queue;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -50,6 +52,7 @@ public class MultiCoupledProcessor<T, R> implements Flow.Processor<T, R>, Multi<
     private Flow.Subscription inletSubscription;
     private Flow.Subscription passedInPublisherSubscription;
     private AtomicBoolean cancelled = new AtomicBoolean(false);
+    private Queue<Runnable> forbiddenCalls = new LinkedList<>();
 
     private MultiCoupledProcessor(Flow.Subscriber<T> passedInSubscriber, Flow.Publisher<R> passedInPublisher) {
         this.passedInSubscriber = SequentialSubscriber.create(passedInSubscriber);
@@ -78,7 +81,7 @@ public void subscribe(Flow.Subscriber<? super R> outletSubscriber) {
 
             @Override
             public void onSubscribe(Flow.Subscription passedInPublisherSubscription) {
-                //Passed in publisher called onSubscribed
+                //Passed in publisher called onSubscribe
                 Objects.requireNonNull(passedInPublisherSubscription);
                 // https://github.com/reactive-streams/reactive-streams-jvm#2.5
                 if (Objects.nonNull(MultiCoupledProcessor.this.passedInPublisherSubscription) || cancelled.get()) {
@@ -91,6 +94,7 @@ public void onSubscribe(Flow.Subscription passedInPublisherSubscription) {
             @Override
             @SuppressWarnings("unchecked")
             public void onNext(R t) {
+                tryForbiddenCalls();
                 //Passed in publisher sent onNext
                 Objects.requireNonNull(t);
                 outletSubscriber.onNext(t);
@@ -105,7 +109,7 @@ public void onError(Throwable t) {
                 passedInSubscriber.onError(t);
                 inletSubscriber.onError(t);
                 //203 https://github.com/eclipse/microprofile-reactive-streams-operators/issues/131
-                Optional.ofNullable(inletSubscription).ifPresent(Flow.Subscription::cancel);
+                forbiddenCalls.offer(() -> Optional.ofNullable(inletSubscription).ifPresent(Flow.Subscription::cancel));
             }
 
             @Override
@@ -115,7 +119,7 @@ public void onComplete() {
                 outletSubscriber.onComplete();
                 passedInSubscriber.onComplete();
                 //203 https://github.com/eclipse/microprofile-reactive-streams-operators/issues/131
-                Optional.ofNullable(inletSubscription).ifPresent(Flow.Subscription::cancel);
+                forbiddenCalls.offer(() -> Optional.ofNullable(inletSubscription).ifPresent(Flow.Subscription::cancel));
             }
         });
 
@@ -124,11 +128,8 @@ public void onComplete() {
             @Override
             public void request(long n) {
                 // Request from outlet subscriber
-                StreamValidationUtils.checkRecursionDepth(
-                        "MultiCoupledProcessor1",
-                        8,
-                        () -> passedInPublisherSubscription.request(n),
-                        (actDepth, t) -> outletSubscriber.onError(t));
+                passedInPublisherSubscription.request(n);
+                tryForbiddenCalls();
             }
 
             @Override
@@ -137,6 +138,7 @@ public void cancel() {
                 passedInSubscriber.onComplete();
                 Optional.ofNullable(inletSubscription).ifPresent(Flow.Subscription::cancel);
                 passedInPublisherSubscription.cancel();
+                tryForbiddenCalls();
             }
         });
     }
@@ -153,11 +155,8 @@ public void onSubscribe(Flow.Subscription inletSubscription) {
         passedInSubscriber.onSubscribe(new Flow.Subscription() {
             @Override
             public void request(long n) {
-                StreamValidationUtils.checkRecursionDepth(
-                        "MultiCoupledProcessor2",
-                        8,
-                        () -> inletSubscription.request(n),
-                        (actDepth, t) -> passedInSubscriber.onError(t));
+                inletSubscription.request(n);
+                tryForbiddenCalls();
             }
 
             @Override
@@ -169,6 +168,7 @@ public void cancel() {
                 inletSubscription.cancel();
                 outletSubscriber.onComplete();
                 passedInPublisherSubscription.cancel();
+                tryForbiddenCalls();
             }
         });
     }
@@ -197,4 +197,14 @@ public void onComplete() {
         passedInPublisherSubscription.cancel();
     }
 
+    private void tryForbiddenCalls() {
+        while (true) {
+            Runnable polledCall = forbiddenCalls.poll();
+            if (Objects.isNull(polledCall)) {
+                return;
+            }
+            polledCall.run();
+        }
+    }
+
 }
diff --git a/common/reactive/src/main/java/io/helidon/common/reactive/StreamValidationUtils.java b/common/reactive/src/main/java/io/helidon/common/reactive/StreamValidationUtils.java
@@ -17,70 +17,18 @@
 
 package io.helidon.common.reactive;
 
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 /**
  * Helper methods for stream validation.
  */
 public class StreamValidationUtils {
 
-    private static ThreadLocal<Map<String, AtomicLong>> recursionDepthThreadLocal = new ThreadLocal<>();
-
 
     private StreamValidationUtils() {
     }
 
-    /**
-     * Validation of Reactive Streams Specification for JVM rule 3.3.
-     * <br>
-     * {@code Subscription.request} MUST place an upper bound on possible synchronous
-     * recursion between {@code Publisher} and {@code Subscriber}.
-     *
-     * @param key          key to differentiate multiple checks on one thread
-     * @param maxDepth     maximal expected recursion depth
-     * @param guardedBlock code block to check recursion for
-     * @param onExceeded   called if recursion is deeper than maxDepth,
-     *                     provided with actual depth and spec compliant exception.
-     * @param <T>          payload type of the subscriber
-     * @return true if valid
-     * @see <a href="https://github.com/reactive-streams/reactive-streams-jvm#3.3">reactive-streams/reactive-streams-jvm#3.3</a>
-     */
-    public static <T> boolean checkRecursionDepth(String key, int maxDepth, Runnable guardedBlock,
-                                                  BiConsumer<Long, Throwable> onExceeded) {
-        Map<String, AtomicLong> counterMap = recursionDepthThreadLocal.get();
-        if (Objects.isNull(counterMap)) {
-            counterMap = new HashMap<>();
-            counterMap.put(key, new AtomicLong(0));
-            recursionDepthThreadLocal.set(counterMap);
-        }
-
-        counterMap.putIfAbsent(key, new AtomicLong(0));
-        AtomicLong recursionDepthCounter = counterMap.get(key);
-        try {
-            if (recursionDepthCounter.incrementAndGet() > maxDepth) {
-                long exceededRecursionDepth = recursionDepthCounter.get();
-                Optional.of(onExceeded)
-                        .ifPresent(onExc -> onExc
-                                .accept(exceededRecursionDepth, new IllegalCallerException(String
-                                        .format("Recursion depth exceeded, max depth expected %d but actual is %d, rule 3.3",
-                                                maxDepth, exceededRecursionDepth))));
-                return false;
-            }
-
-
-            guardedBlock.run();
-            return true;
-        } finally {
-            recursionDepthCounter.decrementAndGet();
-        }
-    }
-
     /**
      * Validation of Reactive Streams Specification for JVM rule 3.9.
      * <br>
diff --git a/common/reactive/src/test/java/io/helidon/common/reactive/RecursionDepthTest.java b/common/reactive/src/test/java/io/helidon/common/reactive/RecursionDepthTest.java
diff --git a/microprofile/tests/tck/tck-reactive-operators/src/test/java/io/helidon/microprofile/reactive/HelidonReactiveStreamsTckTest.java b/microprofile/tests/tck/tck-reactive-operators/src/test/java/io/helidon/microprofile/reactive/HelidonReactiveStreamsTckTest.java
@@ -18,24 +18,16 @@
 package io.helidon.microprofile.reactive;
 
 import org.eclipse.microprofile.reactive.streams.operators.tck.ReactiveStreamsTck;
-import org.eclipse.microprofile.reactive.streams.operators.tck.spi.CoupledStageVerification;
 import org.reactivestreams.tck.TestEnvironment;
 
 public class HelidonReactiveStreamsTckTest extends ReactiveStreamsTck<HelidonReactiveStreamEngine> {
 
     public HelidonReactiveStreamsTckTest() {
-        super(new TestEnvironment(200, 200, false));
+        super(new TestEnvironment(200));
     }
 
     @Override
     protected HelidonReactiveStreamEngine createEngine() {
         return new HelidonReactiveStreamEngine();
     }
-
-    @Override
-    protected boolean isEnabled(Object test) {
-        // Remove when TCK test issues are solved
-        return // https://github.com/eclipse/microprofile-reactive-streams-operators/issues/131
-                !(test instanceof CoupledStageVerification.ProcessorVerification);
-    }
 }
