CircuitBreakerImpl implements CircuitBreaker<ET>, CircuitBreaker.ManualModeTask<ET> { @Override public CompletableFuture<ET> executeAsyncCall(Supplier<CompletableFuture<ET>> eventFutureSupplier) throws CircuitBreakerOpenException { throwExceptionIfCircuitBreakerIsOpen(); CompletableFuture<ET> eventFuture = eventFutureSupplier.get(); eventFuture.whenComplete((event, error) -> { if (error == null) handleEvent(event); else handleException(error); }); callTimeoutNanos.ifPresent(timeoutNanos -> { ScheduledFuture<?> timeoutCheckFuture = scheduler.schedule(() -> { if (!eventFuture.isDone()) { eventFuture.completeExceptionally(generateNewCircuitBreakerTimeoutException(timeoutNanos)); } }, timeoutNanos, TimeUnit.NANOSECONDS); eventFuture.whenComplete((event, error) -> { if (!timeoutCheckFuture.isDone()) timeoutCheckFuture.cancel(false); }); }); return eventFuture; } 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 executeAsyncCall_explodes_immediately_if_circuit_is_OPEN() { Supplier<CompletableFuture<String>> supplierMock = mock(Supplier.class); Whitebox.setInternalState(cbSpy, "currentState", State.OPEN); Throwable cbExplosion = catchThrowable(() -> cbSpy.executeAsyncCall(supplierMock)); assertThat(cbExplosion) .isNotNull() .isInstanceOf(CircuitBreakerOpenException.class); verifyZeroInteractions(supplierMock); }
@Test public void executeAsyncCall_returns_future_from_supplier() { String uuid = UUID.randomUUID().toString(); CompletableFuture<String> future = CompletableFuture.completedFuture(uuid); Supplier<CompletableFuture<String>> supplier = () -> future; CompletableFuture<String> result = cbSpy.executeAsyncCall(supplier); assertThat(result).isSameAs(future); assertThat(result.join()).isEqualTo(uuid); }
@Test public void executeAsyncCall_short_circuit_times_out_the_call_if_it_takes_longer_than_callTimeout() { long callTimeoutMillis = 50; long callExecutionTimeMillis = 150; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } return "foo"; }); long startTimeMillis = System.currentTimeMillis(); CompletableFuture<String> result = cbSpy.executeAsyncCall(() -> future); Exception exThrownByFuture = null; try { result.join(); } catch (Exception ex) { exThrownByFuture = ex; } long endTimeMillis = System.currentTimeMillis(); assertThat(result.isCompletedExceptionally()).isTrue(); assertThat(exThrownByFuture) .isNotNull() .isInstanceOf(CompletionException.class) .hasCauseInstanceOf(CircuitBreakerTimeoutException.class); assertThat((endTimeMillis - startTimeMillis)).isLessThan(callExecutionTimeMillis); }
@Test public void executeAsyncCall_does_not_time_out_the_call_if_it_takes_shorter_than_callTimeout() throws InterruptedException { long callTimeoutMillis = 100; long callExecutionTimeMillis = 50; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } return "foo"; }); long startTimeMillis = System.currentTimeMillis(); CompletableFuture<String> result = cbSpy.executeAsyncCall(() -> future); Exception exThrownByFuture = null; try { result.join(); } catch (Exception ex) { exThrownByFuture = ex; } long endTimeMillis = System.currentTimeMillis(); Thread.sleep(callTimeoutMillis + 1); assertThat(result.isCompletedExceptionally()).isFalse(); assertThat(exThrownByFuture).isNull(); assertThat((endTimeMillis - startTimeMillis)).isGreaterThanOrEqualTo(callExecutionTimeMillis); }
@Test public void executeAsyncCall_cancels_timeout_check_if_future_completes_first() throws InterruptedException { long callTimeoutMillis = 100; long callExecutionTimeMillis = 50; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } return "foo"; }); ScheduledExecutorService schedulerMock = mock(ScheduledExecutorService.class); ScheduledFuture scheduledFutureMock = mock(ScheduledFuture.class); doReturn(scheduledFutureMock).when(schedulerMock) .schedule(any(Runnable.class), any(Long.class), any(TimeUnit.class)); doReturn(false).when(scheduledFutureMock).isDone(); Whitebox.setInternalState(cbSpy, "scheduler", schedulerMock); CompletableFuture<String> result = cbSpy.executeAsyncCall(() -> future); result.join(); Thread.sleep(50); verify(scheduledFutureMock).cancel(false); }
@Test public void executeAsyncCall_does_not_cancel_timeout_check_if_future_takes_too_long() throws InterruptedException { long callTimeoutMillis = 100; long callExecutionTimeMillis = 150; Whitebox .setInternalState(cbSpy, "callTimeoutNanos", Optional.of(Duration.ofMillis(callTimeoutMillis).toNanos())); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(callExecutionTimeMillis); } catch (InterruptedException e) { throw new RuntimeException(e); } return "foo"; }); ScheduledExecutorService schedulerMock = mock(ScheduledExecutorService.class); ScheduledFuture scheduledFutureMock = mock(ScheduledFuture.class); doReturn(scheduledFutureMock).when(schedulerMock) .schedule(any(Runnable.class), any(Long.class), any(TimeUnit.class)); doReturn(true).when(scheduledFutureMock).isDone(); Whitebox.setInternalState(cbSpy, "scheduler", schedulerMock); CompletableFuture<String> result = cbSpy.executeAsyncCall(() -> future); result.join(); Thread.sleep(50); verify(scheduledFutureMock, times(0)).cancel(anyBoolean()); }