private TimeLimiterCoalition createTimeLimiter(String ruleName, CircuitBreakerRule rule) {
    if (rule == null) {
        return null;
    }
    Duration timeout = rule.getTimeoutDuration();
    if (rule.getEnabled() != null && rule.getEnabled() &&
            timeout != null && !timeout.isNegative() && !timeout.isZero()) {
        try {
            TimeLimiterConfig timeLimiterConfig = TimeLimiterConfig.custom()
                    .timeoutDuration(rule.getTimeoutDuration())
                    .cancelRunningFuture(rule.getCancelRunningFuture())
                    .build();
            TimeLimiter timeLimiter = TimeLimiter.of(timeLimiterConfig);
            TimeLimiterCoalition timeLimiterCoalition = new TimeLimiterCoalition(
                    timeLimiter, rule);
            return timeLimiterCoalition;
        } catch (Exception e) {
            logger.error("failed to create timeLimiter,name:{},id{}",
                    ruleName, rule.getRuleId(), e);
            return null;
        }
    }
    return null;
}
 

@Before
public void init() {
    TimeLimiterConfig timeLimiterConfig = TimeLimiterConfig.custom()
            .timeoutDuration(Duration.ofSeconds(1))
            .cancelRunningFuture(true)
            .build();
     timeLimiter = TimeLimiter.of(timeLimiterConfig);

    CustomizableThreadFactory factory = new CustomizableThreadFactory("timeLimiter-");
    factory.setDaemon(true);
    executorService = Executors.newCachedThreadPool(factory);

    CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig
            .custom()
            .enableAutomaticTransitionFromOpenToHalfOpen()
            .failureRateThreshold(50)
            .ringBufferSizeInClosedState(10)
            .ringBufferSizeInHalfOpenState(2)
            .build();

    circuitBreaker = CircuitBreaker.of("backendName", circuitBreakerConfig);
}
 

@Test
public void testTimeout() throws Throwable {
    Callable<Object> callable = () -> {
        Thread.sleep(10000);
        return null;
    };
    Supplier<Future<Object>> supplier = () -> executorService.submit(callable);
    // Wrap your call to BackendService.doSomething() in a future provided by your executor
    Callable<Object> result = TimeLimiter.decorateFutureSupplier(timeLimiter, supplier);
    long start = System.currentTimeMillis();
    try {
        result.call();
        fail("fail");
    } catch (Throwable t) {
        long cost = System.currentTimeMillis() - start;
        assertThat(cost, Matchers.lessThan(1010L));
        assertThat(t, Matchers.instanceOf(TimeoutException.class));
    }
}
 

@Override
public void publishMetrics(TimeLimiter timeLimiter) {

    String name = timeLimiter.getName();
    String successfulName = name(prefix, name, SUCCESSFUL);
    String failedName = name(prefix, name, FAILED);
    String timeoutName = name(prefix, name, TIMEOUT);

    Counter successes = metricRegistry.counter(successfulName);
    Counter failures = metricRegistry.counter(failedName);
    Counter timeouts = metricRegistry.counter(timeoutName);

    timeLimiter.getEventPublisher().onSuccess(event -> successes.inc());
    timeLimiter.getEventPublisher().onError(event -> failures.inc());
    timeLimiter.getEventPublisher().onTimeout(event -> timeouts.inc());

    List<String> metricNames = Arrays.asList(successfulName, failedName, timeoutName);
    metricsNameMap.put(name, new HashSet<>(metricNames));
}
 

/**
 * This method returns a desired Tax(taxAmount, taxPercentage) value when the TaxService is up. 
 * If the TaxService is down, it applies a combination of following fault tolerance patterns 
 * in a sequence: TimeLimiter, CircuitBreaker and Retry using a Callable. When all the attempts 
 * are exhausted it calls a fallback method to recover from failure and offers the default tax value.
 * 
 * @param amount
 * @return
 */
public Tax applyResiliencePatterns(BigDecimal amount) {

	CircuitBreaker circuitBreaker = configureCircuitBreaker();
	TimeLimiter timeLimiter = configureTimeLimiter();
	Retry retry = configureRetry();
	
	Supplier<CompletableFuture<Tax>> futureSupplier = () -> CompletableFuture.supplyAsync(() -> salesOrderService.supplyTax(amount));
	Callable<Tax> callable = TimeLimiter.decorateFutureSupplier(timeLimiter, futureSupplier);
	callable = CircuitBreaker.decorateCallable(circuitBreaker, callable);
	callable = Retry.decorateCallable(retry, callable);
	
	//Executing the decorated callable and recovering from any exception by calling the fallback method
	Try<Tax> result = Try.ofCallable(callable).recover(throwable -> taxServiceFallback(amount));
	return result.get();
}
 

private TimeLimiterMetrics(String prefix, Iterable<TimeLimiter> timeLimiters,
    MetricRegistry metricRegistry) {
    requireNonNull(prefix, PREFIX_NULL);
    requireNonNull(timeLimiters, ITERABLE_NULL);
    requireNonNull(metricRegistry);
    this.metricRegistry = metricRegistry;
    timeLimiters.forEach(timeLimiter -> {
            String name = timeLimiter.getName();
            Counter successes = metricRegistry.counter(name(prefix, name, SUCCESSFUL));
            Counter failures = metricRegistry.counter(name(prefix, name, FAILED));
            Counter timeouts = metricRegistry.counter(name(prefix, name, TIMEOUT));
            timeLimiter.getEventPublisher().onSuccess(event -> successes.inc());
            timeLimiter.getEventPublisher().onError(event -> failures.inc());
            timeLimiter.getEventPublisher().onTimeout(event -> timeouts.inc());
        }
    );
}
 

@Test
public void shouldThrowTimeoutExceptionWithCompletionStage() throws Exception {
    Duration timeoutDuration = Duration.ofMillis(300);
    TimeLimiter timeLimiter = TimeLimiter.of(timeoutDuration);
    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    Supplier<CompletionStage<Integer>> supplier = () -> CompletableFuture.supplyAsync(() -> {
        try {
            // sleep for timeout.
            Thread.sleep(500);
        } catch (InterruptedException e) {
            // nothing
        }
        return 0;
    });

    CompletionStage<Integer> decorated = TimeLimiter
        .decorateCompletionStage(timeLimiter, scheduler, supplier).get();
    Try<Integer> decoratedResult = Try.ofCallable(() -> decorated.toCompletableFuture().get());
    assertThat(decoratedResult.isFailure()).isTrue();
    assertThat(decoratedResult.getCause()).isInstanceOf(ExecutionException.class)
        .hasCauseExactlyInstanceOf(TimeoutException.class);
}
 

@Test
public void shouldThrowTimeoutExceptionAndNotInvokeCancel() throws Exception {
    Duration timeoutDuration = Duration.ofSeconds(1);
    TimeLimiter timeLimiter = TimeLimiter
        .of(TimeLimiterConfig.custom().timeoutDuration(timeoutDuration)
            .cancelRunningFuture(false).build());

    @SuppressWarnings("unchecked")
    Future<Integer> mockFuture = (Future<Integer>) mock(Future.class);

    Supplier<Future<Integer>> supplier = () -> mockFuture;
    given(mockFuture.get(timeoutDuration.toMillis(), TimeUnit.MILLISECONDS))
        .willThrow(new TimeoutException());

    Callable<Integer> decorated = TimeLimiter.decorateFutureSupplier(timeLimiter, supplier);
    Try<Integer> decoratedResult = Try.ofCallable(decorated);

    assertThat(decoratedResult.isFailure()).isTrue();
    assertThat(decoratedResult.getCause()).isInstanceOf(TimeoutException.class);

    then(mockFuture).should(never()).cancel(true);
}
 

@Test
public void shouldReturnResult() throws Exception {
    Duration timeoutDuration = Duration.ofSeconds(1);
    TimeLimiter timeLimiter = TimeLimiter.of(timeoutDuration);

    @SuppressWarnings("unchecked")
    Future<Integer> mockFuture = (Future<Integer>) mock(Future.class);

    Supplier<Future<Integer>> supplier = () -> mockFuture;
    given(mockFuture.get(timeoutDuration.toMillis(), TimeUnit.MILLISECONDS)).willReturn(42);

    int result = timeLimiter.executeFutureSupplier(supplier);
    assertThat(result).isEqualTo(42);

    int result2 = timeLimiter.decorateFutureSupplier(supplier).call();
    assertThat(result2).isEqualTo(42);
}
 

@Test
public void shouldReturnResultWithCompletionStage() throws Exception {
    Duration timeoutDuration = Duration.ofSeconds(1);
    TimeLimiter timeLimiter = TimeLimiter.of(timeoutDuration);
    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    Supplier<CompletionStage<Integer>> supplier = () -> CompletableFuture.supplyAsync(() -> {
        try {
            // sleep but not timeout.
            Thread.sleep(500);
        } catch (InterruptedException e) {
            // nothing
        }
        return 42;
    });

    int result = timeLimiter.executeCompletionStage(scheduler, supplier).toCompletableFuture()
        .get();
    assertThat(result).isEqualTo(42);

    int result2 = timeLimiter.decorateCompletionStage(scheduler, supplier).get()
        .toCompletableFuture().get();
    assertThat(result2).isEqualTo(42);
}
 

@Test
public void shouldRegisterMetrics() throws Exception {
    TimeLimiter timeLimiter = given(metricRegistry);
    String expectedPrefix = "resilience4j.timelimiter.testLimit.";
    Supplier<CompletableFuture<String>> futureSupplier = () ->
        CompletableFuture.completedFuture("Hello world");

    String result = timeLimiter.decorateFutureSupplier(futureSupplier).call();

    then(result).isEqualTo("Hello world");
    assertThat(metricRegistry).hasMetricsSize(3);
    assertThat(metricRegistry).counter(expectedPrefix + SUCCESSFUL)
        .hasValue(1L);
    assertThat(metricRegistry).counter(expectedPrefix + FAILED)
        .hasValue(0L);
    assertThat(metricRegistry).counter(expectedPrefix + TIMEOUT)
        .hasValue(0L);
}
 

@Test
@SuppressWarnings("unchecked")
public void timeLimiterPositiveWithSupplier() {
    TimeLimiterRegistry registry = new InMemoryTimeLimiterRegistry(config);
    Supplier<TimeLimiterConfig> timeLimiterConfigSupplier = mock(Supplier.class);
    given(timeLimiterConfigSupplier.get()).willReturn(config);

    TimeLimiter firstTimeLimiter = registry.timeLimiter("test", timeLimiterConfigSupplier);
    verify(timeLimiterConfigSupplier, times(1)).get();
    TimeLimiter sameAsFirst = registry.timeLimiter("test", timeLimiterConfigSupplier);
    verify(timeLimiterConfigSupplier, times(1)).get();
    TimeLimiter anotherLimit = registry.timeLimiter("test1", timeLimiterConfigSupplier);
    verify(timeLimiterConfigSupplier, times(2)).get();

    then(firstTimeLimiter).isEqualTo(sameAsFirst);
    then(firstTimeLimiter).isNotEqualTo(anotherLimit);
}
 

@Test
public void shouldRecordErrors() {
    TimeLimiter timeLimiter = TimeLimiter.of(TimeLimiterConfig.ofDefaults());
    metricRegistry.registerAll(TimeLimiterMetrics.ofTimeLimiter(timeLimiter));

    timeLimiter.onError(new RuntimeException());
    timeLimiter.onError(new RuntimeException());

    assertThat(metricRegistry).hasMetricsSize(3);
    assertThat(metricRegistry).counter(DEFAULT_PREFIX + SUCCESSFUL)
        .hasValue(0L);
    assertThat(metricRegistry).counter(DEFAULT_PREFIX + FAILED)
        .hasValue(2L);
    assertThat(metricRegistry).counter(DEFAULT_PREFIX + TIMEOUT)
        .hasValue(0L);
}
 

@Test
public void shouldThrowTimeoutException() {
    TimeLimiter timeLimiter = TimeLimiter.of("helloBackend", TimeLimiterConfig.custom()
        .timeoutDuration(Duration.ofMillis(100)).build());
    CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("helloBackend");
    ThreadPoolBulkhead bulkhead = ThreadPoolBulkhead.ofDefaults("helloBackend");
    CompletionStage<String> completionStage = Decorators
        .ofCallable(() -> {
            Thread.sleep(1000);
            return "Bla";
        })
        .withThreadPoolBulkhead(bulkhead)
        .withTimeLimiter(timeLimiter, Executors.newSingleThreadScheduledExecutor())
        .withCircuitBreaker(circuitBreaker)
        .get();

    assertThatThrownBy(() -> completionStage.toCompletableFuture().get())
        .hasCauseInstanceOf(TimeoutException.class);

    CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();
    assertThat(metrics.getNumberOfBufferedCalls()).isEqualTo(1);
    assertThat(metrics.getNumberOfFailedCalls()).isEqualTo(1);
}
 

@Test
public void testDecorateSupplierWithThreadPoolBulkhead()
    throws ExecutionException, InterruptedException {

    given(helloWorldService.returnHelloWorld()).willReturn("Hello world");
    CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("helloBackend");

    CompletableFuture<String> future = Decorators
        .ofSupplier(() -> helloWorldService.returnHelloWorld())
        .withThreadPoolBulkhead(ThreadPoolBulkhead.ofDefaults("helloBackend"))
        .withTimeLimiter(TimeLimiter.ofDefaults(), Executors.newSingleThreadScheduledExecutor())
        .withCircuitBreaker(circuitBreaker)
        .get().toCompletableFuture();

    String result = future.get();

    assertThat(result).isEqualTo("Hello world");
    CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();
    assertThat(metrics.getNumberOfBufferedCalls()).isEqualTo(1);
    assertThat(metrics.getNumberOfSuccessfulCalls()).isEqualTo(1);
    then(helloWorldService).should(times(1)).returnHelloWorld();
}
 

@Test
public void shouldRecordTimeouts() {
    TimeLimiter timeLimiter = TimeLimiter.of(TimeLimiterConfig.custom()
        .timeoutDuration(Duration.ZERO)
        .build());
    metricRegistry.registerAll(TimeLimiterMetrics.ofTimeLimiter(timeLimiter));

    timeLimiter.onError(new TimeoutException());
    timeLimiter.onError(new TimeoutException());

    assertThat(metricRegistry).hasMetricsSize(3);
    assertThat(metricRegistry).counter(DEFAULT_PREFIX + SUCCESSFUL)
        .hasValue(0L);
    assertThat(metricRegistry).counter(DEFAULT_PREFIX + FAILED)
        .hasValue(0L);
    assertThat(metricRegistry).counter(DEFAULT_PREFIX + TIMEOUT)
        .hasValue(2L);
}
 

public InMemoryTimeLimiterRegistry(Map<String, TimeLimiterConfig> configs,
    RegistryEventConsumer<TimeLimiter> registryEventConsumer,
    io.vavr.collection.Map<String, String> tags) {
    this(configs.getOrDefault(DEFAULT_CONFIG, TimeLimiterConfig.ofDefaults()), registryEventConsumer,
        tags);
    this.configurations.putAll(configs);
}
 

public InMemoryTimeLimiterRegistry(Map<String, TimeLimiterConfig> configs,
    List<RegistryEventConsumer<TimeLimiter>> registryEventConsumers,
    io.vavr.collection.Map<String, String> tags) {
    this(configs.getOrDefault(DEFAULT_CONFIG, TimeLimiterConfig.ofDefaults()),
        registryEventConsumers, tags);
    this.configurations.putAll(configs);
}
 

@Override
public TimeLimiter timeLimiter(String name,
    TimeLimiterConfig timeLimiterConfig,
    io.vavr.collection.Map<String, String> tags) {
    return computeIfAbsent(name, () -> TimeLimiter.of(name,
        Objects.requireNonNull(timeLimiterConfig, CONFIG_MUST_NOT_BE_NULL), getAllTags(tags)));
}
 

@Override
public TimeLimiter timeLimiter(String name, String configName,
    io.vavr.collection.Map<String, String> tags) {
    TimeLimiterConfig config = getConfiguration(configName)
        .orElseThrow(() -> new ConfigurationNotFoundException(configName));
    return timeLimiter(name, config, tags);
}
 

@Test
public void shouldReturnCorrectTimeoutDuration() {
    Duration timeoutDuration = Duration.ofSeconds(1);
    TimeLimiter timeLimiter = TimeLimiter.of(timeoutDuration);
    assertThat(timeLimiter).isNotNull();
    assertThat(timeLimiter.getTimeLimiterConfig().getTimeoutDuration())
        .isEqualTo(timeoutDuration);
}
 

@Test
public void unwrapExecutionException() {
    TimeLimiter timeLimiter = TimeLimiter.ofDefaults();
    ExecutorService executorService = Executors.newSingleThreadExecutor();

    Supplier<Future<Integer>> supplier = () -> executorService.submit(() -> {
        throw new RuntimeException();
    });
    Callable<Integer> decorated = TimeLimiter.decorateFutureSupplier(timeLimiter, supplier);

    Try<Integer> decoratedResult = Try.ofCallable(decorated);

    assertThat(decoratedResult.getCause() instanceof RuntimeException).isTrue();
}
 

@Override
protected TimeLimiter given(MetricRegistry metricRegistry) {
    TimeLimiterRegistry timeLimiterRegistry = TimeLimiterRegistry.ofDefaults();
    TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("testLimit");
    metricRegistry.registerAll(TimeLimiterMetrics.ofTimeLimiterRegistry(timeLimiterRegistry));

    return timeLimiter;
}
 

@Override
public void publishMetrics(TimeLimiter entry) {
    String name = entry.getName();
    entry.getEventPublisher()
        .onSuccess(event -> callsCounter.labels(name, KIND_SUCCESSFUL).inc())
        .onError(event -> callsCounter.labels(name, KIND_FAILED).inc())
        .onTimeout(event -> callsCounter.labels(name, KIND_TIMEOUT).inc());
}
 

@Override
protected TimeLimiter given(String prefix, MetricRegistry metricRegistry) {
    TimeLimiterRegistry timeLimiterRegistry = TimeLimiterRegistry.ofDefaults();
    TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("testLimit");
    metricRegistry
        .registerAll(TimeLimiterMetrics.ofTimeLimiterRegistry(prefix, timeLimiterRegistry));

    return timeLimiter;
}
 

@Test
public void customMetricNamesOverrideDefaultOnes() {
    TimeLimiterMetricsPublisher.MetricNames names = TimeLimiterMetricsPublisher.MetricNames
        .custom()
        .callsMetricName("custom_calls")
        .build();
    CollectorRegistry customRegistry = new CollectorRegistry();
    TimeLimiterMetricsPublisher timeLimiterMetricsPublisher = new TimeLimiterMetricsPublisher(
        names);
    timeLimiterMetricsPublisher.register(customRegistry);

    TimeLimiterRegistry timeLimiterRegistry = TimeLimiterRegistry
        .of(TimeLimiterConfig.ofDefaults(), timeLimiterMetricsPublisher);
    TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("backendA");

    timeLimiter.onSuccess();
    timeLimiter.onError(new RuntimeException());
    timeLimiter.onError(new TimeoutException());

    Double successfulCalls = getSampleValue(customRegistry, "custom_calls", KIND_SUCCESSFUL);
    Double failedCalls = getSampleValue(customRegistry, "custom_calls", KIND_FAILED);
    Double timeoutCalls = getSampleValue(customRegistry, "custom_calls", KIND_TIMEOUT);

    assertThat(successfulCalls).isNotNull();
    assertThat(failedCalls).isNotNull();
    assertThat(timeoutCalls).isNotNull();
}
 

@Override
public void bindTo(MeterRegistry registry) {
    for (TimeLimiter timeLimiter : timeLimiterRegistry.getAllTimeLimiters()) {
        addMetrics(registry, timeLimiter);
    }
    timeLimiterRegistry.getEventPublisher()
        .onEntryAdded(event -> addMetrics(registry, event.getAddedEntry()));
    timeLimiterRegistry.getEventPublisher()
        .onEntryRemoved(event -> removeMetrics(registry, event.getRemovedEntry().getName()));
    timeLimiterRegistry.getEventPublisher().onEntryReplaced(event -> {
        removeMetrics(registry, event.getOldEntry().getName());
        addMetrics(registry, event.getNewEntry());
    });
}
 

protected void registerMetrics(MeterRegistry meterRegistry, TimeLimiter timeLimiter, List<Tag> customTags) {
    // Remove previous meters before register
    removeMetrics(meterRegistry, timeLimiter.getName());

    Counter successes = Counter.builder(names.getCallsMetricName())
        .description("The number of successful calls")
        .tag(TagNames.NAME, timeLimiter.getName())
        .tag(TagNames.KIND, KIND_SUCCESSFUL)
        .tags(customTags)
        .register(meterRegistry);
    Counter failures = Counter.builder(names.getCallsMetricName())
        .description("The number of failed calls")
        .tag(TagNames.NAME, timeLimiter.getName())
        .tag(TagNames.KIND, KIND_FAILED)
        .tags(customTags)
        .register(meterRegistry);
    Counter timeouts = Counter.builder(names.getCallsMetricName())
        .description("The number of timed out calls")
        .tag(TagNames.NAME, timeLimiter.getName())
        .tag(TagNames.KIND, KIND_TIMEOUT)
        .tags(customTags)
        .register(meterRegistry);

    timeLimiter.getEventPublisher()
        .onSuccess(event -> successes.increment())
        .onError(event -> failures.increment())
        .onTimeout(event -> timeouts.increment());

    List<Meter.Id> ids = Arrays.asList(successes.getId(), failures.getId(), timeouts.getId());
    meterIdMap.put(timeLimiter.getName(), new HashSet<>(ids));
}
 

@Test
public void shouldReplaceMetrics() {
    Counter before = meterRegistry.get(DEFAULT_TIME_LIMITER_CALLS).counter();
    assertThat(before).isNotNull();
    assertThat(before.count()).isEqualTo(0);
    assertThat(before.getId().getTag(TagNames.NAME)).isEqualTo(timeLimiter.getName());

    timeLimiterRegistry.replace(timeLimiter.getName(), TimeLimiter.ofDefaults());

    Counter after = meterRegistry.get(DEFAULT_TIME_LIMITER_CALLS).counter();
    assertThat(after).isNotNull();
    assertThat(after.count()).isEqualTo(0);
    assertThat(after.getId().getTag(TagNames.NAME))
        .isEqualTo(TimeLimiter.ofDefaults().getName());
}
 

@Test
public void shouldUseCustomPrefix() throws Exception {
    TimeLimiter timeLimiter = given("testPre", metricRegistry);
    String expectedPrefix = "testPre.testLimit.";
    Supplier<CompletableFuture<String>> futureSupplier = () ->
        CompletableFuture.completedFuture("Hello world");

    String result = timeLimiter.decorateFutureSupplier(futureSupplier).call();

    then(result).isEqualTo("Hello world");
    assertThat(metricRegistry).hasMetricsSize(3);
    assertThat(metricRegistry).counter(expectedPrefix + SUCCESSFUL)
        .hasValue(1L);
    assertThat(metricRegistry).counter(expectedPrefix + FAILED)
        .hasValue(0L);
    assertThat(metricRegistry).counter(expectedPrefix + TIMEOUT)
        .hasValue(0L);
}