Merge pull request ReactiveX#246 from szmg/fix-retrofit-enqueued-call

make async retrofit call not make the request when circuit is open

Author: storozhukBM

resilience4j-retrofit/src/main/java/io/github/resilience4j/retrofit/RetrofitCircuitBreaker.java

@@ -58,6 +58,7 @@ public void enqueue(final Callback<T> callback) {
                     CircuitBreakerUtils.isCallPermitted(circuitBreaker);
                 } catch (CircuitBreakerOpenException cb) {
                     callback.onFailure(call, cb);
+                    return;
                 }
 
                 final StopWatch stopWatch = StopWatch.start(circuitBreaker.getName());

resilience4j-retrofit/src/test/java/io/github/resilience4j/retrofit/RetrofitCircuitBreakerTest.java

@@ -21,6 +21,8 @@
 import com.github.tomakehurst.wiremock.junit.WireMockRule;
 import io.github.resilience4j.circuitbreaker.CircuitBreaker;
 import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
+import io.github.resilience4j.circuitbreaker.CircuitBreakerOpenException;
+import okhttp3.Dispatcher;
 import okhttp3.OkHttpClient;
 import org.junit.Before;
 import org.junit.Rule;
@@ -34,6 +36,7 @@
 
 import static com.github.tomakehurst.wiremock.client.WireMock.*;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 
 /**
  * Tests the integration of the Retrofit HTTP client and {@link CircuitBreaker}
@@ -50,14 +53,16 @@ public class RetrofitCircuitBreakerTest {
 
     private CircuitBreaker circuitBreaker = CircuitBreaker.of("test", circuitBreakerConfig);
 
+    private OkHttpClient client;
+
     private RetrofitService service;
 
     @Before
     public void setUp() {
         this.circuitBreaker = CircuitBreaker.of("test", circuitBreakerConfig);
 
         final long TIMEOUT = 300; // ms
-        final OkHttpClient client = new OkHttpClient.Builder()
+        this.client = new OkHttpClient.Builder()
                 .connectTimeout(TIMEOUT, TimeUnit.MILLISECONDS)
                 .readTimeout(TIMEOUT, TimeUnit.MILLISECONDS)
                 .writeTimeout(TIMEOUT, TimeUnit.MILLISECONDS)
@@ -212,6 +217,27 @@ public void decorateUnsuccessfulEnqueuedCall() throws Throwable {
         assertThat(metrics.getNumberOfFailedCalls()).isEqualTo(1);
     }
 
+    @Test
+    public void shouldNotCallServiceOnEnqueueWhenOpen() throws Throwable {
+        stubFor(get(urlPathEqualTo("/greeting"))
+                .willReturn(aResponse()
+                        .withStatus(200)
+                        .withHeader("Content-Type", "text/plain")
+                        .withBody("hello world")));
+
+        circuitBreaker.transitionToOpenState();
+
+        try {
+            EnqueueDecorator.enqueue(service.greeting());
+            fail("CircuitBreakerOpenException was expected");
+        } catch (CircuitBreakerOpenException ignore) {
+
+        }
+
+        ensureAllRequestsAreExecuted(Duration.ofSeconds(1));
+        verify(0, getRequestedFor(urlPathEqualTo("/greeting")));
+    }
+
     @Test(expected = IllegalArgumentException.class)
     public void shouldThrowOnBadService() {
         BadRetrofitService badService = new Retrofit.Builder()
@@ -222,4 +248,16 @@ public void shouldThrowOnBadService() {
 
         badService.greeting();
     }
+
+    private void ensureAllRequestsAreExecuted(Duration timeout) throws InterruptedException {
+        long end = System.nanoTime() + timeout.toNanos();
+        Dispatcher dispatcher = client.dispatcher();
+        while (System.nanoTime() < end) {
+            if (dispatcher.queuedCallsCount() <= 0 && dispatcher.runningCallsCount() <= 0) {
+                return;
+            }
+            Thread.sleep(10);
+        }
+        fail("Timeout exceeded while waiting for requests to be finished");
+    }
 }