ReactiveX#245 Fix CircuitBreakerSubscriber for Reactor doesn't count successes when using Mono/Flux.toFuture()

Circuit breaker on Mono will count success onNext or onComplete.

Author: madgnome

resilience4j-reactor/src/main/java/io/github/resilience4j/reactor/circuitbreaker/operator/CircuitBreakerSubscriber.java

@@ -22,6 +22,8 @@
 import org.reactivestreams.Subscriber;
 import reactor.core.CoreSubscriber;
 
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -33,23 +35,39 @@ class CircuitBreakerSubscriber<T> extends ResilienceBaseSubscriber<T> {
 
     private final CircuitBreaker circuitBreaker;
     private StopWatch stopWatch;
+    private final boolean singleProducer;
+
+    private volatile int successSignaled = 0;
+    private static final AtomicIntegerFieldUpdater<CircuitBreakerSubscriber> SUCCESS_SIGNALED =
+            AtomicIntegerFieldUpdater.newUpdater(CircuitBreakerSubscriber.class, "successSignaled");
 
     public CircuitBreakerSubscriber(CircuitBreaker circuitBreaker,
-                                    CoreSubscriber<? super T> actual) {
+                                    CoreSubscriber<? super T> actual,
+                                    boolean singleProducer) {
         super(actual);
         this.circuitBreaker = requireNonNull(circuitBreaker);
+        this.singleProducer = singleProducer;
     }
 
     @Override
     protected void hookOnNext(T value) {
+        if (singleProducer) {
+            if (SUCCESS_SIGNALED.compareAndSet(this, 0, 1)) {
+                markSuccess();
+            }
+        }
+
         if (notCancelled() && wasCallPermitted()) {
             actual.onNext(value);
         }
     }
 
     @Override
     protected void hookOnComplete() {
-        markSuccess();
+        if (SUCCESS_SIGNALED.compareAndSet(this, 0, 1)) {
+            markSuccess();
+        }
+
         if (wasCallPermitted()) {
             actual.onComplete();
         }

resilience4j-reactor/src/main/java/io/github/resilience4j/reactor/circuitbreaker/operator/FluxCircuitBreaker.java

@@ -31,7 +31,7 @@ public FluxCircuitBreaker(Flux<? extends T> source, CircuitBreaker circuitBreake
 
     @Override
     public void subscribe(CoreSubscriber<? super T> actual) {
-        source.subscribe(new CircuitBreakerSubscriber<>(circuitBreaker, actual));
+        source.subscribe(new CircuitBreakerSubscriber<>(circuitBreaker, actual, false));
     }
 
 }

resilience4j-reactor/src/main/java/io/github/resilience4j/reactor/circuitbreaker/operator/MonoCircuitBreaker.java

@@ -30,6 +30,6 @@ public MonoCircuitBreaker(Mono<? extends T> source, CircuitBreaker circuitBreake
 
     @Override
     public void subscribe(CoreSubscriber<? super T> actual) {
-        source.subscribe(new CircuitBreakerSubscriber<>(circuitBreaker, actual));
+        source.subscribe(new CircuitBreakerSubscriber<>(circuitBreaker, actual, true));
     }
 }

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/circuitbreaker/operator/CircuitBreakerSubscriberWhiteboxVerification.java

@@ -31,7 +31,7 @@ public CircuitBreakerSubscriberWhiteboxVerification() {
 
     @Override
     public Subscriber<Integer> createSubscriber(WhiteboxSubscriberProbe<Integer> probe) {
-        return new CircuitBreakerSubscriber<Integer>(CircuitBreaker.ofDefaults("verification"), MonoProcessor.create()) {
+        return new CircuitBreakerSubscriber<Integer>(CircuitBreaker.ofDefaults("verification"), MonoProcessor.create(), false) {
 
             @Override
             protected void hookOnSubscribe(Subscription subscription) {

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/circuitbreaker/operator/MonoCircuitBreakerTest.java

@@ -22,6 +22,9 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.Assert.fail;
 
 public class MonoCircuitBreakerTest extends CircuitBreakerAssertions {
 
@@ -36,6 +39,16 @@ public void shouldEmitEvent() {
         assertSingleSuccessfulCall();
     }
 
+    @Test
+    public void shouldEmptyMonoShouldBeSuccessful() {
+        StepVerifier.create(
+                Mono.empty()
+                        .transform(CircuitBreakerOperator.of(circuitBreaker)))
+                .verifyComplete();
+
+        assertSingleSuccessfulCall();
+    }
+
     @Test
     public void shouldPropagateError() {
         StepVerifier.create(
@@ -82,4 +95,19 @@ public void shouldEmitErrorWithCircuitBreakerOpenException() {
 
         assertNoRegisteredCall();
     }
+
+    @Test
+    public void shouldRecordSuccessWhenUsingToFuture() {
+        try {
+            Mono.just("Event")
+                    .transform(CircuitBreakerOperator.of(circuitBreaker))
+                    .toFuture()
+                    .get();
+
+            assertSingleSuccessfulCall();
+        } catch (InterruptedException | ExecutionException e) {
+            fail();
+        }
+
+    }
 }