Properly handle "cancellations" of a ratpack promise for bulkheads (R…

…eactiveX#543)
jihoonson · Jul 12, 2019 · 1a87d08 · 1a87d08
1 parent 9c66779
commit 1a87d08
Show file tree

Hide file tree

Showing 3 changed files with 58 additions and 12 deletions.
diff --git a/...4j-ratpack/src/main/java/io/github/resilience4j/ratpack/bulkhead/BulkheadTransformer.java b/...4j-ratpack/src/main/java/io/github/resilience4j/ratpack/bulkhead/BulkheadTransformer.java
@@ -76,7 +76,7 @@ public void error(Throwable throwable) {
 
                     @Override
                     public void complete() {
-                        bulkhead.onComplete();
+                        bulkhead.releasePermission();
                         down.complete();
                     }
                 });

diff --git a/...ck/src/test/groovy/io/github/resilience4j/ratpack/bulkhead/BulkheadTransformerSpec.groovy b/...ck/src/test/groovy/io/github/resilience4j/ratpack/bulkhead/BulkheadTransformerSpec.groovy
@@ -20,6 +20,7 @@ import io.github.resilience4j.bulkhead.BulkheadConfig
 import io.github.resilience4j.bulkhead.BulkheadFullException
 import ratpack.exec.Blocking
 import ratpack.exec.ExecResult
+import ratpack.exec.Promise
 import ratpack.test.exec.ExecHarness
 import spock.lang.AutoCleanup
 import spock.lang.Shared
@@ -51,7 +52,7 @@ class BulkheadTransformerSpec extends Specification {
                 times.getAndIncrement();
                 "s"
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         then:
@@ -75,7 +76,7 @@ class BulkheadTransformerSpec extends Specification {
                 times.getAndIncrement();
                 "r1"
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         and:
@@ -84,7 +85,7 @@ class BulkheadTransformerSpec extends Specification {
                 times.getAndIncrement();
                 "r2"
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         then:
@@ -117,7 +118,7 @@ class BulkheadTransformerSpec extends Specification {
                 times.getAndIncrement();
                 throw new RuntimeException("Expected")
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         and:
@@ -126,7 +127,7 @@ class BulkheadTransformerSpec extends Specification {
                 times.getAndIncrement();
                 "r2"
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         then:
@@ -147,6 +148,53 @@ class BulkheadTransformerSpec extends Specification {
         times.get() == 2
     }
 
+    def "transformer can be reused multiple times when upstream onError is handled"() {
+        given:
+        def bulkhead = buildBulkhead()
+        BulkheadTransformer<String> transformer = BulkheadTransformer.of(bulkhead)
+
+        and: "setup an event listener to track the number of onCallFinish"
+        def bulkheadEvents = bulkhead.getEventPublisher()
+        AtomicInteger timesOnCallFinished = new AtomicInteger(0)
+        bulkheadEvents.onCallFinished({ timesOnCallFinished.getAndIncrement()})
+
+
+        when: "The upstream has an error, but is swallowed by `onError`"
+        def r1 = ExecHarness.yieldSingle {
+            Blocking.<String> get {
+                throw new RuntimeException("Expected")
+            }
+                    .onError { e -> Promise.value("not foo") }
+                    .transform(transformer)
+        }
+
+        and: ""
+        def r2 = ExecHarness.yieldSingle {
+            Blocking.<String> get {
+                "r2"
+            }
+                    .transform(transformer)
+        }
+
+        then:
+        with(r1) {
+            value == null
+            !error
+            throwable == null
+        }
+
+        and:
+        with(r2) {
+            value == "r2"
+            !error
+            throwable == null
+        }
+
+        and:
+        timesOnCallFinished.get() == 1
+
+    }
+
     def "exception is thrown when execution is blocked with one execution"() {
         given:
         def bulkhead = buildBulkhead()
@@ -166,7 +214,7 @@ class BulkheadTransformerSpec extends Specification {
                     times.getAndIncrement()
                     "r"
                 }
-                .transform(transformer)
+                        .transform(transformer)
             }
         } as Callable<ExecResult<String>>)
 
@@ -176,7 +224,7 @@ class BulkheadTransformerSpec extends Specification {
             Blocking.<String> get {
                 assert false: "Should never be called"
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         then:
@@ -217,7 +265,7 @@ class BulkheadTransformerSpec extends Specification {
                     times.getAndIncrement()
                     "r"
                 }
-                .transform(transformer)
+                        .transform(transformer)
             }
         } as Callable<ExecResult<String>>)
 
@@ -227,7 +275,7 @@ class BulkheadTransformerSpec extends Specification {
             Blocking.<String> get {
                 assert false: "Should never be called"
             }
-            .transform(transformer)
+                    .transform(transformer)
         }
 
         then:

diff --git a/...groovy/io/github/resilience4j/ratpack/circuitbreaker/CircuitBreakerTransformerSpec.groovy b/...groovy/io/github/resilience4j/ratpack/circuitbreaker/CircuitBreakerTransformerSpec.groovy
@@ -18,8 +18,6 @@ package io.github.resilience4j.ratpack.circuitbreaker
 import io.github.resilience4j.circuitbreaker.CallNotPermittedException
 import io.github.resilience4j.circuitbreaker.CircuitBreaker
 import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig
-import io.github.resilience4j.retry.Retry
-import io.github.resilience4j.retry.RetryConfig
 import ratpack.exec.Blocking
 import ratpack.exec.Promise
 import ratpack.test.exec.ExecHarness