CircuitBreakerImpl implements CircuitBreaker<ET>, CircuitBreaker.ManualModeTask<ET> { @Override public ET executeBlockingCall(Callable<ET> eventSupplier) throws Exception { throwExceptionIfCircuitBreakerIsOpen(); ET event = null; Exception eventException = null; long beforeCallNanos = System.nanoTime(); try { event = eventSupplier.call(); } catch (Exception ex) { eventException = ex; } long afterCallNanos = System.nanoTime(); Exception timeoutException = null; if (callTimeoutNanos.isPresent()) { long timeoutNanos = callTimeoutNanos.get(); if ((afterCallNanos - beforeCallNanos) > timeoutNanos) { timeoutException = generateNewCircuitBreakerTimeoutException(timeoutNanos); } } Exception exceptionToHandle = (timeoutException == null) ? eventException : timeoutException; if (exceptionToHandle != null) { handleException(exceptionToHandle); throw exceptionToHandle; } handleEvent(event); return event; } CircuitBreakerImpl();  CircuitBreakerImpl(Optional<ScheduledExecutorService> scheduler,
                              Optional<ExecutorService> stateChangeNotificationExecutor,
                              Optional<Integer> maxConsecutiveFailuresAllowed,
                              Optional<Duration> resetTimeout,
                              Optional<Duration> callTimeout,
                              Optional<String> id,
                              Optional<BreakingEventStrategy<ET>> breakingEventStrategy,
                              Optional<BreakingExceptionStrategy> breakingExceptionStrategy); static Builder<E> newBuilder(); static Builder<E> newBuilder(BreakingEventStrategy<E> breakingEventStrategy); @Override ManualModeTask<ET> newManualModeTask(); @Override void throwExceptionIfCircuitBreakerIsOpen(); @Override void handleEvent(ET event); @Override void handleException(Throwable throwable); @Override CircuitBreaker<ET> originatingCircuitBreaker(); @Override CompletableFuture<ET> executeAsyncCall(Supplier<CompletableFuture<ET>> eventFutureSupplier); @Override ET executeBlockingCall(Callable<ET> eventSupplier); @Override CircuitBreakerImpl<ET> onClose(Runnable listener); @Override CircuitBreakerImpl<ET> onHalfOpen(Runnable listener); @Override CircuitBreakerImpl<ET> onOpen(Runnable listener); @Override String getId(); static final String DEFAULT_MAX_CONSECUTIVE_FAILURES_ALLOWED_SYSTEM_PROP_KEY; static final String DEFAULT_RESET_TIMEOUT_SECONDS_SYSTEM_PROP_KEY; static final int DEFAULT_MAX_CONSECUTIVE_FAILURES_ALLOWED; static final long DEFAULT_RESET_TIMEOUT_SECONDS; static final int FALLBACK_DEFAULT_MAX_CONSECUTIVE_FAILURES_ALLOWED; static final long FALLBACK_DEFAULT_RESET_TIMEOUT_SECONDS; static final BreakingEventStrategy<?> DEFAULT_BREAKING_EVENT_STRATEGY; static final BreakingExceptionStrategy DEFAULT_BREAKING_EXCEPTION_STRATEGY; }
@Test public void executeBlockingCall_explodes_immediately_if_circuit_is_OPEN() { Callable<String> supplierMock = mock(Callable.class); Whitebox.setInternalState(cbSpy, "currentState", State.OPEN); Throwable cbExplosion = catchThrowable(() -> cbSpy.executeBlockingCall(supplierMock)); assertThat(cbExplosion) .isNotNull() .isInstanceOf(CircuitBreakerOpenException.class); verifyZeroInteractions(supplierMock); }
@Test public void executeBlockingCall_throws_CircuitBreakerTimeoutException_after_supplier_completes_if_supplier_takes_longer_than_callTimeout() { long callTimeoutMillis = 50; long callExecutionTimeMillis = 100; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); Callable<String> supplier = () -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } return "foo"; }; long startTimeMillis = System.currentTimeMillis(); Throwable cbExplosion = catchThrowable(() -> cbSpy.executeBlockingCall(supplier)); long endTimeMillis = System.currentTimeMillis(); assertThat(cbExplosion) .isNotNull() .isInstanceOf(CircuitBreakerTimeoutException.class); assertThat((endTimeMillis - startTimeMillis)).isGreaterThanOrEqualTo(callExecutionTimeMillis); }
@Test public void executeBlockingCall_does_not_throw_CircuitBreakerTimeoutException_if_call_takes_shorter_than_callTimeout() { long callTimeoutMillis = 100; long callExecutionTimeMillis = 50; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); Callable<String> supplier = () -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } return "foo"; }; long startTimeMillis = System.currentTimeMillis(); Throwable cbExplosion = catchThrowable(() -> cbSpy.executeBlockingCall(supplier)); long endTimeMillis = System.currentTimeMillis(); assertThat(cbExplosion).isNull(); assertThat((endTimeMillis - startTimeMillis)).isGreaterThanOrEqualTo(callExecutionTimeMillis); }
@Test public void executeBlockingCall_gives_precedence_to_CircuitBreakerTimeoutException_if_supplier_takes_longer_than_callTimeout_even_if_supplier_throws_exception() { long callTimeoutMillis = 50; long callExecutionTimeMillis = 100; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); Callable<String> supplier = () -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } throw new Exception("crunch"); }; long startTimeMillis = System.currentTimeMillis(); Throwable cbExplosion = catchThrowable(() -> cbSpy.executeBlockingCall(supplier)); long endTimeMillis = System.currentTimeMillis(); assertThat(cbExplosion) .isNotNull() .isInstanceOf(CircuitBreakerTimeoutException.class); assertThat((endTimeMillis - startTimeMillis)).isGreaterThanOrEqualTo(callExecutionTimeMillis); }