COMMAND git -C /tmp/java-series-resilience4j-v2.3.0 rev-parse HEAD
c2c6575114fc0650177fb21e1ff967f14acde39c

READ README.adoc
= Fault tolerance library designed for functional programming
:author: Robert Winkler and Bohdan Storozhuk
:icons:
:toc: macro
:numbered: 1
ifdef::env-github[]
:tip-caption: :bulb:
:note-caption: :information_source:
:important-caption: :heavy_exclamation_mark:
:caution-caption: :fire:
:warning-caption: :warning:
endif::[]

image:https://github.com/resilience4j/resilience4j/actions/workflows/gradle-build.yml/badge.svg["Build Status"]
image:https://img.shields.io/nexus/r/io.github.resilience4j/resilience4j-circuitbreaker?server=https%3A%2F%2Foss.sonatype.org["Release"]
image:https://img.shields.io/nexus/s/io.github.resilience4j/resilience4j-circuitbreaker?server=https%3A%2F%2Foss.sonatype.org["Snapshot"]
image:http://img.shields.io/badge/license-ASF2-blue.svg["Apache License 2", link="http://www.apache.org/licenses/LICENSE-2.0.txt"]

image:https://sonarcloud.io/api/project_badges/measure?project=resilience4j_resilience4j&metric=coverage["Coverage", link="https://sonarcloud.io/dashboard?id=resilience4j_resilience4j"]
image:https://sonarcloud.io/api/project_badges/measure?project=resilience4j_resilience4j&metric=sqale_rating["Maintainability", link="https://sonarcloud.io/dashboard?id=resilience4j_resilience4j"]
image:https://sonarcloud.io/api/project_badges/measure?project=resilience4j_resilience4j&metric=reliability_rating["Reliability", link="https://sonarcloud.io/dashboard?id=resilience4j_resilience4j"]
image:https://sonarcloud.io/api/project_badges/measure?project=resilience4j_resilience4j&metric=security_rating["Security", link="https://sonarcloud.io/dashboard?id=resilience4j_resilience4j"]
image:https://sonarcloud.io/api/project_badges/measure?project=resilience4j_resilience4j&metric=vulnerabilities["Vulnerabilities", link="https://sonarcloud.io/dashboard?id=resilience4j_resilience4j"]
image:https://sonarcloud.io/api/project_badges/measure?project=resilience4j_resilience4j&metric=bugs["Bugs", link="https://sonarcloud.io/dashboard?id=resilience4j_resilience4j"]

image:https://raw.githubusercontent.com/vshymanskyy/StandWithUkraine/main/banner2-direct.svg["SWUbanner",link="https://vshymanskyy.github.io/StandWithUkraine"]

toc::[]

== Introduction

Resilience4j is a lightweight fault tolerance library designed for functional programming.
Resilience4j provides higher-order functions (decorators) to enhance any functional interface,
lambda expression or method reference with a Circuit Breaker, Rate Limiter, Retry or Bulkhead.
You can stack more than one decorator on any functional interface, lambda expression or method reference.
The advantage is that you have the choice to select the decorators you need and nothing else.

Resilience4j 2 requires Java 17.

[source,java]
----
// Create a CircuitBreaker with default configuration
CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("backendService");

// Create a Retry with default configuration
// 3 retry attempts and a fixed time interval between retries of 500ms
Retry retry = Retry.ofDefaults("backendService");

// Create a Bulkhead with default configuration
Bulkhead bulkhead = Bulkhead.ofDefaults("backendService");

Supplier<String> supplier = () -> backendService
  .doSomething(param1, param2);

// Decorate your call to backendService.doSomething()
// with a Bulkhead, CircuitBreaker and Retry
// **note: you will need the resilience4j-all dependency for this
Supplier<String> decoratedSupplier = Decorators.ofSupplier(supplier)
  .withCircuitBreaker(circuitBreaker)
  .withBulkhead(bulkhead)
  .withRetry(retry)
  .decorate();

// Execute the decorated supplier and recover from any exception
String result = Try.ofSupplier(decoratedSupplier)
  .recover(throwable -> "Hello from Recovery").get();

// When you don't want to decorate your lambda expression,
// but just execute it and protect the call by a CircuitBreaker.
String result = circuitBreaker
  .executeSupplier(backendService::doSomething);

// You can also run the supplier asynchronously in a ThreadPoolBulkhead
 ThreadPoolBulkhead threadPoolBulkhead = ThreadPoolBulkhead
  .ofDefaults("backendService");

// The Scheduler is needed to schedule a timeout on a non-blocking CompletableFuture
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
TimeLimiter timeLimiter = TimeLimiter.of(Duration.ofSeconds(1));

CompletableFuture<String> future = Decorators.ofSupplier(supplier)
    .withThreadPoolBulkhead(threadPoolBulkhead)
    .withTimeLimiter(timeLimiter, scheduler)
    .withCircuitBreaker(circuitBreaker)
    .withFallback(asList(TimeoutException.class, CallNotPermittedException.class, BulkheadFullException.class),
      throwable -> "Hello from Recovery")
    .get().toCompletableFuture();
----

NOTE: With Resilience4j you don’t have to go all-in, you can
https://mvnrepository.com/artifact/io.github.resilience4j[*pick what you need*].

==  Documentation

Setup and usage is described in our *https://resilience4j.readme.io/docs[User Guide]*.

- https://github.com/resilience4j-docs-ja/resilience4j-docs-ja[有志による日本語訳(非公式) Japanese translation by volunteers(Unofficial)]

- https://github.com/lmhmhl/Resilience4j-Guides-Chinese[这是Resilience4j的非官方中文文档 Chinese translation by volunteers(Unofficial)]

== Overview

Resilience4j provides several core modules:

* resilience4j-circuitbreaker: Circuit breaking
* resilience4j-ratelimiter: Rate limiting
* resilience4j-bulkhead: Bulkheading
* resilience4j-retry: Automatic retrying (sync and async)
* resilience4j-timelimiter: Timeout handling
* resilience4j-cache: Result caching

There are also add-on modules for metrics, Feign, Kotlin, Spring, Ratpack, Vertx, RxJava2 and more.

NOTE: Find out full list of modules in our *https://resilience4j.readme.io/docs#section-modularization[User Guide]*.

TIP: For core modules package or `+Decorators+` builder see *https://mvnrepository.com/artifact/io.github.resilience4j/resilience4j-all[resilience4j-all]*.

== Resilience patterns

[cols="<.<*", options="header"]
|===
|name |how does it work? |description |links

|*Retry*
|repeats failed executions
|Many faults are transient and may self-correct after a short delay.
|<<circuitbreaker-retry-fallback,overview>>,
https://resilience4j.readme.io/docs/retry[documentation],
https://resilience4j.readme.io/docs/getting-started-3#annotations[Spring]

|**Circuit Breaker**
|temporary blocks possible failures
|When a system is seriously struggling, failing fast is better than making clients wait.
|<<circuitbreaker-retry-fallback,overview>>,
https://resilience4j.readme.io/docs/circuitbreaker[documentation],
https://resilience4j.readme.io/docs/feign[Feign],
https://resilience4j.readme.io/docs/getting-started-3#annotations[Spring]

|**Rate Limiter**
|limits executions/period
|Limit the rate of incoming requests.
|<<ratelimiter,overview>>,
https://resilience4j.readme.io/docs/ratelimiter[documentation],
https://resilience4j.readme.io/docs/feign[Feign],
https://resilience4j.readme.io/docs/getting-started-3#annotations[Spring]

|**Time Limiter**
|limits duration of execution
|Beyond a certain wait interval, a successful result is unlikely.
|https://resilience4j.readme.io/docs/timeout[documentation],
https://resilience4j.readme.io/docs/getting-started-3#annotations[Spring]

|**Bulkhead**
|limits concurrent executions
|Resources are isolated into pools so that if one fails, the others will continue working.
|<<bulkhead,overview>>,
https://resilience4j.readme.io/docs/bulkhead[documentation],
https://resilience4j.readme.io/docs/getting-started-3#annotations[Spring]

|**Cache**
|memorizes a successful result
|Some proportion of requests may be similar.
|https://resilience4j.readme.io/docs/cache[documentation]

|**Fallback**
|provides an alternative result for failures
|Things will still fail - plan what you will do when that happens.
|<<circuitbreaker-retry-fallback,Try::recover>>,
https://resilience4j.readme.io/docs/getting-started-3#section-annotations[Spring],
https://resilience4j.readme.io/docs/feign[Feign]

|===

_Above table is based on https://github.com/App-vNext/Polly#resilience-policies[Polly: resilience policies]._

NOTE: To find more information about resilience patterns check link:#Talks[*Talks*] section.
Find out more about components in our *https://resilience4j.readme.io/docs/getting-started-2[User Guide]*.

== Spring Boot

Setup and usage in Spring Boot 2 is demonstrated https://github.com/resilience4j/resilience4j-spring-boot2-demo[here].

== Usage examples

[[circuitbreaker-retry-fallback]]
=== CircuitBreaker, Retry and Fallback

The following example shows how to decorate a lambda expression (Supplier) with a CircuitBreaker and how to retry the call at most 3 times when an exception occurs.
You can configure the wait interval between retries and also configure a custom backoff algorithm.

The example uses Vavr's Try Monad to recover from an exception and invoke another lambda expression as a fallback, when even all retries have failed.

[source,java]
----
// Simulates a Backend Service
public interface BackendService {
    String doSomething();
}

// Create a CircuitBreaker (use default configuration)
CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("backendName");
// Create a Retry with at most 3 retries and a fixed time interval between retries of 500ms
Retry retry = Retry.ofDefaults("backendName");

// Decorate your call to BackendService.doSomething() with a CircuitBreaker
Supplier<String> decoratedSupplier = CircuitBreaker
    .decorateSupplier(circuitBreaker, backendService::doSomething);

// Decorate your call with automatic retry
decoratedSupplier = Retry
    .decorateSupplier(retry, decoratedSupplier);

// Use of Vavr's Try to
// execute the decorated supplier and recover from any exception
String result = Try.ofSupplier(decoratedSupplier)
    .recover(throwable -> "Hello from Recovery").get();

// When you don't want to decorate your lambda expression,
// but just execute it and protect the call by a CircuitBreaker.
String result = circuitBreaker.executeSupplier(backendService::doSomething);
----

==== CircuitBreaker and RxJava2

The following example shows how to decorate an Observable by using the custom RxJava operator.

[source,java]
----
CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("testName");
Observable.fromCallable(backendService::doSomething)
    .compose(CircuitBreakerOperator.of(circuitBreaker))
----

NOTE: Resilience4j also provides RxJava operators for `+RateLimiter+`, `+Bulkhead+`, `+TimeLimiter+` and `+Retry+`.
Find out more in our *https://resilience4j.readme.io/docs/getting-started-2[User Guide]*.

==== CircuitBreaker and Spring Reactor

The following example shows how to decorate a Mono by using the custom Reactor operator.

[source,java]
----
CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("testName");
Mono.fromCallable(backendService::doSomething)
    .transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
----

NOTE: Resilience4j also provides Reactor operators for `+RateLimiter+`, `+Bulkhead+`, `+TimeLimiter+` and `+Retry+`.
Find out more in our *https://resilience4j.readme.io/docs/getting-started-1[User Guide]*.

[[ratelimiter]]
=== RateLimiter

The following example shows how to restrict the calling rate of some method to be not higher than 1 request/second.

[source,java]
----
// Create a custom RateLimiter configuration
RateLimiterConfig config = RateLimiterConfig.custom()
    .timeoutDuration(Duration.ofMillis(100))
    .limitRefreshPeriod(Duration.ofSeconds(1))
    .limitForPeriod(1)
    .build();
// Create a RateLimiter
RateLimiter rateLimiter = RateLimiter.of("backendName", config);

// Decorate your call to BackendService.doSomething()
Supplier<String> restrictedSupplier = RateLimiter
    .decorateSupplier(rateLimiter, backendService::doSomething);

// First call is successful
Try<String> firstTry = Try.ofSupplier(restrictedSupplier);
assertThat(firstTry.isSuccess()).isTrue();

// Second call fails, because the call was not permitted
Try<String> secondTry = Try.of(restrictedSupplier);
assertThat(secondTry.isFailure()).isTrue();
assertThat(secondTry.getCause()).isInstanceOf(RequestNotPermitted.class);
----

[[bulkhead]]
=== Bulkhead
There are two isolation strategies and bulkhead implementations.

==== SemaphoreBulkhead
The following example shows how to decorate a lambda expression with a Bulkhead.
A Bulkhead can be used to limit the amount of parallel executions.
This bulkhead abstraction should work well across a variety of threading and io models.
It is based on a semaphore, and unlike Hystrix, does not provide "shadow" thread pool option.

[source,java]
----
// Create a custom Bulkhead configuration
BulkheadConfig config = BulkheadConfig.custom()
    .maxConcurrentCalls(150)
    .maxWaitDuration(100)
    .build();

Bulkhead bulkhead = Bulkhead.of("backendName", config);

Supplier<String> supplier = Bulkhead
    .decorateSupplier(bulkhead, backendService::doSomething);
----

[[threadpoolbulkhead]]
==== ThreadPoolBulkhead
The following example shows how to use a lambda expression with a ThreadPoolBulkhead which uses a bounded queue and a fixed thread pool.

[source,java]
----
// Create a custom ThreadPoolBulkhead configuration
ThreadPoolBulkheadConfig config = ThreadPoolBulkheadConfig.custom()
    .maxThreadPoolSize(10)
    .coreThreadPoolSize(2)
    .queueCapacity(20)
    .build();

ThreadPoolBulkhead bulkhead = ThreadPoolBulkhead.of("backendName", config);

// Decorate or execute immediately a lambda expression with a ThreadPoolBulkhead.
Supplier<CompletionStage<String>> supplier = ThreadPoolBulkhead
    .decorateSupplier(bulkhead, backendService::doSomething);

CompletionStage<String> execution = bulkhead
    .executeSupplier(backendService::doSomething);
----

[[events]]
== Consume emitted events

`+CircuitBreaker+`, `+RateLimiter+`, `+Cache+`, `+Bulkhead+`, `+TimeLimiter+` and `+Retry+` components emit a stream of events.
It can be consumed for logging, assertions and any other purpose.

=== Examples

A `+CircuitBreakerEvent+` can be a state transition, a circuit breaker reset, a successful call, a recorded error or an ignored error.
All events contains additional information like event creation time and processing duration of the call.
If you want to consume events, you have to register an event consumer.

[source,java]
----
circuitBreaker.getEventPublisher()
    .onSuccess(event -> logger.info(...))
    .onError(event -> logger.info(...))
    .onIgnoredError(event -> logger.info(...))
    .onReset(event -> logger.info(...))
    .onStateTransition(event -> logger.info(...));
// Or if you want to register a consumer listening to all events, you can do:
circuitBreaker.getEventPublisher()
    .onEvent(event -> logger.info(...));
----

You can use RxJava or Spring Reactor Adapters to convert the `+EventPublisher+` into a Reactive Stream.
The advantage of a Reactive Stream is that you can use RxJava's `+observeOn+` operator to specify a different Scheduler that the CircuitBreaker will use to send notifications to its observers/consumers.

[source,java]
----
RxJava2Adapter.toFlowable(circuitBreaker.getEventPublisher())
    .filter(event -> event.getEventType() == Type.ERROR)
    .cast(CircuitBreakerOnErrorEvent.class)
    .subscribe(event -> logger.info(...))
----

NOTE: You can also consume events from other components.
Find out more in our *https://resilience4j.readme.io/[User Guide]*.

== Talks

[cols="4*"]
|===

|0:34
|https://www.youtube.com/watch?v=kR2sm1zelI4[Battle of the Circuit Breakers: Resilience4J vs Istio]
|Nicolas Frankel
|GOTO Berlin

|0:33
|https://www.youtube.com/watch?v=AwcjOhD91Q0[Battle of the Circuit Breakers: Istio vs. Hystrix/Resilience4J]
|Nicolas Frankel
|JFuture

|0:42
|https://www.youtube.com/watch?v=KosSsZEqS-k&t=157[Resilience patterns in the post-Hystrix world]
|Tomasz Skowroński
|Cloud Native Warsaw

|0:52
|https://www.youtube.com/watch?v=NHVxrLb3jFI[Building Robust and Resilient Apps Using Spring Boot and Resilience4j]
|David Caron
|SpringOne

|0:22
|https://www.youtube.com/watch?v=gvDvOWtPLVY&t=140[Hystrix is dead, now what?]
|Tomasz Skowroński
|DevoxxPL

|===

== Companies that use Resilience4j

* *Deutsche Telekom* (In an application with over 400 million requests per day)
* *AOL* (In an application with low latency requirements)
* *Netpulse* (In a system with 40+ integrations)
* *wescale.de* (In a B2B integration platform)
* *Topia* (In an HR application built with microservices architecture)
* *Auto Trader Group plc* (The largest Britain digital automotive marketplace)
* *PlayStation Network* (A platform backend)
* *TUI InfoTec GmbH* (Backend applications inside of reservation booking workflow streams for accommodations)

== License

Copyright 2020 Robert Winkler, Bohdan Storozhuk, Mahmoud Romeh, Dan Maas and others

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and limitations under the License.

READ resilience4j-retry/src/main/java/io/github/resilience4j/retry/Retry.java
/*
 *
 *  Copyright 2016 Robert Winkler
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *
 */
package io.github.resilience4j.retry;

import io.github.resilience4j.core.EventConsumer;
import io.github.resilience4j.core.functions.CheckedConsumer;
import io.github.resilience4j.core.functions.CheckedFunction;
import io.github.resilience4j.core.functions.CheckedRunnable;
import io.github.resilience4j.core.functions.CheckedSupplier;
import io.github.resilience4j.retry.event.*;
import io.github.resilience4j.retry.internal.RetryImpl;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * A Retry instance is thread-safe can be used to decorate multiple requests. A Retry.
 */
public interface Retry {

    /**
     * Creates a Retry with a custom Retry configuration.
     *
     * @param name        the ID of the Retry
     * @param retryConfig a custom Retry configuration
     * @return a Retry with a custom Retry configuration.
     */
    static Retry of(String name, RetryConfig retryConfig) {
        return of(name, retryConfig, Collections.emptyMap());
    }

    /**
     * Creates a Retry with a custom Retry configuration.
     *
     * @param name        the ID of the Retry
     * @param retryConfig a custom Retry configuration
     * @param tags        tags to assign to the Retry
     * @return a Retry with a custom Retry configuration.
     */
    static Retry of(String name, RetryConfig retryConfig, Map<String, String> tags) {
        return new RetryImpl(name, retryConfig, tags);
    }

    /**
     * Creates a Retry with a custom Retry configuration.
     *
     * @param name                the ID of the Retry
     * @param retryConfigSupplier a supplier of a custom Retry configuration
     * @return a Retry with a custom Retry configuration.
     */
    static Retry of(String name, Supplier<RetryConfig> retryConfigSupplier) {
        return of(name, retryConfigSupplier.get(), Collections.emptyMap());
    }

    /**
     * Creates a Retry with a custom Retry configuration.
     *
     * @param name                the ID of the Retry
     * @param retryConfigSupplier a supplier of a custom Retry configuration
     * @param tags                tags to assign to the Retry
     * @return a Retry with a custom Retry configuration.
     */
    static Retry of(String name, Supplier<RetryConfig> retryConfigSupplier,
                    Map<String, String> tags) {
        return new RetryImpl(name, retryConfigSupplier.get(), tags);
    }

    /**
     * Creates a Retry with default configuration.
     *
     * @param name the ID of the Retry
     * @return a Retry with default configuration
     */
    static Retry ofDefaults(String name) {
        return of(name, RetryConfig.ofDefaults(), Collections.emptyMap());
    }

    /**
     * Decorates CompletionStageSupplier with Retry
     *
     * @param retry     the retry context
     * @param scheduler execution service to use to schedule retries
     * @param supplier  completion stage supplier
     * @param <T>       type of completion stage result
     * @return decorated supplier
     */
    static <T> Supplier<CompletionStage<T>> decorateCompletionStage(
        Retry retry,
        ScheduledExecutorService scheduler,
        Supplier<CompletionStage<T>> supplier
    ) {
        return () -> {

            final CompletableFuture<T> promise = new CompletableFuture<>();
            final Runnable block = new AsyncRetryBlock<>(scheduler, retry.asyncContext(), supplier,
                promise);
            block.run();

            return promise;
        };
    }

    /**
     * Decorates CompletionStageSupplier using current instance as context.
     *
     * @param scheduler execution service to use to schedule retries
     * @param supplier  completion stage supplier
     * @param <T>       type of completion stage result
     * @return decorated supplier
     */
    default <T> Supplier<CompletionStage<T>> decorateCompletionStage(
            ScheduledExecutorService scheduler,
            Supplier<CompletionStage<T>> supplier
    ) {
        return decorateCompletionStage(this, scheduler, supplier);
    }

    /**
     * Creates a retryable supplier.
     *
     * @param retry    the retry context
     * @param supplier the original function
     * @param <T>      the type of results supplied by this supplier
     * @return a retryable function
     */
    static <T> CheckedSupplier<T> decorateCheckedSupplier(Retry retry,
                                                          CheckedSupplier<T> supplier) {
        return () -> {
            Retry.Context<T> context = retry.context();
            do {
                try {
                    T result = supplier.get();
                    final boolean validationOfResult = context.onResult(result);
                    if (!validationOfResult) {
                        context.onComplete();
                        return result;
                    }
                } catch (Exception exception) {
                    context.onError(exception);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable supplier using current instance as context.
     *
     * @param supplier the original function
     * @param <T>      the type of results supplied by this supplier
     * @return a retryable function
     */
    default <T> CheckedSupplier<T> decorateCheckedSupplier(CheckedSupplier<T> supplier) {
        return decorateCheckedSupplier(this, supplier);
    }

    /**
     * Creates a retryable runnable.
     *
     * @param retry    the retry context
     * @param runnable the original runnable
     * @return a retryable runnable
     */
    static CheckedRunnable decorateCheckedRunnable(Retry retry, CheckedRunnable runnable) {
        return () -> {
            Retry.Context context = retry.context();
            do {
                try {
                    runnable.run();
                    final boolean validationOfResult = context.onResult(null);
                    if (!validationOfResult) {
                        context.onComplete();
                        break;
                    }
                } catch (Exception exception) {
                    context.onError(exception);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable runnable using current instance as context.
     *
     * @param runnable the original runnable
     * @return a retryable runnable
     */
    default CheckedRunnable decorateCheckedRunnable(CheckedRunnable runnable) {
        return decorateCheckedRunnable(this, runnable);
    }

    /**
     * Creates a retryable function.
     *
     * @param retry    the retry context
     * @param function the original function
     * @param <T>      the type of the input to the function
     * @param <R>      the result type of the function
     * @return a retryable function
     */
    static <T, R> CheckedFunction<T, R> decorateCheckedFunction(Retry retry,
                                                                CheckedFunction<T, R> function) {
        return (T t) -> {
            Retry.Context<R> context = retry.context();
            do {
                try {
                    R result = function.apply(t);
                    final boolean validationOfResult = context.onResult(result);
                    if (!validationOfResult) {
                        context.onComplete();
                        return result;
                    }
                } catch (Exception exception) {
                    context.onError(exception);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable function using current instance as context.
     *
     * @param function the original function
     * @param <T>      the type of the input to the function
     * @param <R>      the result type of the function
     * @return a retryable function
     */
    default <T, R> CheckedFunction<T, R> decorateCheckedFunction(CheckedFunction<T, R> function) {
        return decorateCheckedFunction(this, function);
    }

    /**
     * Creates a retryable consumer.
     *
     * @param retry    the retry context
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a retryable consumer
     */
    static <T> CheckedConsumer<T> decorateCheckedConsumer(Retry retry, CheckedConsumer<T> consumer) {
        return (T t) -> {
            Retry.Context context = retry.context();
            do {
                try {
                    consumer.accept(t);
                    final boolean validationOfResult = context.onResult(null);
                    if (!validationOfResult) {
                        context.onComplete();
                        break;
                    }
                } catch (Exception exception) {
                    context.onError(exception);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable consumer using current instance as context.
     *
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a retryable consumer
     */
    default <T> CheckedConsumer<T> decorateCheckedConsumer(CheckedConsumer<T> consumer) {
        return decorateCheckedConsumer(this, consumer);
    }

    /**
     * Creates a retryable supplier.
     *
     * @param retry    the retry context
     * @param supplier the original function
     * @param <T>      the type of results supplied by this supplier
     * @return a retryable function
     */
    static <T> Supplier<T> decorateSupplier(Retry retry, Supplier<T> supplier) {
        return () -> {
            Retry.Context<T> context = retry.context();
            do {
                try {
                    T result = supplier.get();
                    final boolean validationOfResult = context.onResult(result);
                    if (!validationOfResult) {
                        context.onComplete();
                        return result;
                    }
                } catch (RuntimeException runtimeException) {
                    context.onRuntimeError(runtimeException);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable supplier using current instance as context.
     *
     * @param supplier the original function
     * @param <T>      the type of results supplied by this supplier
     * @return a retryable function
     */
    default <T> Supplier<T> decorateSupplier(Supplier<T> supplier) {
        return decorateSupplier(this, supplier);
    }

    /**
     * Creates a retryable callable.
     *
     * @param retry    the retry context
     * @param supplier the original function
     * @param <T>      the type of results supplied by this supplier
     * @return a retryable function
     */
    static <T> Callable<T> decorateCallable(Retry retry, Callable<T> supplier) {
        return () -> {
            Retry.Context<T> context = retry.context();
            do {
                try {
                    T result = supplier.call();
                    final boolean validationOfResult = context.onResult(result);
                    if (!validationOfResult) {
                        context.onComplete();
                        return result;
                    }
                } catch (Exception exception) {
                    context.onError(exception);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable callable using current instance as context.
     *
     * @param supplier the original function
     * @param <T>      the type of results supplied by this supplier
     * @return a retryable function
     */
    default <T> Callable<T> decorateCallable(Callable<T> supplier) {
        return decorateCallable(this, supplier);
    }


    /**
     * Creates a retryable runnable.
     *
     * @param retry    the retry context
     * @param runnable the original runnable
     * @return a retryable runnable
     */
    static Runnable decorateRunnable(Retry retry, Runnable runnable) {
        return () -> {
            Retry.Context context = retry.context();
            do {
                try {
                    runnable.run();
                    final boolean validationOfResult = context.onResult(null);
                    if (!validationOfResult) {
                        context.onComplete();
                        break;
                    }
                } catch (RuntimeException runtimeException) {
                    context.onRuntimeError(runtimeException);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable runnable using current instance as context.
     *
     * @param runnable the original runnable
     * @return a retryable runnable
     */
    default Runnable decorateRunnable(Runnable runnable) {
        return decorateRunnable(this, runnable);
    }

    /**
     * Creates a retryable function.
     *
     * @param retry    the retry context
     * @param function the original function
     * @param <T>      the type of the input to the function
     * @param <R>      the result type of the function
     * @return a retryable function
     */
    static <T, R> Function<T, R> decorateFunction(Retry retry, Function<T, R> function) {
        return (T t) -> {
            Retry.Context<R> context = retry.context();
            do {
                try {
                    R result = function.apply(t);
                    final boolean validationOfResult = context.onResult(result);
                    if (!validationOfResult) {
                        context.onComplete();
                        return result;
                    }
                } catch (RuntimeException runtimeException) {
                    context.onRuntimeError(runtimeException);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable consumer.
     *
     * @param retry    the retry context
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a retryable consumer
     */
    static <T> Consumer<T> decorateConsumer(Retry retry, Consumer<T> consumer) {
        return (T t) -> {
            Retry.Context context = retry.context();
            do {
                try {
                    consumer.accept(t);
                    final boolean validationOfResult = context.onResult(null);
                    if (!validationOfResult) {
                        context.onComplete();
                        break;
                    }
                } catch (RuntimeException runtimeException) {
                    context.onRuntimeError(runtimeException);
                }
            } while (true);
        };
    }

    /**
     * Creates a retryable consumer using current instance as context.
     *
     * @param consumer the original consumer
     * @return a retryable consumer
     */
    default <T> Consumer<T> decorateConsumer(Consumer<T> consumer) {
        return decorateConsumer(this, consumer);
    }

    /**
     * Returns the ID of this Retry.
     *
     * @return the ID of this Retry
     */
    String getName();

    /**
     * Creates a retry Context.
     *
     * @return the retry Context
     */
    <T> Retry.Context<T> context();

    /**
     * Creates a async retry Context.
     *
     * @return the async retry Context
     */
    <T> Retry.AsyncContext<T> asyncContext();

    /**
     * Returns the RetryConfig of this Retry.
     *
     * @return the RetryConfig of this Retry
     */
    RetryConfig getRetryConfig();

    /**
     * Returns an unmodifiable map with tags assigned to this Retry.
     *
     * @return the tags assigned to this Retry in an unmodifiable map
     */
    Map<String, String> getTags();

    /**
     * Returns an EventPublisher can be used to register event consumers.
     *
     * @return an EventPublisher
     */
    EventPublisher getEventPublisher();

    /**
     * Decorates and executes the decorated Supplier.
     *
     * @param checkedSupplier the original Supplier
     * @param <T>             the type of results supplied by this supplier
     * @return the result of the decorated Supplier.
     * @throws Throwable if something goes wrong applying this function to the given arguments
     */
    default <T> T executeCheckedSupplier(CheckedSupplier<T> checkedSupplier) throws Throwable {
        return decorateCheckedSupplier(this, checkedSupplier).get();
    }

    /**
     * Decorates and executes the decorated Supplier.
     *
     * @param supplier the original Supplier
     * @param <T>      the type of results supplied by this supplier
     * @return the result of the decorated Supplier.
     */
    default <T> T executeSupplier(Supplier<T> supplier) {
        return decorateSupplier(this, supplier).get();
    }

    /**
     * Decorates and executes the decorated Callable.
     *
     * @param callable the original Callable
     * @param <T>      the result type of callable
     * @return the result of the decorated Callable.
     * @throws Exception if unable to compute a result
     */
    default <T> T executeCallable(Callable<T> callable) throws Exception {
        return decorateCallable(this, callable).call();
    }

    /**
     * Decorates and executes the decorated Runnable.
     *
     * @param runnable the original Runnable
     */
    default void executeRunnable(Runnable runnable) {
        decorateRunnable(this, runnable).run();
    }

    /**
     * Decorates and executes the decorated CompletionStage.
     *
     * @param scheduler execution service to use to schedule retries
     * @param supplier  the original CompletionStage
     * @param <T>       the type of results supplied by this supplier
     * @return the decorated CompletionStage.
     */
    default <T> CompletionStage<T> executeCompletionStage(ScheduledExecutorService scheduler,
                                                          Supplier<CompletionStage<T>> supplier) {
        return decorateCompletionStage(this, scheduler, supplier).get();
    }

    /**
     * Get the Metrics of this Retry instance.
     *
     * @return the Metrics of this Retry instance
     */
    Metrics getMetrics();

    interface Metrics {

        /**
         * Returns the number of successful calls without a retry attempt.
         *
         * @return the number of successful calls without a retry attempt
         */
        long getNumberOfSuccessfulCallsWithoutRetryAttempt();

        /**
         * Returns the number of failed calls without a retry attempt.
         *
         * @return the number of failed calls without a retry attempt
         */
        long getNumberOfFailedCallsWithoutRetryAttempt();

        /**
         * Returns the number of successful calls after a retry attempt.
         *
         * @return the number of successful calls after a retry attempt
         */
        long getNumberOfSuccessfulCallsWithRetryAttempt();

        /**
         * Returns the number of failed calls after all retry attempts.
         *
         * @return the number of failed calls after all retry attempts
         */
        long getNumberOfFailedCallsWithRetryAttempt();


        /**
         * Returns the number of total calls after all retry attempts.
         *
         * @return the number of total calls after all retry attempts
         */
        long getNumberOfTotalCalls();
    }

    interface AsyncContext<T> {

        /**
         * Records a successful call or retryable call with the needed generated retry events. When
         * there is a successful retry before reaching the max retries limit, it will generate
         * {@link RetryOnSuccessEvent}. When the retry reach the max retries limit, it will generate
         * {@link RetryOnErrorEvent} with last exception or {@link MaxRetriesExceeded} if no other
         * exception is thrown.
         */
        void onComplete();

        /**
         * Records an failed call.
         *
         * @param throwable the exception to handle
         * @return delay in milliseconds until the next try
         */
        long onError(Throwable throwable);

        /**
         * check the result call.
         *
         * @param result the  result to validate
         * @return delay in milliseconds until the next try if the result match the predicate
         */
        long onResult(T result);
    }

    /**
     * the retry context which will be used during the retry iteration to decide what can be done on
     * error , result, on runtime error
     *
     * @param <T> the result type
     */
    interface Context<T> {

        /**
         * Records a successful call or retryable call with the needed generated retry events. When
         * there is a successful retry before reaching the max retries limit, it will generate a
         * {@link RetryOnSuccessEvent}. When the retry reaches the max retries limit, it will generate a
         * {@link RetryOnErrorEvent} with last exception or {@link MaxRetriesExceeded} if no other
         * exceptions is thrown.
         */
        void onComplete();

        /**
         * @param result the returned result from the called logic
         * @return true if we need to retry again or false if no retry anymore
         */
        boolean onResult(T result);

        /**
         * Handles a checked exception
         *
         * @param exception the exception to handle
         * @throws Exception when retry count has exceeded
         */
        void onError(Exception exception) throws Exception;

        /**
         * Handles a runtime exception
         *
         * @param runtimeException the exception to handle
         * @throws RuntimeException when retry count has exceeded
         */
        void onRuntimeError(RuntimeException runtimeException);
    }

    /**
     * An EventPublisher which subscribes to the reactive stream of RetryEvents and can be used to
     * register event consumers.
     * <p>
     * To understand when the handlers are called, see the documentation of the respective events.
     */
    interface EventPublisher extends io.github.resilience4j.core.EventPublisher<RetryEvent> {

        EventPublisher onRetry(EventConsumer<RetryOnRetryEvent> eventConsumer);

        EventPublisher onSuccess(EventConsumer<RetryOnSuccessEvent> eventConsumer);

        EventPublisher onError(EventConsumer<RetryOnErrorEvent> eventConsumer);

        EventPublisher onIgnoredError(EventConsumer<RetryOnIgnoredErrorEvent> eventConsumer);

    }

    class AsyncRetryBlock<T> implements Runnable {

        private final ScheduledExecutorService scheduler;
        private final Retry.AsyncContext<T> retryContext;
        private final Supplier<CompletionStage<T>> supplier;
        private final CompletableFuture<T> promise;

        AsyncRetryBlock(
            ScheduledExecutorService scheduler,
            Retry.AsyncContext<T> retryContext,
            Supplier<CompletionStage<T>> supplier,
            CompletableFuture<T> promise
        ) {
            this.scheduler = scheduler;
            this.retryContext = retryContext;
            this.supplier = supplier;
            this.promise = promise;
        }

        @Override
        public void run() {
            final CompletionStage<T> stage = supplier.get();

            stage.whenComplete((result, throwable) -> {
                try {
                    if (throwable != null) {
                        if (throwable instanceof Exception) {
                            onError((Exception) throwable);
                        } else {
                            promise.completeExceptionally(throwable);
                        }
                    } else {
                        onResult(result);
                    }
                } catch (Throwable unknownTh) {
                    promise.completeExceptionally(unknownTh);
                }
            });
        }

        private void onError(Exception t) {
            final long delay = retryContext.onError(t);

            if (delay < 0) {
                promise.completeExceptionally(t);
            } else {
                scheduler.schedule(this, delay, TimeUnit.MILLISECONDS);
            }
        }

        private void onResult(T result) {
            final long delay = retryContext.onResult(result);

            if (delay < 0) {
                try {
                    retryContext.onComplete();
                    promise.complete(result);
                } catch (Exception e) {
                    promise.completeExceptionally(e);
                }
            } else {
                scheduler.schedule(this, delay, TimeUnit.MILLISECONDS);
            }
        }
    }
}

READ resilience4j-circuitbreaker/src/main/java/io/github/resilience4j/circuitbreaker/CircuitBreaker.java
/*
 *
 *  Copyright 2017: Robert Winkler
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *
 */
package io.github.resilience4j.circuitbreaker;

import io.github.resilience4j.circuitbreaker.event.*;
import io.github.resilience4j.circuitbreaker.internal.CircuitBreakerStateMachine;
import io.github.resilience4j.core.EventConsumer;
import io.github.resilience4j.core.functions.*;

import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * A CircuitBreaker instance is thread-safe can be used to decorate multiple requests.
 * <p>
 * A {@link CircuitBreaker} manages the state of a backend system. The CircuitBreaker is implemented
 * via a finite state machine with five states: CLOSED, OPEN, HALF_OPEN, DISABLED AND FORCED_OPEN.
 * The CircuitBreaker does not know anything about the backend's state by itself, but uses the
 * information provided by the decorators via {@link CircuitBreaker#onSuccess} and {@link
 * CircuitBreaker#onError} events. Before communicating with the backend, the permission to do so
 * must be obtained via the method {@link CircuitBreaker#tryAcquirePermission()}.
 * <p>
 * The state of the CircuitBreaker changes from CLOSED to OPEN when the failure rate is greater than or
 * equal to a (configurable) threshold. Then, all access to the backend is rejected for a (configurable) time
 * duration. No further calls are permitted.
 * <p>
 * After the time duration has elapsed, the CircuitBreaker state changes from OPEN to HALF_OPEN and
 * allows a number of calls to see if the backend is still unavailable or has become available
 * again. If the failure rate is greater than or equal to the configured threshold, the state changes back to OPEN.
 * If the failure rate is below or equal to the threshold, the state changes back to CLOSED.
 */
public interface CircuitBreaker {

    /**
     * Returns a supplier which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param supplier       the original supplier
     * @param <T>            the type of results supplied by this supplier
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    static <T> CheckedSupplier<T> decorateCheckedSupplier(CircuitBreaker circuitBreaker, CheckedSupplier<T> supplier) {
        return () -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                T result = supplier.get();
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), result);
                return result;
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a supplier which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param supplier       the original supplier
     * @param <T>            the type of the returned CompletionStage's result
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    static <T> Supplier<CompletionStage<T>> decorateCompletionStage(
        CircuitBreaker circuitBreaker,
        Supplier<CompletionStage<T>> supplier
    ) {
        return () -> {

            final CompletableFuture<T> promise = new CompletableFuture<>();

            if (!circuitBreaker.tryAcquirePermission()) {
                promise.completeExceptionally(
                    CallNotPermittedException.createCallNotPermittedException(circuitBreaker));

            } else {
                final long start = circuitBreaker.getCurrentTimestamp();
                try {
                    supplier.get().whenComplete((result, throwable) -> {
                        long duration = circuitBreaker.getCurrentTimestamp() - start;
                        if (throwable != null) {
                            if (throwable instanceof Exception) {
                                circuitBreaker
                                    .onError(duration, circuitBreaker.getTimestampUnit(), throwable);
                            }
                            promise.completeExceptionally(throwable);
                        } else {
                            circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), result);
                            promise.complete(result);
                        }
                    });
                } catch (Exception exception) {
                    long duration = circuitBreaker.getCurrentTimestamp() - start;
                    circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                    promise.completeExceptionally(exception);
                }
            }

            return promise;
        };
    }

    /**
     * Returns a runnable which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param runnable       the original runnable
     * @return a runnable which is decorated by a CircuitBreaker.
     */
    static CheckedRunnable decorateCheckedRunnable(CircuitBreaker circuitBreaker, CheckedRunnable runnable) {
        return () -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                runnable.run();
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onSuccess(duration, circuitBreaker.getTimestampUnit());
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a callable which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param callable       the original Callable
     * @param <T>            the result type of callable
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    static <T> Callable<T> decorateCallable(CircuitBreaker circuitBreaker, Callable<T> callable) {
        return () -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                T result = callable.call();
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), result);
                return result;
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a supplier which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param supplier       the original supplier
     * @param <T>            the type of results supplied by this supplier
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    static <T> Supplier<T> decorateSupplier(CircuitBreaker circuitBreaker, Supplier<T> supplier) {
        return () -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                T result = supplier.get();
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), result);
                return result;
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a consumer which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param consumer       the original consumer
     * @param <T>            the type of the input to the consumer
     * @return a consumer which is decorated by a CircuitBreaker.
     */
    static <T> Consumer<T> decorateConsumer(CircuitBreaker circuitBreaker, Consumer<T> consumer) {
        return (t) -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                consumer.accept(t);
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onSuccess(duration, circuitBreaker.getTimestampUnit());
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a consumer which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param consumer       the original consumer
     * @param <T>            the type of the input to the consumer
     * @return a consumer which is decorated by a CircuitBreaker.
     */
    static <T> CheckedConsumer<T> decorateCheckedConsumer(CircuitBreaker circuitBreaker, CheckedConsumer<T> consumer) {
        return (t) -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                consumer.accept(t);
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onSuccess(duration, circuitBreaker.getTimestampUnit());
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a runnable which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param runnable       the original runnable
     * @return a runnable which is decorated by a CircuitBreaker.
     */
    static Runnable decorateRunnable(CircuitBreaker circuitBreaker, Runnable runnable) {
        return () -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                runnable.run();
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onSuccess(duration, circuitBreaker.getTimestampUnit());
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a function which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param function       the original function
     * @param <T>            the type of the input to the function
     * @param <R>            the type of the result of the function
     * @return a function which is decorated by a CircuitBreaker.
     */
    static <T, R> Function<T, R> decorateFunction(CircuitBreaker circuitBreaker,
        Function<T, R> function) {
        return (T t) -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                R returnValue = function.apply(t);
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), returnValue);
                return returnValue;
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Returns a function which is decorated by a CircuitBreaker.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param function       the original function
     * @param <T>            the type of the input to the function
     * @param <R>            the type of the result of the function
     * @return a function which is decorated by a CircuitBreaker.
     */
    static <T, R> CheckedFunction<T, R> decorateCheckedFunction(CircuitBreaker circuitBreaker, CheckedFunction<T, R> function) {
        return (T t) -> {
            circuitBreaker.acquirePermission();
            final long start = circuitBreaker.getCurrentTimestamp();
            try {
                R result = function.apply(t);
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onResult(duration, circuitBreaker.getTimestampUnit(), result);
                return result;
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long duration = circuitBreaker.getCurrentTimestamp() - start;
                circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), exception);
                throw exception;
            }
        };
    }

    /**
     * Creates a CircuitBreaker with a default CircuitBreaker configuration.
     *
     * @param name the name of the CircuitBreaker
     * @return a CircuitBreaker with a default CircuitBreaker configuration.
     */
    static CircuitBreaker ofDefaults(String name) {
        return new CircuitBreakerStateMachine(name);
    }

    /**
     * Creates a CircuitBreaker with a custom CircuitBreaker configuration.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig a custom CircuitBreaker configuration
     * @return a CircuitBreaker with a custom CircuitBreaker configuration.
     */
    static CircuitBreaker of(String name, CircuitBreakerConfig circuitBreakerConfig) {
        return new CircuitBreakerStateMachine(name, circuitBreakerConfig);
    }

    /**
     * Creates a CircuitBreaker with a custom CircuitBreaker configuration.
     * <p>
     * The {@code tags} passed will be appended to the tags already configured for the registry.
     * When tags (keys) of the two collide the tags passed with this method will override the tags
     * of the registry.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig a custom CircuitBreaker configuration
     * @param tags                 tags added to the CircuitBreaker
     * @return a CircuitBreaker with a custom CircuitBreaker configuration.
     */
    static CircuitBreaker of(String name, CircuitBreakerConfig circuitBreakerConfig, Map<String, String> tags) {
        return new CircuitBreakerStateMachine(name, circuitBreakerConfig, tags);
    }

    /**
     * Creates a CircuitBreaker with a custom CircuitBreaker configuration.
     *
     * @param name                         the name of the CircuitBreaker
     * @param circuitBreakerConfigSupplier a supplier of a custom CircuitBreaker configuration
     * @return a CircuitBreaker with a custom CircuitBreaker configuration.
     */
    static CircuitBreaker of(String name,
        Supplier<CircuitBreakerConfig> circuitBreakerConfigSupplier) {
        return new CircuitBreakerStateMachine(name, circuitBreakerConfigSupplier);
    }

    /**
     * Creates a CircuitBreaker with a custom CircuitBreaker configuration.
     * <p>
     * The {@code tags} passed will be appended to the tags already configured for the registry.
     * When tags (keys) of the two collide the tags passed with this method will override the tags
     * of the registry.
     *
     * @param name                         the name of the CircuitBreaker
     * @param circuitBreakerConfigSupplier a supplier of a custom CircuitBreaker configuration
     * @param tags                         tags added to the CircuitBreaker
     * @return a CircuitBreaker with a custom CircuitBreaker configuration.
     */
    static CircuitBreaker of(String name,
        Supplier<CircuitBreakerConfig> circuitBreakerConfigSupplier, Map<String, String> tags) {
        return new CircuitBreakerStateMachine(name, circuitBreakerConfigSupplier, tags);
    }

    /**
     * Returns a supplier of type Future which is decorated by a CircuitBreaker. The elapsed time
     * includes {@link Future#get()} evaluation time even if the underlying call took less time to
     * return. Any delays in evaluating Future by caller will add towards total time.
     *
     * @param circuitBreaker the CircuitBreaker
     * @param supplier       the original supplier
     * @param <T>            the type of the returned Future's result
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    static <T> Supplier<Future<T>> decorateFuture(CircuitBreaker circuitBreaker,
        Supplier<Future<T>> supplier) {
        return () -> {
            if (!circuitBreaker.tryAcquirePermission()) {
                CompletableFuture<T> promise = new CompletableFuture<>();
                promise.completeExceptionally(
                    CallNotPermittedException.createCallNotPermittedException(circuitBreaker));
                return promise;
            } else {
                final long start = circuitBreaker.getCurrentTimestamp();
                try {
                    return new CircuitBreakerFuture<>(circuitBreaker, supplier.get(), start);
                } catch (Exception e) {
                    long duration = circuitBreaker.getCurrentTimestamp() - start;
                    circuitBreaker.onError(duration, circuitBreaker.getTimestampUnit(), e);
                    throw e;
                }
            }
        };
    }

    /**
     * Acquires a permission to execute a call, only if one is available at the time of invocation.
     * If a call is not permitted, the number of not permitted calls is increased.
     * <p>
     * Returns false when the state is OPEN or FORCED_OPEN. Returns true when the state is CLOSED or
     * DISABLED. Returns true when the state is HALF_OPEN and further test calls are allowed.
     * Returns false when the state is HALF_OPEN and the number of test calls has been reached. If
     * the state is HALF_OPEN, the number of allowed test calls is decreased. Important: Make sure
     * to call onSuccess or onError after the call is finished. If the call is cancelled before it
     * is invoked, you have to release the permission again.
     *
     * @return {@code true} if a permission was acquired and {@code false} otherwise
     */
    boolean tryAcquirePermission();

    /**
     * Releases a permission.
     * <p>
     * Should only be used when a permission was acquired but not used. Otherwise use {@link
     * CircuitBreaker#onSuccess(long, TimeUnit)} or
     * {@link CircuitBreaker#onError(long, TimeUnit, Throwable)} to signal a completed or failed call.
     * <p>
     * If the state is HALF_OPEN, the number of allowed test calls is increased by one.
     */
    void releasePermission();

    /**
     * Try to obtain a permission to execute a call. If a call is not permitted, the number of not
     * permitted calls is increased.
     * <p>
     * Throws a CallNotPermittedException when the state is OPEN or FORCED_OPEN. Returns when the
     * state is CLOSED or DISABLED. Returns when the state is HALF_OPEN and further test calls are
     * allowed. Throws a CallNotPermittedException when the state is HALF_OPEN and the number of
     * test calls has been reached. If the state is HALF_OPEN, the number of allowed test calls is
     * decreased. Important: Make sure to call onSuccess or onError after the call is finished. If
     * the call is cancelled before it is invoked, you have to release the permission again.
     *
     * @throws CallNotPermittedException when CircuitBreaker is OPEN or HALF_OPEN and no further
     *                                   test calls are permitted.
     */
    void acquirePermission();

    /**
     * Records a failed call. This method must be invoked when a call failed.
     *
     * @param duration     The elapsed time duration of the call
     * @param durationUnit The duration unit
     * @param throwable    The throwable which must be recorded
     */
    void onError(long duration, TimeUnit durationUnit, Throwable throwable);

    /**
     * Records a successful call. This method must be invoked when a call was
     * successful.
     *
     * @param duration     The elapsed time duration of the call
     * @param durationUnit The duration unit
     */
    void onSuccess(long duration, TimeUnit durationUnit);

    /**
     * This method must be invoked when a call returned a result
     * and the result predicate should decide if the call was successful or not.
     *
     * @param duration     The elapsed time duration of the call
     * @param durationUnit The duration unit
     * @param result       The result of the protected function
     */
    void onResult(long duration, TimeUnit durationUnit, Object result);

    /**
     * Returns the circuit breaker to its original closed state, losing statistics.
     * <p>
     * Should only be used, when you want to fully reset the circuit breaker without
     * creating a new one.
     */
    void reset();

    /**
     * Transitions the state machine to CLOSED state. This call is idempotent and will not have
     * any effect if the state machine is already in CLOSED state.
     * <p>
     * Should only be used, when you want to force a state transition. State transition are normally
     * done internally.
     */
    void transitionToClosedState();

    /**
     * Transitions the state machine to OPEN state. This call is idempotent and will not have
     * any effect if the state machine is already in OPEN state.
     * <p>
     * Should only be used, when you want to force a state transition. State transition are normally
     * done internally.
     */
    void transitionToOpenState();

    /**
     * Same as {@link #transitionToOpenState()} but waits in open state for the given amount of time
     * instead of relaying on configurations to determine it.
     *
     * @param waitDuration how long should we wait in open state
     */
    void transitionToOpenStateFor(Duration waitDuration);

    /**
     * Same as {@link #transitionToOpenState()} but waits in open state until the given in time
     * instead of relaying on configurations to determine it.
     *
     * @param waitUntil how long should we wait in open state
     */
    void transitionToOpenStateUntil(Instant waitUntil);

    /**
     * Transitions the state machine to HALF_OPEN state. This call is idempotent and will not have
     * any effect if the state machine is already in HALF_OPEN state.
     * <p>
     * Should only be used, when you want to force a state transition. State transition are normally
     * done internally.
     */
    void transitionToHalfOpenState();

    /**
     * Transitions the state machine to a DISABLED state, stopping state transition, metrics and
     * event publishing. This call is idempotent and will not have any effect if the
     * state machine is already in DISABLED state.
     * <p>
     * Should only be used, when you want to disable the circuit breaker allowing all calls to pass.
     * To recover from this state you must force a new state transition
     */
    void transitionToDisabledState();

    /**
     * Transitions the state machine to METRICS_ONLY state, stopping all state transitions but
     * continues to capture metrics and publish events. This call is idempotent and will not have
     * any effect if the state machine is already in METRICS_ONLY state.
     * <p>
     * Should only be used when you want to collect metrics while keeping the circuit breaker
     * disabled, allowing all calls to pass.
     * To recover from this state you must force a new state transition.
     */
    void transitionToMetricsOnlyState();

    /**
     * Transitions the state machine to a FORCED_OPEN state,  stopping state transition, metrics and
     * event publishing. This call is idempotent and will not have any effect if the state machine
     * is already in FORCED_OPEN state.
     * <p>
     * Should only be used, when you want to disable the circuit breaker allowing no call to pass.
     * To recover from this state you must force a new state transition
     */
    void transitionToForcedOpenState();

    /**
     * Returns the name of this CircuitBreaker.
     *
     * @return the name of this CircuitBreaker
     */
    String getName();

    /**
     * Returns the state of this CircuitBreaker.
     *
     * @return the state of this CircuitBreaker
     */
    State getState();

    /**
     * Returns the CircuitBreakerConfig of this CircuitBreaker.
     *
     * @return the CircuitBreakerConfig of this CircuitBreaker
     */
    CircuitBreakerConfig getCircuitBreakerConfig();

    /**
     * Returns the Metrics of this CircuitBreaker.
     *
     * @return the Metrics of this CircuitBreaker
     */
    Metrics getMetrics();

    /**
     * Returns an unmodifiable map with tags assigned to this CircuitBreaker.
     *
     * @return the tags assigned to this CircuitBreaker in an unmodifiable map
     */
    Map<String, String> getTags();

    /**
     * Returns an EventPublisher which can be used to register event consumers.
     *
     * @return an EventPublisher
     */
    EventPublisher getEventPublisher();

    /**
     * Returns the current time with respect to the CircuitBreaker currentTimeFunction.
     * Returns System.nanoTime() by default.
     *
     * @return current timestamp
     */
    long getCurrentTimestamp();

    /**
     * Returns the timeUnit of current timestamp.
     * Default is TimeUnit.NANOSECONDS.
     *
     * @return the timeUnit of current timestamp
     */
    TimeUnit getTimestampUnit();

    /**
     * Decorates and executes the decorated Supplier.
     *
     * @param supplier the original Supplier
     * @param <T>      the type of results supplied by this supplier
     * @return the result of the decorated Supplier.
     */
    default <T> T executeSupplier(Supplier<T> supplier) {
        return decorateSupplier(this, supplier).get();
    }

    /**
     * Returns a supplier which is decorated by a CircuitBreaker.
     *
     * @param supplier the original supplier
     * @param <T>      the type of results supplied by this supplier
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    default <T> Supplier<T> decorateSupplier(Supplier<T> supplier) {
        return decorateSupplier(this, supplier);
    }

    /**
     * Decorates and executes the decorated Callable.
     *
     * @param callable the original Callable
     * @param <T>      the result type of callable
     * @return the result of the decorated Callable.
     * @throws Exception if unable to compute a result
     */
    default <T> T executeCallable(Callable<T> callable) throws Exception {
        return decorateCallable(this, callable).call();
    }

    /**
     * Returns a callable which is decorated by a CircuitBreaker.
     *
     * @param callable the original Callable
     * @param <T>      the result type of callable
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    default <T> Callable<T> decorateCallable(Callable<T> callable) {
        return decorateCallable(this, callable);
    }

    /**
     * Decorates and executes the decorated Runnable.
     *
     * @param runnable the original Runnable
     */
    default void executeRunnable(Runnable runnable) {
        decorateRunnable(this, runnable).run();
    }

    /**
     * Returns a runnable which is decorated by a CircuitBreaker.
     *
     * @param runnable the original runnable
     * @return a runnable which is decorated by a CircuitBreaker.
     */
    default Runnable decorateRunnable(Runnable runnable) {
        return decorateRunnable(this, runnable);
    }

    /**
     * Decorates and executes the decorated CompletionStage.
     *
     * @param supplier the original CompletionStage
     * @param <T>      the type of results supplied by this supplier
     * @return the decorated CompletionStage.
     */
    default <T> CompletionStage<T> executeCompletionStage(Supplier<CompletionStage<T>> supplier) {
        return decorateCompletionStage(this, supplier).get();
    }

    /**
     * Returns a supplier which is decorated by a CircuitBreaker.
     *
     * @param supplier the original supplier
     * @param <T>      the type of the returned CompletionStage's result
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    default <T> Supplier<CompletionStage<T>> decorateCompletionStage(
        Supplier<CompletionStage<T>> supplier) {
        return decorateCompletionStage(this, supplier);
    }

    /**
     * Decorates and executes the decorated Supplier.
     *
     * @param checkedSupplier the original Supplier
     * @param <T>             the type of results supplied by this supplier
     * @return the result of the decorated Supplier.
     * @throws Throwable if something goes wrong applying this function to the given arguments
     */
    default <T> T executeCheckedSupplier(CheckedSupplier<T> checkedSupplier) throws Throwable {
        return decorateCheckedSupplier(this, checkedSupplier).get();
    }

    /**
     * Returns a supplier which is decorated by a CircuitBreaker.
     *
     * @param checkedSupplier the original supplier
     * @param <T>             the type of results supplied by this supplier
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    default <T> CheckedSupplier<T> decorateCheckedSupplier(CheckedSupplier<T> checkedSupplier) {
        return decorateCheckedSupplier(this, checkedSupplier);
    }

    /**
     * Returns a runnable which is decorated by a CircuitBreaker.
     *
     * @param runnable the original runnable
     * @return a runnable which is decorated by a CircuitBreaker.
     */
    default CheckedRunnable decorateCheckedRunnable(CheckedRunnable runnable) {
        return decorateCheckedRunnable(this, runnable);
    }

    /**
     * Decorates and executes the decorated Runnable.
     *
     * @param runnable the original runnable
     */
    default void executeCheckedRunnable(CheckedRunnable runnable) throws Throwable {
        decorateCheckedRunnable(this, runnable).run();
    }

    /**
     * Returns a consumer which is decorated by a CircuitBreaker.
     *
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a consumer which is decorated by a CircuitBreaker.
     */
    default <T> Consumer<T> decorateConsumer(Consumer<T> consumer) {
        return decorateConsumer(this, consumer);
    }

    /**
     * Returns a consumer which is decorated by a CircuitBreaker.
     *
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a consumer which is decorated by a CircuitBreaker.
     */
    default <T> CheckedConsumer<T> decorateCheckedConsumer(CheckedConsumer<T> consumer) {
        return decorateCheckedConsumer(this, consumer);
    }

    /**
     * Returns a supplier of type Future which is decorated by a CircuitBreaker. The elapsed time
     * includes {@link Future#get()} evaluation time even if the underlying call took less time to
     * return. Any delays in evaluating Future by caller will add towards total time.
     *
     * @param supplier the original supplier
     * @param <T>      the type of the returned CompletionStage's result
     * @return a supplier which is decorated by a CircuitBreaker.
     */
    default <T> Supplier<Future<T>> decorateFuture(Supplier<Future<T>> supplier) {
        return decorateFuture(this, supplier);
    }

    /**
     * States of the CircuitBreaker state machine.
     */
    enum State {
        /**
         * A DISABLED breaker is not operating (no state transition, no events) and allowing all
         * requests through.
         */
        DISABLED(3, false),
        /**
         * A METRICS_ONLY breaker is collecting metrics, publishing events and allowing all requests
         * through but is not transitioning to other states.
         */
        METRICS_ONLY(5, true),
        /**
         * A CLOSED breaker is operating normally and allowing requests through.
         */
        CLOSED(0, true),
        /**
         * An OPEN breaker has tripped and will not allow requests through.
         */
        OPEN(1, true),
        /**
         * A FORCED_OPEN breaker is not operating (no state transition, no events) and not allowing
         * any requests through.
         */
        FORCED_OPEN(4, false),
        /**
         * A HALF_OPEN breaker has completed its wait interval and will allow requests
         */
        HALF_OPEN(2, true);

        public final boolean allowPublish;
        private final int order;

        /**
         * Order is a FIXED integer, it should be preserved regardless of the ordinal number of the
         * enumeration. While a State.ordinal() does mostly the same, it is prone to changing the
         * order based on how the programmer  sets the enum. If more states are added the "order"
         * should be preserved. For example, if there is a state inserted between CLOSED and
         * HALF_OPEN (say FIXED_OPEN) then the order of HALF_OPEN remains at 2 and the new state
         * takes 3 regardless of its order in the enum.
         *
         * @param order
         * @param allowPublish
         */
        State(int order, boolean allowPublish) {
            this.order = order;
            this.allowPublish = allowPublish;
        }

        public int getOrder() {
            return order;
        }
    }

    /**
     * State transitions of the CircuitBreaker state machine.
     */
    enum StateTransition {
        CLOSED_TO_CLOSED(State.CLOSED, State.CLOSED),
        CLOSED_TO_OPEN(State.CLOSED, State.OPEN),
        CLOSED_TO_DISABLED(State.CLOSED, State.DISABLED),
        CLOSED_TO_METRICS_ONLY(State.CLOSED, State.METRICS_ONLY),
        CLOSED_TO_FORCED_OPEN(State.CLOSED, State.FORCED_OPEN),
        HALF_OPEN_TO_HALF_OPEN(State.HALF_OPEN, State.HALF_OPEN),
        HALF_OPEN_TO_CLOSED(State.HALF_OPEN, State.CLOSED),
        HALF_OPEN_TO_OPEN(State.HALF_OPEN, State.OPEN),
        HALF_OPEN_TO_DISABLED(State.HALF_OPEN, State.DISABLED),
        HALF_OPEN_TO_METRICS_ONLY(State.HALF_OPEN, State.METRICS_ONLY),
        HALF_OPEN_TO_FORCED_OPEN(State.HALF_OPEN, State.FORCED_OPEN),
        OPEN_TO_OPEN(State.OPEN, State.OPEN),
        OPEN_TO_CLOSED(State.OPEN, State.CLOSED),
        OPEN_TO_HALF_OPEN(State.OPEN, State.HALF_OPEN),
        OPEN_TO_DISABLED(State.OPEN, State.DISABLED),
        OPEN_TO_METRICS_ONLY(State.OPEN, State.METRICS_ONLY),
        OPEN_TO_FORCED_OPEN(State.OPEN, State.FORCED_OPEN),
        FORCED_OPEN_TO_FORCED_OPEN(State.FORCED_OPEN, State.FORCED_OPEN),
        FORCED_OPEN_TO_CLOSED(State.FORCED_OPEN, State.CLOSED),
        FORCED_OPEN_TO_OPEN(State.FORCED_OPEN, State.OPEN),
        FORCED_OPEN_TO_DISABLED(State.FORCED_OPEN, State.DISABLED),
        FORCED_OPEN_TO_METRICS_ONLY(State.FORCED_OPEN, State.METRICS_ONLY),
        FORCED_OPEN_TO_HALF_OPEN(State.FORCED_OPEN, State.HALF_OPEN),
        DISABLED_TO_DISABLED(State.DISABLED, State.DISABLED),
        DISABLED_TO_CLOSED(State.DISABLED, State.CLOSED),
        DISABLED_TO_OPEN(State.DISABLED, State.OPEN),
        DISABLED_TO_FORCED_OPEN(State.DISABLED, State.FORCED_OPEN),
        DISABLED_TO_HALF_OPEN(State.DISABLED, State.HALF_OPEN),
        DISABLED_TO_METRICS_ONLY(State.DISABLED, State.METRICS_ONLY),
        METRICS_ONLY_TO_METRICS_ONLY(State.METRICS_ONLY, State.METRICS_ONLY),
        METRICS_ONLY_TO_CLOSED(State.METRICS_ONLY, State.CLOSED),
        METRICS_ONLY_TO_FORCED_OPEN(State.METRICS_ONLY, State.FORCED_OPEN),
        METRICS_ONLY_TO_DISABLED(State.METRICS_ONLY, State.DISABLED);

        private static final Map<Map.Entry<State, State>, StateTransition> STATE_TRANSITION_MAP = Arrays
            .stream(StateTransition.values())
            .collect(Collectors.toMap(v -> Map.entry(v.fromState, v.toState), Function.identity()));
        private final State fromState;
        private final State toState;

        StateTransition(State fromState, State toState) {
            this.fromState = fromState;
            this.toState = toState;
        }

        public static StateTransition transitionBetween(String name, State fromState,
            State toState) {
            final StateTransition stateTransition = STATE_TRANSITION_MAP
                .get(Map.entry(fromState, toState));
            if (stateTransition == null) {
                throw new IllegalStateTransitionException(name, fromState, toState);
            }
            return stateTransition;
        }

        public State getFromState() {
            return fromState;
        }

        public State getToState() {
            return toState;
        }

        public static boolean isInternalTransition(final StateTransition transition) {
            return transition.getToState() == transition.getFromState();
        }

        @Override
        public String toString() {
            return String.format("State transition from %s to %s", fromState, toState);
        }
    }

    /**
     * An EventPublisher can be used to register event consumers.
     */
    interface EventPublisher extends
        io.github.resilience4j.core.EventPublisher<CircuitBreakerEvent> {

        EventPublisher onSuccess(EventConsumer<CircuitBreakerOnSuccessEvent> eventConsumer);

        EventPublisher onError(EventConsumer<CircuitBreakerOnErrorEvent> eventConsumer);

        EventPublisher onStateTransition(
            EventConsumer<CircuitBreakerOnStateTransitionEvent> eventConsumer);

        EventPublisher onReset(EventConsumer<CircuitBreakerOnResetEvent> eventConsumer);

        EventPublisher onIgnoredError(
            EventConsumer<CircuitBreakerOnIgnoredErrorEvent> eventConsumer);

        EventPublisher onCallNotPermitted(
            EventConsumer<CircuitBreakerOnCallNotPermittedEvent> eventConsumer);

        EventPublisher onFailureRateExceeded(
            EventConsumer<CircuitBreakerOnFailureRateExceededEvent> eventConsumer);

        EventPublisher onSlowCallRateExceeded(
            EventConsumer<CircuitBreakerOnSlowCallRateExceededEvent> eventConsumer);
    }

    interface Metrics {

        /**
         * Returns the current failure rate in percentage. If the number of measured calls is below
         * the minimum number of measured calls, it returns -1.
         *
         * @return the failure rate in percentage
         */
        float getFailureRate();

        /**
         * Returns the current percentage of calls which were slower than a certain threshold. If
         * the number of measured calls is below the minimum number of measured calls, it returns
         * -1.
         *
         * @return the failure rate in percentage
         */
        float getSlowCallRate();

        /**
         * Returns the current total number of calls which were slower than a certain threshold.
         *
         * @return the current total number of calls which were slower than a certain threshold
         */
        int getNumberOfSlowCalls();

        /**
         * Returns the current number of successful calls which were slower than a certain
         * threshold.
         *
         * @return the current number of successful calls which were slower than a certain threshold
         */
        int getNumberOfSlowSuccessfulCalls();

        /**
         * Returns the current number of failed calls which were slower than a certain threshold.
         *
         * @return the current number of failed calls which were slower than a certain threshold
         */
        int getNumberOfSlowFailedCalls();

        /**
         * Returns the current total number of buffered calls in the ring buffer.
         *
         * @return he current total number of buffered calls in the ring buffer
         */
        int getNumberOfBufferedCalls();

        /**
         * Returns the current number of failed buffered calls in the ring buffer.
         *
         * @return the current number of failed buffered calls in the ring buffer
         */
        int getNumberOfFailedCalls();

        /**
         * Returns the current number of not permitted calls, when the state is OPEN.
         * <p>
         * The number of denied calls is always 0, when the CircuitBreaker state is CLOSED or
         * HALF_OPEN. The number of denied calls is only increased when the CircuitBreaker state is
         * OPEN.
         *
         * @return the current number of not permitted calls
         */
        long getNumberOfNotPermittedCalls();

        /**
         * Returns the current number of successful buffered calls in the ring buffer.
         *
         * @return the current number of successful buffered calls in the ring buffer
         */
        int getNumberOfSuccessfulCalls();
    }

    /**
     * This class decorates future to add CircuitBreaking functionality around invocation.
     *
     * @param <T> of return type
     */
    final class CircuitBreakerFuture<T> implements Future<T> {

        private final Future<T> future;
        private final OnceConsumer<CircuitBreaker> onceToCircuitbreaker;
        private final long start;

        CircuitBreakerFuture(CircuitBreaker circuitBreaker, Future<T> future) {
            this(circuitBreaker, future, circuitBreaker.getCurrentTimestamp());
        }

        CircuitBreakerFuture(CircuitBreaker circuitBreaker, Future<T> future, long start) {
            Objects.requireNonNull(future, "Non null Future is required to decorate");
            this.onceToCircuitbreaker = OnceConsumer.of(circuitBreaker);
            this.future = future;
            this.start = start;
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return future.cancel(mayInterruptIfRunning);
        }

        @Override
        public boolean isCancelled() {
            return future.isCancelled();
        }

        @Override
        public boolean isDone() {
            return future.isDone();
        }

        @Override
        public T get() throws InterruptedException, ExecutionException {
            try {
                T v = future.get();
                onceToCircuitbreaker
                    .applyOnce(cb ->
                        cb.onResult(cb.getCurrentTimestamp() - start, cb.getTimestampUnit(), v)
                    );
                return v;
            } catch (CancellationException | InterruptedException e) {
                onceToCircuitbreaker.applyOnce(cb -> cb.releasePermission());
                throw e;
            } catch (Exception e) {
                onceToCircuitbreaker.applyOnce(
                    cb -> cb.onError(cb.getCurrentTimestamp() - start, cb.getTimestampUnit(), e));
                throw e;
            }
        }

        @Override
        public T get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            try {
                T v = future.get(timeout, unit);
                onceToCircuitbreaker
                    .applyOnce(cb ->
                        cb.onResult(cb.getCurrentTimestamp()  - start, cb.getTimestampUnit(), v)
                    );
                return v;
            } catch (CancellationException | InterruptedException e) {
                onceToCircuitbreaker.applyOnce(CircuitBreaker::releasePermission);
                throw e;
            } catch (Exception e) {
                onceToCircuitbreaker.applyOnce(
                    cb -> cb.onError(cb.getCurrentTimestamp() - start, cb.getTimestampUnit(), e));
                throw e;
            }
        }
    }
}

READ resilience4j-circuitbreaker/src/main/java/io/github/resilience4j/circuitbreaker/internal/CircuitBreakerStateMachine.java
/*
 *
 *  Copyright 2019 Robert Winkler
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *
 */
package io.github.resilience4j.circuitbreaker.internal;

import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.ResultRecordedAsFailureException;
import io.github.resilience4j.circuitbreaker.event.*;
import io.github.resilience4j.core.EventConsumer;
import io.github.resilience4j.core.EventProcessor;
import io.github.resilience4j.core.functions.Either;
import io.github.resilience4j.core.lang.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;

import static io.github.resilience4j.circuitbreaker.CircuitBreaker.State.*;
import static io.github.resilience4j.circuitbreaker.internal.CircuitBreakerMetrics.Result;
import static io.github.resilience4j.circuitbreaker.internal.CircuitBreakerMetrics.Result.BELOW_THRESHOLDS;
import static java.time.temporal.ChronoUnit.MILLIS;
import static java.util.Collections.emptyMap;

/**
 * A CircuitBreaker finite state machine.
 */
public final class CircuitBreakerStateMachine implements CircuitBreaker {

    private static final Logger LOG = LoggerFactory.getLogger(CircuitBreakerStateMachine.class);

    private final String name;
    private final AtomicReference<CircuitBreakerState> stateReference;
    private final CircuitBreakerConfig circuitBreakerConfig;
    private final Map<String, String> tags;
    private final CircuitBreakerEventProcessor eventProcessor;
    private final Clock clock;
    private final SchedulerFactory schedulerFactory;
    private final Function<Clock, Long> currentTimestampFunction;
    private final TimeUnit timestampUnit;
    private final ReentrantLock lock = new ReentrantLock();

    /**
     * Creates a circuitBreaker.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig The CircuitBreaker configuration.
     * @param schedulerFactory     A SchedulerFactory which can be mocked in tests.
     */
    private CircuitBreakerStateMachine(String name, CircuitBreakerConfig circuitBreakerConfig,
        SchedulerFactory schedulerFactory, Map<String, String> tags) {
        this.name = name;
        this.circuitBreakerConfig = Objects
            .requireNonNull(circuitBreakerConfig, "Config must not be null");
        this.eventProcessor = new CircuitBreakerEventProcessor();
        this.clock = circuitBreakerConfig.getClock();
        this.stateReference = new AtomicReference<>(new ClosedState());
        this.schedulerFactory = schedulerFactory;
        this.tags = Objects.requireNonNull(tags, "Tags must not be null");
        this.currentTimestampFunction = circuitBreakerConfig.getCurrentTimestampFunction();
        this.timestampUnit = circuitBreakerConfig.getTimestampUnit();
    }

    /**
     * Creates a circuitBreaker.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig The CircuitBreaker configuration.
     * @param schedulerFactory     A SchedulerFactory which can be mocked in tests.
     */
    public CircuitBreakerStateMachine(String name, CircuitBreakerConfig circuitBreakerConfig,
        SchedulerFactory schedulerFactory) {
        this(name, circuitBreakerConfig, schedulerFactory, emptyMap());
    }

    /**
     * Creates a circuitBreaker.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig The CircuitBreaker configuration.
     */
    public CircuitBreakerStateMachine(String name, CircuitBreakerConfig circuitBreakerConfig) {
        this(name, circuitBreakerConfig, SchedulerFactory.getInstance(), emptyMap());
    }

    /**
     * Creates a circuitBreaker.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig The CircuitBreaker configuration.
     */
    public CircuitBreakerStateMachine(String name, CircuitBreakerConfig circuitBreakerConfig,
        Map<String, String> tags) {
        this(name, circuitBreakerConfig, SchedulerFactory.getInstance(), tags);
    }

    /**
     * Creates a circuitBreaker with default config.
     *
     * @param name the name of the CircuitBreaker
     */
    public CircuitBreakerStateMachine(String name) {
        this(name, CircuitBreakerConfig.ofDefaults());
    }

    /**
     * Creates a circuitBreaker.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig The CircuitBreaker configuration supplier.
     */
    public CircuitBreakerStateMachine(String name,
        Supplier<CircuitBreakerConfig> circuitBreakerConfig) {
        this(name, circuitBreakerConfig.get());
    }

    /**
     * Creates a circuitBreaker.
     *
     * @param name                 the name of the CircuitBreaker
     * @param circuitBreakerConfig The CircuitBreaker configuration supplier.
     */
    public CircuitBreakerStateMachine(String name,
        Supplier<CircuitBreakerConfig> circuitBreakerConfig,
        Map<String, String> tags) {
        this(name, circuitBreakerConfig.get(), tags);
    }

    @Override
    public long getCurrentTimestamp() {
        return this.currentTimestampFunction.apply(clock);
    }

    @Override
    public TimeUnit getTimestampUnit() {
        return timestampUnit;
    }

    @Override
    public boolean tryAcquirePermission() {
        boolean callPermitted = stateReference.get().tryAcquirePermission();
        if (!callPermitted) {
            publishCallNotPermittedEvent();
        }
        return callPermitted;
    }

    @Override
    public void releasePermission() {
        stateReference.get().releasePermission();
    }

    @Override
    public void acquirePermission() {
        try {
            stateReference.get().acquirePermission();
        } catch (Exception e) {
            publishCallNotPermittedEvent();
            throw e;
        }
    }

    @Override
    public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
        // Handle the case if the completable future throws a CompletionException wrapping the original exception
        // where original exception is the one to retry not the CompletionException.
        if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
            Throwable cause = throwable.getCause();
            handleThrowable(duration, durationUnit, cause);
        } else {
            handleThrowable(duration, durationUnit, throwable);
        }
    }

    private void handleThrowable(long duration, TimeUnit durationUnit, Throwable throwable) {
        if (circuitBreakerConfig.getIgnoreExceptionPredicate().test(throwable)) {
            LOG.debug("CircuitBreaker '{}' ignored an exception:", name, throwable);
            releasePermission();
            publishCircuitIgnoredErrorEvent(name, duration, durationUnit, throwable);
            return;
        }
        if (circuitBreakerConfig.getRecordExceptionPredicate().test(throwable)) {
            LOG.debug("CircuitBreaker '{}' recorded an exception as failure:", name, throwable);
            publishCircuitErrorEvent(name, duration, durationUnit, throwable);
            stateReference.get().onError(duration, durationUnit, throwable);
        } else {
            LOG.debug("CircuitBreaker '{}' recorded an exception as success:", name, throwable);
            publishSuccessEvent(duration, durationUnit);
            stateReference.get().onSuccess(duration, durationUnit);
        }
        handlePossibleTransition(Either.right(throwable));
    }

    @Override
    public void onSuccess(long duration, TimeUnit durationUnit) {
        LOG.debug("CircuitBreaker '{}' succeeded:", name);
        publishSuccessEvent(duration, durationUnit);
        stateReference.get().onSuccess(duration, durationUnit);
    }

    @Override
    public void onResult(long duration, TimeUnit durationUnit, @Nullable Object result) {
        if (result != null && circuitBreakerConfig.getRecordResultPredicate().test(result)) {
            LOG.debug("CircuitBreaker '{}' recorded a result type '{}' as failure:", name, result.getClass());
            ResultRecordedAsFailureException failure = new ResultRecordedAsFailureException(name, result);
            publishCircuitErrorEvent(name, duration, durationUnit, failure);
            stateReference.get().onError(duration, durationUnit, failure);
        } else {
            onSuccess(duration, durationUnit);
            if (result != null) {
                handlePossibleTransition(Either.left(result));
            }
        }
    }

    private void handlePossibleTransition(Either<Object, Throwable> result) {
        CircuitBreakerConfig.TransitionCheckResult transitionCheckResult = circuitBreakerConfig.getTransitionOnResult()
            .apply(result);
        stateReference.get().handlePossibleTransition(transitionCheckResult);
    }

    /**
     * Get the state of this CircuitBreaker.
     *
     * @return the state of this CircuitBreaker
     */
    @Override
    public State getState() {
        return this.stateReference.get().getState();
    }

    /**
     * Get the name of this CircuitBreaker.
     *
     * @return the name of this CircuitBreaker
     */
    @Override
    public String getName() {
        return this.name;
    }

    /**
     * Get the config of this CircuitBreaker.
     *
     * @return the config of this CircuitBreaker
     */
    @Override
    public CircuitBreakerConfig getCircuitBreakerConfig() {
        return circuitBreakerConfig;
    }

    @Override
    public Metrics getMetrics() {
        return this.stateReference.get().getMetrics();
    }

    @Override
    public Map<String, String> getTags() {
        return tags;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return String.format("CircuitBreaker '%s'", this.name);
    }

    @Override
    public void reset() {
        CircuitBreakerState previousState = stateReference
            .getAndUpdate(currentState -> new ClosedState());
        if (previousState.getState() != CLOSED) {
            publishStateTransitionEvent(
                StateTransition.transitionBetween(getName(), previousState.getState(), CLOSED));
        }
        publishResetEvent();
    }

    private void stateTransition(State newState,
        UnaryOperator<CircuitBreakerState> newStateGenerator) {
        CircuitBreakerState previousState = stateReference.getAndUpdate(currentState -> {
            StateTransition.transitionBetween(getName(), currentState.getState(), newState);
            currentState.preTransitionHook();
            return newStateGenerator.apply(currentState);
        });
        publishStateTransitionEvent(
            StateTransition.transitionBetween(getName(), previousState.getState(), newState));
    }

    @Override
    public void transitionToDisabledState() {
        stateTransition(DISABLED, currentState -> new DisabledState());
    }

    @Override
    public void transitionToMetricsOnlyState() {
        stateTransition(METRICS_ONLY, currentState -> new MetricsOnlyState());
    }

    @Override
    public void transitionToForcedOpenState() {
        stateTransition(FORCED_OPEN,
            currentState -> new ForcedOpenState(currentState.attempts() + 1));
    }

    @Override
    public void transitionToClosedState() {
        stateTransition(CLOSED, currentState -> new ClosedState());
    }

    @Override
    public void transitionToOpenState() {
        stateTransition(OPEN,
            currentState -> new OpenState(currentState.attempts() + 1, currentState.getMetrics()));
    }

    @Override
    public void transitionToOpenStateFor(Duration waitDuration) {
        stateTransition(OPEN,
            currentState -> new OpenState(currentState.attempts(), waitDuration, currentState.getMetrics()));
    }

    @Override
    public void transitionToOpenStateUntil(Instant waitUntil) {
        stateTransition(OPEN,
            currentState -> new OpenState(currentState.attempts(), waitUntil, currentState.getMetrics()));
    }

    @Override
    public void transitionToHalfOpenState() {
        stateTransition(HALF_OPEN, currentState -> new HalfOpenState(currentState.attempts()));
    }

    private boolean shouldPublishEvents(CircuitBreakerEvent event) {
        return stateReference.get().shouldPublishEvents(event);
    }

    private void publishEventIfHasConsumer(CircuitBreakerEvent event) {
        if (!eventProcessor.hasConsumers()) {
            LOG.debug("No Consumers: Event {} not published", event.getEventType());
            return;
        }

        publishEvent(event);
    }

    private void publishEvent(CircuitBreakerEvent event) {
        if (shouldPublishEvents(event)) {
            try {
                eventProcessor.consumeEvent(event);
                LOG.debug("Event {} published: {}", event.getEventType(), event);
            } catch (Throwable t) {
                LOG.warn("Failed to handle event {}", event.getEventType(), t);
            }
        } else {
            LOG.debug("Publishing not allowed: Event {} not published", event.getEventType());
        }
    }

    private void publishStateTransitionEvent(final StateTransition stateTransition) {
        if (StateTransition.isInternalTransition(stateTransition)) {
            return;
        }
        publishEventIfHasConsumer(new CircuitBreakerOnStateTransitionEvent(name, stateTransition));
    }

    private void publishResetEvent() {
        publishEventIfHasConsumer(new CircuitBreakerOnResetEvent(name));
    }

    private void publishCallNotPermittedEvent() {
        publishEventIfHasConsumer(new CircuitBreakerOnCallNotPermittedEvent(name));
    }

    private void publishSuccessEvent(final long duration, TimeUnit durationUnit) {
        if (eventProcessor.hasConsumers()) {
            publishEvent(new CircuitBreakerOnSuccessEvent(name, elapsedDuration(duration, durationUnit)));
        }
    }

    private Duration elapsedDuration(final long duration, TimeUnit durationUnit) {
        return Duration.ofNanos(durationUnit.toNanos(duration));
    }

    private void publishCircuitErrorEvent(final String name, final long duration, final TimeUnit durationUnit,
                                          final Throwable throwable) {
        if (eventProcessor.hasConsumers()) {
            final Duration elapsedDuration = elapsedDuration(duration, durationUnit);
            publishEvent(new CircuitBreakerOnErrorEvent(name, elapsedDuration, throwable));
        }
    }

    private void publishCircuitIgnoredErrorEvent(final String name, long duration, final TimeUnit durationUnit,
                                                 final Throwable throwable) {
        final Duration elapsedDuration = elapsedDuration(duration, durationUnit);
        publishEventIfHasConsumer(new CircuitBreakerOnIgnoredErrorEvent(name, elapsedDuration, throwable));
    }

    private void publishCircuitFailureRateExceededEvent(final String name, float failureRate) {
        publishEventIfHasConsumer(new CircuitBreakerOnFailureRateExceededEvent(name, failureRate));
    }

    private void publishCircuitSlowCallRateExceededEvent(final String name, float slowCallRate) {
        publishEventIfHasConsumer(new CircuitBreakerOnSlowCallRateExceededEvent(name, slowCallRate));
    }

    private void publishCircuitThresholdsExceededEvent(final Result result, final CircuitBreakerMetrics metrics) {
        if (Result.hasFailureRateExceededThreshold(result)) {
            publishCircuitFailureRateExceededEvent(getName(), metrics.getFailureRate());
        }

        if (Result.hasSlowCallRateExceededThreshold(result)) {
            publishCircuitSlowCallRateExceededEvent(getName(), metrics.getSlowCallRate());
        }
    }

    @Override
    public EventPublisher getEventPublisher() {
        return eventProcessor;
    }

    private interface CircuitBreakerState {

        boolean tryAcquirePermission();

        void acquirePermission();

        void releasePermission();

        void onError(long duration, TimeUnit durationUnit, Throwable throwable);

        void onSuccess(long duration, TimeUnit durationUnit);

        void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result);

        int attempts();

        CircuitBreaker.State getState();

        CircuitBreakerMetrics getMetrics();

        /**
         * Should the CircuitBreaker in this state publish events
         *
         * @return a boolean signaling if the events should be published
         */
        default boolean shouldPublishEvents(CircuitBreakerEvent event) {
            return event.getEventType().forcePublish || getState().allowPublish;
        }

        /**
         * This method is invoked before transit to other CircuitBreakerState.
         */
        default void preTransitionHook() {
            // noOp
        }
    }

    private class CircuitBreakerEventProcessor extends
        EventProcessor<CircuitBreakerEvent> implements EventConsumer<CircuitBreakerEvent>,
        EventPublisher {

        @Override
        public EventPublisher onSuccess(
            EventConsumer<CircuitBreakerOnSuccessEvent> onSuccessEventConsumer) {
            registerConsumer(CircuitBreakerOnSuccessEvent.class.getName(),
                onSuccessEventConsumer);
            return this;
        }

        @Override
        public EventPublisher onError(
            EventConsumer<CircuitBreakerOnErrorEvent> onErrorEventConsumer) {
            registerConsumer(CircuitBreakerOnErrorEvent.class.getName(),
                onErrorEventConsumer);
            return this;
        }

        @Override
        public EventPublisher onStateTransition(
            EventConsumer<CircuitBreakerOnStateTransitionEvent> onStateTransitionEventConsumer) {
            registerConsumer(CircuitBreakerOnStateTransitionEvent.class.getName(),
                onStateTransitionEventConsumer);
            return this;
        }

        @Override
        public EventPublisher onReset(
            EventConsumer<CircuitBreakerOnResetEvent> onResetEventConsumer) {
            registerConsumer(CircuitBreakerOnResetEvent.class.getName(),
                onResetEventConsumer);
            return this;
        }

        @Override
        public EventPublisher onIgnoredError(
            EventConsumer<CircuitBreakerOnIgnoredErrorEvent> onIgnoredErrorEventConsumer) {
            registerConsumer(CircuitBreakerOnIgnoredErrorEvent.class.getName(),
                onIgnoredErrorEventConsumer);
            return this;
        }

        @Override
        public EventPublisher onCallNotPermitted(
            EventConsumer<CircuitBreakerOnCallNotPermittedEvent> onCallNotPermittedEventConsumer) {
            registerConsumer(CircuitBreakerOnCallNotPermittedEvent.class.getName(),
                onCallNotPermittedEventConsumer);
            return this;
        }

        @Override
        public EventPublisher onFailureRateExceeded(
            EventConsumer<CircuitBreakerOnFailureRateExceededEvent> onFailureRateExceededConsumer) {
            registerConsumer(CircuitBreakerOnFailureRateExceededEvent.class.getName(),
                onFailureRateExceededConsumer);
            return this;
        }

        @Override
        public EventPublisher onSlowCallRateExceeded(
            EventConsumer<CircuitBreakerOnSlowCallRateExceededEvent> onSlowCallRateExceededConsumer) {
            registerConsumer(CircuitBreakerOnSlowCallRateExceededEvent.class.getName(),
                onSlowCallRateExceededConsumer);
            return this;
        }

        @Override
        public void consumeEvent(CircuitBreakerEvent event) {
            super.processEvent(event);
        }
    }

    private class ClosedState implements CircuitBreakerState {

        private final CircuitBreakerMetrics circuitBreakerMetrics;
        private final AtomicBoolean isClosed;

        ClosedState() {
            this.circuitBreakerMetrics = CircuitBreakerMetrics.forClosed(getCircuitBreakerConfig());
            this.isClosed = new AtomicBoolean(true);
        }

        /**
         * Returns always true, because the CircuitBreaker is closed.
         *
         * @return always true, because the CircuitBreaker is closed.
         */
        @Override
        public boolean tryAcquirePermission() {
            return isClosed.get();
        }

        /**
         * Does not throw an exception, because the CircuitBreaker is closed.
         */
        @Override
        public void acquirePermission() {
            // noOp
        }

        @Override
        public void releasePermission() {
            // noOp
        }

        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            // CircuitBreakerMetrics is thread-safe
            checkIfThresholdsExceeded(circuitBreakerMetrics.onError(duration, durationUnit));
        }

        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            // CircuitBreakerMetrics is thread-safe
            checkIfThresholdsExceeded(circuitBreakerMetrics.onSuccess(duration, durationUnit));
        }

        @Override
        public void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result) {
            if (result.isTransitionToOpen() && isClosed.compareAndSet(true, false)) {
                if (result.getWaitDuration() != null) {
                    transitionToOpenStateFor(result.getWaitDuration());
                } else if (result.getWaitUntil() != null) {
                    transitionToOpenStateUntil(result.getWaitUntil());
                } else {
                    throw new IllegalArgumentException("Transition check resulted in open request but now wait " +
                        "attribute was set? This should never happen");
                }
            }
        }

        @Override
        public int attempts() {
            return 0;
        }

        /**
         * Transitions to open state when thresholds have been exceeded.
         *
         * @param result the Result
         */
        private void checkIfThresholdsExceeded(Result result) {
            if (Result.hasExceededThresholds(result) && isClosed.compareAndSet(true, false)) {
                publishCircuitThresholdsExceededEvent(result, circuitBreakerMetrics);
                transitionToOpenState();
            }
        }

        /**
         * Get the state of the CircuitBreaker
         */
        @Override
        public CircuitBreaker.State getState() {
            return CircuitBreaker.State.CLOSED;
        }

        /**
         * Get metrics of the CircuitBreaker
         */
        @Override
        public CircuitBreakerMetrics getMetrics() {
            return circuitBreakerMetrics;
        }
    }

    private class OpenState implements CircuitBreakerState {

        private final int attempts;
        private final Instant retryAfterWaitDuration;
        private final CircuitBreakerMetrics circuitBreakerMetrics;
        private final AtomicBoolean isOpen;

        @Nullable
        private final ScheduledFuture<?> transitionToHalfOpenFuture;

        OpenState(final int attempts, CircuitBreakerMetrics circuitBreakerMetrics) {
            this(attempts, circuitBreakerConfig.getWaitIntervalFunctionInOpenState().apply(attempts),
                circuitBreakerMetrics);
        }

        OpenState(final int attempts, final Duration waitDuration, CircuitBreakerMetrics circuitBreakerMetrics) {
            this(attempts, waitDuration.toMillis(), circuitBreakerMetrics);
        }

        OpenState(final int attempts, final long waitDurationInMillis, CircuitBreakerMetrics circuitBreakerMetrics) {
            this(attempts, waitDurationInMillis, clock.instant().plus(waitDurationInMillis, MILLIS),
                circuitBreakerMetrics);
        }

        OpenState(final int attempts, final Instant waitUntil, CircuitBreakerMetrics circuitBreakerMetrics) {
            this(attempts, Duration.between(clock.instant(), waitUntil).toMillis(), waitUntil,
                circuitBreakerMetrics);
        }

        OpenState(final int attempts, final long waitDurationInMillis, final Instant retryAfterWaitDuration,
                  CircuitBreakerMetrics circuitBreakerMetrics) {
            this.attempts = attempts;
            this.retryAfterWaitDuration = retryAfterWaitDuration;
            this.circuitBreakerMetrics = circuitBreakerMetrics;

            if (circuitBreakerConfig.isAutomaticTransitionFromOpenToHalfOpenEnabled()) {
                ScheduledExecutorService scheduledExecutorService = schedulerFactory.getScheduler();
                transitionToHalfOpenFuture = scheduledExecutorService
                    .schedule(this::toHalfOpenState, waitDurationInMillis, TimeUnit.MILLISECONDS);
            } else {
                transitionToHalfOpenFuture = null;
            }
            isOpen = new AtomicBoolean(true);
        }

        /**
         * Returns false, if the wait duration has not elapsed. Returns true, if the wait duration
         * has elapsed and transitions the state machine to HALF_OPEN state.
         *
         * @return false, if the wait duration has not elapsed. true, if the wait duration has
         * elapsed.
         */
        @Override
        public boolean tryAcquirePermission() {
            // Thread-safe
            if (clock.instant().isAfter(retryAfterWaitDuration)) {
                toHalfOpenState();
                // Check if the call is allowed to run in HALF_OPEN state after state transition
                // super.tryAcquirePermission() doesn't work right that's why the code is copied
                boolean callPermitted = stateReference.get().tryAcquirePermission();
                if (!callPermitted) {
                    publishCallNotPermittedEvent();
                    circuitBreakerMetrics.onCallNotPermitted();
                }
                return callPermitted;
            }
            circuitBreakerMetrics.onCallNotPermitted();
            return false;
        }

        @Override
        public void acquirePermission() {
            if (!tryAcquirePermission()) {
                throw CallNotPermittedException
                    .createCallNotPermittedException(CircuitBreakerStateMachine.this);
            }
        }

        @Override
        public void releasePermission() {
            // noOp
        }

        /**
         * Should never be called when tryAcquirePermission returns false.
         */
        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            // Could be called when Thread 1 invokes acquirePermission when the state is CLOSED, but in the meantime another
            // Thread 2 calls onError and the state changes from CLOSED to OPEN before Thread 1 calls onError.
            // But the onError event should still be recorded, even if it happened after the state transition.
            circuitBreakerMetrics.onError(duration, durationUnit);
        }

        /**
         * Should never be called when tryAcquirePermission returns false.
         */
        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            // Could be called when Thread 1 invokes acquirePermission when the state is CLOSED, but in the meantime another
            // Thread 2 calls onError and the state changes from CLOSED to OPEN before Thread 1 calls onSuccess.
            // But the onSuccess event should still be recorded, even if it happened after the state transition.
            circuitBreakerMetrics.onSuccess(duration, durationUnit);
        }

        @Override
        public void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result) {
            // noOp
        }

        @Override
        public int attempts() {
            return attempts;
        }

        /**
         * Get the state of the CircuitBreaker
         */
        @Override
        public CircuitBreaker.State getState() {
            return CircuitBreaker.State.OPEN;
        }

        @Override
        public CircuitBreakerMetrics getMetrics() {
            return circuitBreakerMetrics;
        }

        @Override
        public void preTransitionHook() {
            cancelAutomaticTransitionToHalfOpen();
        }

        private void toHalfOpenState() {
            lock.lock();

            try {
                if (isOpen.compareAndSet(true, false)) {
                    transitionToHalfOpenState();
                }
            } finally {
                lock.unlock();
            }
        }

        private void cancelAutomaticTransitionToHalfOpen() {
            if (transitionToHalfOpenFuture != null) {
                transitionToHalfOpenFuture.cancel(false);
            }
        }

    }

    private class DisabledState implements CircuitBreakerState {

        private final CircuitBreakerMetrics circuitBreakerMetrics;

        DisabledState() {
            this.circuitBreakerMetrics = CircuitBreakerMetrics
                .forDisabled(getCircuitBreakerConfig());
        }

        /**
         * Returns always true, because the CircuitBreaker is disabled.
         *
         * @return always true, because the CircuitBreaker is disabled.
         */
        @Override
        public boolean tryAcquirePermission() {
            return true;
        }

        /**
         * Does not throw an exception, because the CircuitBreaker is disabled.
         */
        @Override
        public void acquirePermission() {
            // noOp
        }

        @Override
        public void releasePermission() {
            // noOp
        }

        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            // noOp
        }

        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            // noOp
        }

        @Override
        public void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result) {
            // noOp
        }

        @Override
        public int attempts() {
            return 0;
        }

        /**
         * Get the state of the CircuitBreaker
         */
        @Override
        public CircuitBreaker.State getState() {
            return CircuitBreaker.State.DISABLED;
        }

        /**
         * Get metrics of the CircuitBreaker
         */
        @Override
        public CircuitBreakerMetrics getMetrics() {
            return circuitBreakerMetrics;
        }
    }

    private class MetricsOnlyState implements CircuitBreakerState {

        private final CircuitBreakerMetrics circuitBreakerMetrics;
        private final AtomicBoolean isFailureRateExceeded;
        private final AtomicBoolean isSlowCallRateExceeded;

        MetricsOnlyState() {
            circuitBreakerMetrics = CircuitBreakerMetrics
                .forMetricsOnly(getCircuitBreakerConfig());
            isFailureRateExceeded = new AtomicBoolean(false);
            isSlowCallRateExceeded = new AtomicBoolean(false);
        }

        /**
         * Returns always true, because the CircuitBreaker is always closed in this state.
         *
         * @return always true, because the CircuitBreaker is always closed in this state.
         */
        @Override
        public boolean tryAcquirePermission() {
            return true;
        }

        /**
         * Does not throw an exception, because the CircuitBreaker is always closed in this state.
         */
        @Override
        public void acquirePermission() {
            // noOp
        }

        @Override
        public void releasePermission() {
            // noOp
        }

        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            checkIfThresholdsExceeded(circuitBreakerMetrics.onError(duration, durationUnit));
        }

        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            checkIfThresholdsExceeded(circuitBreakerMetrics.onSuccess(duration, durationUnit));
        }

        @Override
        public void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result) {
            // noOp
        }

        private void checkIfThresholdsExceeded(Result result) {
            if (!Result.hasExceededThresholds(result)) {
                return;
            }

            if (shouldPublishFailureRateExceededEvent(result)) {
                publishCircuitFailureRateExceededEvent(getName(), circuitBreakerMetrics.getFailureRate());
            }

            if (shouldPublishSlowCallRateExceededEvent(result)) {
                publishCircuitSlowCallRateExceededEvent(getName(), circuitBreakerMetrics.getSlowCallRate());
            }
        }

        private boolean shouldPublishFailureRateExceededEvent(Result result) {
            return Result.hasFailureRateExceededThreshold(result) &&
                isFailureRateExceeded.compareAndSet(false, true);
        }

        private boolean shouldPublishSlowCallRateExceededEvent(Result result) {
            return Result.hasSlowCallRateExceededThreshold(result) &&
                isSlowCallRateExceeded.compareAndSet(false, true);
        }

        @Override
        public int attempts() {
            return 0;
        }

        /**
         * Get the state of the CircuitBreaker
         */
        @Override
        public CircuitBreaker.State getState() {
            return CircuitBreaker.State.METRICS_ONLY;
        }

        /**
         * Get metrics of the CircuitBreaker
         */
        @Override
        public CircuitBreakerMetrics getMetrics() {
            return circuitBreakerMetrics;
        }
    }

    private class ForcedOpenState implements CircuitBreakerState {

        private final CircuitBreakerMetrics circuitBreakerMetrics;
        private final int attempts;

        ForcedOpenState(int attempts) {
            this.attempts = attempts;
            this.circuitBreakerMetrics = CircuitBreakerMetrics.forForcedOpen(circuitBreakerConfig);
        }

        /**
         * Returns always false, and records the rejected call.
         *
         * @return always false, since the FORCED_OPEN state always denies calls.
         */
        @Override
        public boolean tryAcquirePermission() {
            circuitBreakerMetrics.onCallNotPermitted();
            return false;
        }

        @Override
        public void acquirePermission() {
            circuitBreakerMetrics.onCallNotPermitted();
            throw CallNotPermittedException
                .createCallNotPermittedException(CircuitBreakerStateMachine.this);
        }

        @Override
        public void releasePermission() {
            // noOp
        }

        /**
         * Should never be called when tryAcquirePermission returns false.
         */
        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            // noOp
        }

        /**
         * Should never be called when tryAcquirePermission returns false.
         */
        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            // noOp
        }

        @Override
        public void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result) {
            // noOp
        }

        @Override
        public int attempts() {
            return attempts;
        }

        /**
         * Get the state of the CircuitBreaker
         */
        @Override
        public CircuitBreaker.State getState() {
            return CircuitBreaker.State.FORCED_OPEN;
        }

        @Override
        public CircuitBreakerMetrics getMetrics() {
            return circuitBreakerMetrics;
        }
    }

    private class HalfOpenState implements CircuitBreakerState {

        private final AtomicInteger permittedNumberOfCalls;
        private final AtomicBoolean isHalfOpen;
        private final int attempts;
        private final CircuitBreakerMetrics circuitBreakerMetrics;
        @Nullable
        private final ScheduledFuture<?> transitionToOpenFuture;

        HalfOpenState(int attempts) {
            int permittedNumberOfCallsInHalfOpenState = circuitBreakerConfig
                .getPermittedNumberOfCallsInHalfOpenState();
            this.circuitBreakerMetrics = CircuitBreakerMetrics
                .forHalfOpen(permittedNumberOfCallsInHalfOpenState, getCircuitBreakerConfig());
            this.permittedNumberOfCalls = new AtomicInteger(permittedNumberOfCallsInHalfOpenState);
            this.isHalfOpen = new AtomicBoolean(true);
            this.attempts = attempts;

            final long maxWaitDurationInHalfOpenState = circuitBreakerConfig.getMaxWaitDurationInHalfOpenState().toMillis();
            if (maxWaitDurationInHalfOpenState >= 1) {
                ScheduledExecutorService scheduledExecutorService = schedulerFactory.getScheduler();
                transitionToOpenFuture = scheduledExecutorService
                    .schedule(this::toOpenState, maxWaitDurationInHalfOpenState, TimeUnit.MILLISECONDS);
            } else {
                transitionToOpenFuture = null;
            }
        }

        /**
         * Checks if test request is allowed.
         * <p>
         * Returns true, if test request counter is not zero. Returns false, if test request counter
         * is zero.
         *
         * @return true, if test request counter is not zero.
         */
        @Override
        public boolean tryAcquirePermission() {
            if (permittedNumberOfCalls.getAndUpdate(current -> current == 0 ? current : --current)
                > 0) {
                return true;
            }
            circuitBreakerMetrics.onCallNotPermitted();
            return false;
        }

        @Override
        public void acquirePermission() {
            if (!tryAcquirePermission()) {
                throw CallNotPermittedException
                    .createCallNotPermittedException(CircuitBreakerStateMachine.this);
            }
        }

        @Override
        public void preTransitionHook() {
            cancelAutomaticTransitionToOpen();
        }

        private void cancelAutomaticTransitionToOpen() {
            if (transitionToOpenFuture != null && !transitionToOpenFuture.isDone()) {
                transitionToOpenFuture.cancel(true);
            }
        }

        private void toOpenState() {
            if (isHalfOpen.compareAndSet(true, false)) {
                transitionToOpenState();
            }
        }

        @Override
        public void releasePermission() {
            permittedNumberOfCalls.incrementAndGet();
        }

        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            // CircuitBreakerMetrics is thread-safe
            checkIfThresholdsExceeded(circuitBreakerMetrics.onError(duration, durationUnit));
        }

        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            // CircuitBreakerMetrics is thread-safe
            checkIfThresholdsExceeded(circuitBreakerMetrics.onSuccess(duration, durationUnit));
        }

        @Override
        public void handlePossibleTransition(CircuitBreakerConfig.TransitionCheckResult result) {
            // noOp
        }

        @Override
        public int attempts() {
            return attempts;
        }

        /**
         * Transitions to open state when thresholds have been exceeded. Transitions to closed state
         * when thresholds have not been exceeded.
         *
         * @param result the result
         */
        private void checkIfThresholdsExceeded(Result result) {
            if (Result.hasExceededThresholds(result) && isHalfOpen.compareAndSet(true, false)) {
                transitionToOpenState();
            }
            if (result == BELOW_THRESHOLDS && isHalfOpen.compareAndSet(true, false)) {
                transitionToClosedState();
            }
        }

        /**
         * Get the state of the CircuitBreaker
         */
        @Override
        public CircuitBreaker.State getState() {
            return CircuitBreaker.State.HALF_OPEN;
        }

        @Override
        public CircuitBreakerMetrics getMetrics() {
            return circuitBreakerMetrics;
        }
    }
}

READ resilience4j-ratelimiter/src/main/java/io/github/resilience4j/ratelimiter/internal/AtomicRateLimiter.java
/*
 *
 *  Copyright 2016 Robert Winkler and Bohdan Storozhuk
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *
 */
package io.github.resilience4j.ratelimiter.internal;

import static java.lang.Long.min;
import static java.lang.System.nanoTime;
import static java.lang.Thread.currentThread;
import static java.util.Collections.emptyMap;
import static java.util.concurrent.locks.LockSupport.parkNanos;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.UnaryOperator;

import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import io.github.resilience4j.ratelimiter.event.RateLimiterOnDrainedEvent;
import io.github.resilience4j.ratelimiter.event.RateLimiterOnFailureEvent;
import io.github.resilience4j.ratelimiter.event.RateLimiterOnSuccessEvent;

/**
 * {@link AtomicRateLimiter} splits all nanoseconds from the start of epoch into cycles.
 * <p>Each cycle has duration of {@link RateLimiterConfig#getLimitRefreshPeriod} in nanoseconds.
 * <p>By contract on start of each cycle {@link AtomicRateLimiter} should
 * set {@link State#activePermissions} to {@link RateLimiterConfig#getLimitForPeriod}. For the {@link
 * AtomicRateLimiter} callers it is really looks so, but under the hood there is some optimisations
 * that will skip this refresh if {@link AtomicRateLimiter} is not used actively.
 * <p>All {@link AtomicRateLimiter} updates are atomic and state is encapsulated in {@link
 * AtomicReference} to {@link AtomicRateLimiter.State}
 */
public class AtomicRateLimiter implements RateLimiter {

    private final long nanoTimeStart;
    private final String name;
    private final AtomicInteger waitingThreads;
    private final AtomicReference<State> state;
    private final Map<String, String> tags;
    private final RateLimiterEventProcessor eventProcessor;

    public AtomicRateLimiter(String name, RateLimiterConfig rateLimiterConfig) {
        this(name, rateLimiterConfig, emptyMap());
    }

    public AtomicRateLimiter(String name, RateLimiterConfig rateLimiterConfig,
                             Map<String, String> tags) {
        this.name = name;
        this.tags = tags;
        this.nanoTimeStart = nanoTime();

        waitingThreads = new AtomicInteger(0);
        state = new AtomicReference<>(new State(
            rateLimiterConfig, 0, rateLimiterConfig.getLimitForPeriod(), 0
        ));
        eventProcessor = new RateLimiterEventProcessor();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void changeTimeoutDuration(final Duration timeoutDuration) {
        RateLimiterConfig newConfig = RateLimiterConfig.from(state.get().config)
            .timeoutDuration(timeoutDuration)
            .build();
        state.updateAndGet(currentState -> new State(
            newConfig, currentState.activeCycle, currentState.activePermissions,
            currentState.nanosToWait
        ));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void changeLimitForPeriod(final int limitForPeriod) {
        RateLimiterConfig newConfig = RateLimiterConfig.from(state.get().config)
            .limitForPeriod(limitForPeriod)
            .build();
        state.updateAndGet(currentState -> new State(
            newConfig, currentState.activeCycle, currentState.activePermissions,
            currentState.nanosToWait
        ));
    }

    /**
     * Calculates time elapsed from the class loading.
     */
    private long currentNanoTime() {
        return nanoTime() - nanoTimeStart;
    }

    long getNanoTimeStart() {
        return this.nanoTimeStart;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean acquirePermission(final int permits) {
        long timeoutInNanos = state.get().config.getTimeoutDuration().toNanos();
        State modifiedState = updateStateWithBackOff(permits, timeoutInNanos);
        boolean result = waitForPermissionIfNecessary(timeoutInNanos, modifiedState.nanosToWait);
        publishRateLimiterAcquisitionEvent(result, permits);
        return result;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public long reservePermission(final int permits) {
        long timeoutInNanos = state.get().config.getTimeoutDuration().toNanos();
        State modifiedState = updateStateWithBackOff(permits, timeoutInNanos);

        boolean canAcquireImmediately = modifiedState.nanosToWait <= 0;
        if (canAcquireImmediately) {
            publishRateLimiterAcquisitionEvent(true, permits);
            return 0;
        }

        boolean canAcquireInTime = timeoutInNanos >= modifiedState.nanosToWait;
        if (canAcquireInTime) {
            publishRateLimiterAcquisitionEvent(true, permits);
            return modifiedState.nanosToWait;
        }

        publishRateLimiterAcquisitionEvent(false, permits);
        return -1;
    }

    @Override
    public void drainPermissions() {
        AtomicRateLimiter.State prev;
        AtomicRateLimiter.State next;
        do {
            prev = state.get();
            next = calculateNextState(prev.activePermissions, 0, prev);
        } while (!compareAndSet(prev, next));
        if (eventProcessor.hasConsumers()) {
            eventProcessor.consumeEvent(new RateLimiterOnDrainedEvent(name, Math.min(prev.activePermissions, 0)));
        }
    }

    /**
     * Atomically updates the current {@link State} with the results of applying the {@link
     * AtomicRateLimiter#calculateNextState}, returning the updated {@link State}. It differs from
     * {@link AtomicReference#updateAndGet(UnaryOperator)} by constant back off. It means that after
     * one try to {@link AtomicReference#compareAndSet(Object, Object)} this method will wait for a
     * while before try one more time. This technique was originally described in this
     * <a href="https://arxiv.org/abs/1305.5800"> paper</a>
     * and showed great results with {@link AtomicRateLimiter} in benchmark tests.
     *
     * @param timeoutInNanos a side-effect-free function
     * @return the updated value
     */
    private State updateStateWithBackOff(final int permits, final long timeoutInNanos) {
        AtomicRateLimiter.State prev;
        AtomicRateLimiter.State next;
        do {
            prev = state.get();
            next = calculateNextState(permits, timeoutInNanos, prev);
        } while (!compareAndSet(prev, next));
        return next;
    }

    /**
     * Atomically sets the value to the given updated value if the current value {@code ==} the
     * expected value. It differs from {@link AtomicReference#updateAndGet(UnaryOperator)} by
     * constant back off. It means that after one try to {@link AtomicReference#compareAndSet(Object,
     * Object)} this method will wait for a while before try one more time. This technique was
     * originally described in this
     * <a href="https://arxiv.org/abs/1305.5800"> paper</a>
     * and showed great results with {@link AtomicRateLimiter} in benchmark tests.
     *
     * @param current the expected value
     * @param next    the new value
     * @return {@code true} if successful. False return indicates that the actual value was not
     * equal to the expected value.
     */
    private boolean compareAndSet(final State current, final State next) {
        if (state.compareAndSet(current, next)) {
            return true;
        }
        parkNanos(1); // back-off
        return false;
    }

    /**
     * A side-effect-free function that can calculate next {@link State} from current. It determines
     * time duration that you should wait for the given number of permits and reserves it for you,
     * if you'll be able to wait long enough.
     *
     * @param permits        number of permits
     * @param timeoutInNanos max time that caller can wait for permission in nanoseconds
     * @param activeState    current state of {@link AtomicRateLimiter}
     * @return next {@link State}
     */
    private State calculateNextState(final int permits, final long timeoutInNanos,
                                     final State activeState) {
        long cyclePeriodInNanos = activeState.config.getLimitRefreshPeriod().toNanos();
        int permissionsPerCycle = activeState.config.getLimitForPeriod();

        long currentNanos = currentNanoTime();
        long currentCycle = currentNanos / cyclePeriodInNanos;

        long nextCycle = activeState.activeCycle;
        int nextPermissions = activeState.activePermissions;
        if (nextCycle != currentCycle) {
            long elapsedCycles = currentCycle - nextCycle;
            long accumulatedPermissions = elapsedCycles * permissionsPerCycle;
            nextCycle = currentCycle;
            nextPermissions = (int) min(nextPermissions + accumulatedPermissions,
                permissionsPerCycle);
        }
        long nextNanosToWait = nanosToWaitForPermission(
            permits, cyclePeriodInNanos, permissionsPerCycle, nextPermissions, currentNanos,
            currentCycle
        );
        State nextState = reservePermissions(activeState.config, permits, timeoutInNanos, nextCycle,
            nextPermissions, nextNanosToWait);
        return nextState;
    }

    /**
     * Calculates time to wait for the required permits of permissions to get accumulated
     *
     * @param permits              permits of required permissions
     * @param cyclePeriodInNanos   current configuration values
     * @param permissionsPerCycle  current configuration values
     * @param availablePermissions currently available permissions, can be negative if some
     *                             permissions have been reserved
     * @param currentNanos         current time in nanoseconds
     * @param currentCycle         current {@link AtomicRateLimiter} cycle    @return nanoseconds to
     *                             wait for the next permission
     */
    private long nanosToWaitForPermission(final int permits, final long cyclePeriodInNanos,
                                          final int permissionsPerCycle,
                                          final int availablePermissions, final long currentNanos, final long currentCycle) {
        if (availablePermissions >= permits) {
            return 0L;
        }
        long nextCycleTimeInNanos = (currentCycle + 1) * cyclePeriodInNanos;
        long nanosToNextCycle = nextCycleTimeInNanos - currentNanos;
        int permissionsAtTheStartOfNextCycle = availablePermissions + permissionsPerCycle;
        int fullCyclesToWait = divCeil(-(permissionsAtTheStartOfNextCycle - permits), permissionsPerCycle);
        return (fullCyclesToWait * cyclePeriodInNanos) + nanosToNextCycle;
    }

    /**
     * Divide two integers and round result to the bigger near mathematical integer.
     *
     * @param x - should be > 0
     * @param y - should be > 0
     */
    private static int divCeil(int x, int y) {
        return (x + y - 1) / y;
    }

    /**
     * Determines whether caller can acquire permission before timeout or not and then creates
     * corresponding {@link State}. Reserves permissions only if caller can successfully wait for
     * permission.
     *
     * @param config
     * @param permits        permits of permissions
     * @param timeoutInNanos max time that caller can wait for permission in nanoseconds
     * @param cycle          cycle for new {@link State}
     * @param permissions    permissions for new {@link State}
     * @param nanosToWait    nanoseconds to wait for the next permission
     * @return new {@link State} with possibly reserved permissions and time to wait
     */
    private State reservePermissions(final RateLimiterConfig config, final int permits,
                                     final long timeoutInNanos,
                                     final long cycle, final int permissions, final long nanosToWait) {
        boolean canAcquireInTime = timeoutInNanos >= nanosToWait;
        int permissionsWithReservation = permissions;
        if (canAcquireInTime) {
            permissionsWithReservation -= permits;
        }
        return new State(config, cycle, permissionsWithReservation, nanosToWait);
    }

    /**
     * If nanosToWait is bigger than 0 it tries to park {@link Thread} for nanosToWait but not
     * longer then timeoutInNanos.
     *
     * @param timeoutInNanos max time that caller can wait
     * @param nanosToWait    nanoseconds caller need to wait
     * @return true if caller was able to wait for nanosToWait without {@link Thread#interrupt} and
     * not exceed timeout
     */
    private boolean waitForPermissionIfNecessary(final long timeoutInNanos,
                                                 final long nanosToWait) {
        boolean canAcquireImmediately = nanosToWait <= 0;
        boolean canAcquireInTime = timeoutInNanos >= nanosToWait;

        if (canAcquireImmediately) {
            return true;
        }
        if (canAcquireInTime) {
            return waitForPermission(nanosToWait);
        }
        waitForPermission(timeoutInNanos);
        return false;
    }

    /**
     * Parks {@link Thread} for nanosToWait.
     * <p>If the current thread is {@linkplain Thread#interrupted}
     * while waiting for a permit then it won't throw {@linkplain InterruptedException}, but its
     * interrupt status will be set.
     *
     * @param nanosToWait nanoseconds caller need to wait
     * @return true if caller was not {@link Thread#interrupted} while waiting
     */
    private boolean waitForPermission(final long nanosToWait) {
        waitingThreads.incrementAndGet();
        long deadline = currentNanoTime() + nanosToWait;
        boolean wasInterrupted = false;
        while (currentNanoTime() < deadline && !wasInterrupted) {
            long sleepBlockDuration = deadline - currentNanoTime();
            parkNanos(sleepBlockDuration);
            wasInterrupted = Thread.interrupted();
        }
        waitingThreads.decrementAndGet();
        if (wasInterrupted) {
            currentThread().interrupt();
        }
        return !wasInterrupted;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String getName() {
        return name;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public RateLimiterConfig getRateLimiterConfig() {
        return state.get().config;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Map<String, String> getTags() {
        return tags;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Metrics getMetrics() {
        return new AtomicRateLimiterMetrics();
    }

    @Override
    public EventPublisher getEventPublisher() {
        return eventProcessor;
    }

    @Override
    public String toString() {
        return "AtomicRateLimiter{" +
            "name='" + name + '\'' +
            ", rateLimiterConfig=" + state.get().config +
            '}';
    }

    /**
     * Get the enhanced Metrics with some implementation specific details.
     *
     * @return the detailed metrics
     */
    public AtomicRateLimiterMetrics getDetailedMetrics() {
        return new AtomicRateLimiterMetrics();
    }

    private void publishRateLimiterAcquisitionEvent(boolean permissionAcquired, int permits) {
        if (!eventProcessor.hasConsumers()) {
            return;
        }
        if (permissionAcquired) {
            eventProcessor.consumeEvent(new RateLimiterOnSuccessEvent(name, permits));
            return;
        }
        eventProcessor.consumeEvent(new RateLimiterOnFailureEvent(name, permits));
    }

    /**
     * <p>{@link AtomicRateLimiter.State} represents immutable state of {@link AtomicRateLimiter}
     * where:
     * <ul>
     * <li>activeCycle - {@link AtomicRateLimiter} cycle number that was used
     * by the last {@link AtomicRateLimiter#acquirePermission()} call.</li>
     * <p>
     * <li>activePermissions - count of available permissions after
     * the last {@link AtomicRateLimiter#acquirePermission()} call.
     * Can be negative if some permissions where reserved.</li>
     * <p>
     * <li>nanosToWait - count of nanoseconds to wait for permission for
     * the last {@link AtomicRateLimiter#acquirePermission()} call.</li>
     * </ul>
     */
    private static class State {

        private final RateLimiterConfig config;

        private final long activeCycle;
        private final int activePermissions;
        private final long nanosToWait;

        private State(RateLimiterConfig config,
                      final long activeCycle, final int activePermissions, final long nanosToWait) {
            this.config = config;
            this.activeCycle = activeCycle;
            this.activePermissions = activePermissions;
            this.nanosToWait = nanosToWait;
        }

    }

    /**
     * Enhanced {@link Metrics} with some implementation specific details
     */
    public class AtomicRateLimiterMetrics implements Metrics {

        private AtomicRateLimiterMetrics() {
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public int getNumberOfWaitingThreads() {
            return waitingThreads.get();
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public int getAvailablePermissions() {
            State currentState = state.get();
            State estimatedState = calculateNextState(1, -1, currentState);
            return estimatedState.activePermissions;
        }

        /**
         * @return estimated time duration in nanos to wait for the next permission
         */
        public long getNanosToWait() {
            State currentState = state.get();
            State estimatedState = calculateNextState(1, -1, currentState);
            return estimatedState.nanosToWait;
        }

        /**
         * @return estimated current cycle
         */
        public long getCycle() {
            State currentState = state.get();
            State estimatedState = calculateNextState(1, -1, currentState);
            return estimatedState.activeCycle;
        }

    }
}

READ resilience4j-bulkhead/src/main/java/io/github/resilience4j/bulkhead/Bulkhead.java
/*
 *
 *  Copyright 2017: Robert Winkler, Lucas Lech
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 *
 */
package io.github.resilience4j.bulkhead;

import io.github.resilience4j.bulkhead.event.BulkheadEvent;
import io.github.resilience4j.bulkhead.event.BulkheadOnCallFinishedEvent;
import io.github.resilience4j.bulkhead.event.BulkheadOnCallPermittedEvent;
import io.github.resilience4j.bulkhead.event.BulkheadOnCallRejectedEvent;
import io.github.resilience4j.bulkhead.internal.SemaphoreBulkhead;
import io.github.resilience4j.core.EventConsumer;
import io.github.resilience4j.core.exception.AcquirePermissionCancelledException;
import io.github.resilience4j.core.functions.*;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

import static java.util.Collections.emptyMap;

/**
 * A Bulkhead instance is thread-safe can be used to decorate multiple requests.
 * <p>
 * A {@link Bulkhead} represent an entity limiting the amount of parallel operations. It does not
 * assume nor does it mandate usage of any particular concurrency and/or io model. These details are
 * left for the client to manage. This bulkhead, depending on the underlying concurrency/io model
 * can be used to shed load, and, where it makes sense, limit resource use (i.e. limit amount of
 * threads/actors involved in a particular flow, etc).
 * <p>
 * In order to execute an operation protected by this bulkhead, a permission must be obtained by
 * calling {@link Bulkhead#tryAcquirePermission()} ()} If the bulkhead is full, no additional
 * operations will be permitted to execute until space is available.
 * <p>
 * Once the operation is complete, regardless of the result, client needs to call {@link
 * Bulkhead#onComplete()} in order to maintain integrity of internal bulkhead state.
 */
public interface Bulkhead {

    /**
     * Returns a supplier which is decorated by a bulkhead.
     *
     * @param bulkhead the Bulkhead
     * @param supplier the original supplier
     * @param <T>      the type of results supplied by this supplier
     * @return a supplier which is decorated by a Bulkhead.
     */
    static <T> CheckedSupplier<T> decorateCheckedSupplier(Bulkhead bulkhead,
                                                          CheckedSupplier<T> supplier) {
        return () -> {
            bulkhead.acquirePermission();
            try {
                return supplier.get();
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a supplier which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param supplier the original supplier
     * @param <T>      the type of the returned CompletionStage's result
     * @return a supplier which is decorated by a Bulkhead.
     */
    static <T> Supplier<CompletionStage<T>> decorateCompletionStage(Bulkhead bulkhead,
        Supplier<CompletionStage<T>> supplier) {
        return () -> {

            final CompletableFuture<T> promise = new CompletableFuture<>();

            if (!bulkhead.tryAcquirePermission()) {
                promise.completeExceptionally(
                    BulkheadFullException.createBulkheadFullException(bulkhead));
            } else {
                try {
                    supplier.get()
                        .whenComplete(
                            (result, throwable) -> {
                                bulkhead.onComplete();
                                if (throwable != null) {
                                    promise.completeExceptionally(throwable);
                                } else {
                                    promise.complete(result);
                                }
                            }
                        );
                } catch (Throwable throwable) {
                    bulkhead.onComplete();
                    promise.completeExceptionally(throwable);
                }
            }

            return promise;
        };
    }

    /**
     * Returns a supplier of type Future which is decorated by a bulkhead. Bulkhead will reserve permission until {@link Future#get()}
     * or {@link Future#get(long, TimeUnit)} is evaluated even if the underlying call took less time to return. Any delays in evaluating
     * future will result in holding of permission in the underlying Semaphore.
     *
     * @param bulkhead the bulkhead
     * @param supplier the original supplier
     * @param <T> the type of the returned Future result
     * @return a supplier which is decorated by a Bulkhead.
     */
    static <T> Supplier<Future<T>> decorateFuture(Bulkhead bulkhead, Supplier<Future<T>> supplier) {
        return () -> {
            if (!bulkhead.tryAcquirePermission()) {
                final CompletableFuture<T> promise = new CompletableFuture<>();
                promise.completeExceptionally(BulkheadFullException.createBulkheadFullException(bulkhead));
                return promise;
            }
            try {
                return new BulkheadFuture<>(bulkhead, supplier.get());
            } catch (Throwable e) {
                bulkhead.onComplete();
                throw e;
            }
        };
    }

    /**
     * Returns a runnable which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param runnable the original runnable
     * @return a runnable which is decorated by a Bulkhead.
     */
    static CheckedRunnable decorateCheckedRunnable(Bulkhead bulkhead, CheckedRunnable runnable) {
        return () -> {
            bulkhead.acquirePermission();
            try {
                runnable.run();
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a callable which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param callable the original Callable
     * @param <T>      the result type of callable
     * @return a supplier which is decorated by a Bulkhead.
     */
    static <T> Callable<T> decorateCallable(Bulkhead bulkhead, Callable<T> callable) {
        return () -> {
            bulkhead.acquirePermission();
            try {
                return callable.call();
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a supplier which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param supplier the original supplier
     * @param <T>      the type of results supplied by this supplier
     * @return a supplier which is decorated by a Bulkhead.
     */
    static <T> Supplier<T> decorateSupplier(Bulkhead bulkhead, Supplier<T> supplier) {
        return () -> {
            bulkhead.acquirePermission();
            try {
                return supplier.get();
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a consumer which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a consumer which is decorated by a Bulkhead.
     */
    static <T> Consumer<T> decorateConsumer(Bulkhead bulkhead, Consumer<T> consumer) {
        return t -> {
            bulkhead.acquirePermission();
            try {
                consumer.accept(t);
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a consumer which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param consumer the original consumer
     * @param <T>      the type of the input to the consumer
     * @return a consumer which is decorated by a Bulkhead.
     */
    static <T> CheckedConsumer<T> decorateCheckedConsumer(Bulkhead bulkhead,
                                                          CheckedConsumer<T> consumer) {
        return t -> {
            bulkhead.acquirePermission();
            try {
                consumer.accept(t);
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a runnable which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param runnable the original runnable
     * @return a runnable which is decorated by a bulkhead.
     */
    static Runnable decorateRunnable(Bulkhead bulkhead, Runnable runnable) {
        return () -> {
            bulkhead.acquirePermission();
            try {
                runnable.run();
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a function which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param function the original function
     * @param <T>      the type of the input to the function
     * @param <R>      the type of the result of the function
     * @return a function which is decorated by a bulkhead.
     */
    static <T, R> Function<T, R> decorateFunction(Bulkhead bulkhead, Function<T, R> function) {
        return (T t) -> {
            bulkhead.acquirePermission();
            try {
                return function.apply(t);
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Returns a function which is decorated by a bulkhead.
     *
     * @param bulkhead the bulkhead
     * @param function the original function
     * @param <T>      the type of the input to the function
     * @param <R>      the type of the result of the function
     * @return a function which is decorated by a bulkhead.
     */
    static <T, R> CheckedFunction<T, R> decorateCheckedFunction(Bulkhead bulkhead,
                                                                CheckedFunction<T, R> function) {
        return (T t) -> {
            bulkhead.acquirePermission();
            try {
                return function.apply(t);
            } finally {
                bulkhead.onComplete();
            }
        };
    }

    /**
     * Create a Bulkhead with a default configuration.
     *
     * @param name the name of the bulkhead
     * @return a Bulkhead instance
     */
    static Bulkhead ofDefaults(String name) {
        return new SemaphoreBulkhead(name);
    }

    /**
     * Creates a bulkhead with a custom configuration
     *
     * @param name   the name of the bulkhead
     * @param config a custom BulkheadConfig configuration
     * @return a Bulkhead instance
     */
    static Bulkhead of(String name, BulkheadConfig config) {
        return of(name, config, emptyMap());
    }

    /**
     * Creates a bulkhead with a custom configuration
     *
     * @param name   the name of the bulkhead
     * @param config a custom BulkheadConfig configuration
     * @param tags   tags added to the Bulkhead
     * @return a Bulkhead instance
     */
    static Bulkhead of(String name, BulkheadConfig config, Map<String, String> tags) {
        return new SemaphoreBulkhead(name, config, tags);
    }

    /**
     * Creates a bulkhead with a custom configuration
     *
     * @param name                   the name of the bulkhead
     * @param bulkheadConfigSupplier custom configuration supplier
     * @return a Bulkhead instance
     */
    static Bulkhead of(String name, Supplier<BulkheadConfig> bulkheadConfigSupplier) {
        return of(name, bulkheadConfigSupplier, emptyMap());
    }

    /**
     * Creates a bulkhead with a custom configuration
     *
     * @param name                   the name of the bulkhead
     * @param bulkheadConfigSupplier custom configuration supplier
     * @param tags                   tags added to the Bulkhead
     * @return a Bulkhead instance
     */
    static Bulkhead of(String name, Supplier<BulkheadConfig> bulkheadConfigSupplier, Map<String, String> tags) {
        return new SemaphoreBulkhead(name, bulkheadConfigSupplier, tags);
    }

    /**
     * Dynamic bulkhead configuration change. NOTE! New `maxWaitTime` duration won't affect threads
     * that are currently waiting for permission.
     *
     * @param newConfig new BulkheadConfig
     */
    void changeConfig(BulkheadConfig newConfig);

    /**
     * Acquires a permission to execute a call, only if one is available at the time of invocation.
     * If the current thread is {@linkplain Thread#interrupt interrupted} while waiting for a permit
     * then it won't throw {@linkplain InterruptedException}, but its interrupt status will be set.
     *
     * @return {@code true} if a permission was acquired and {@code false} otherwise
     */
    boolean tryAcquirePermission();

    /**
     * Acquires a permission to execute a call, only if one is available at the time of invocation
     * If the current thread is {@linkplain Thread#interrupt interrupted} while waiting for a permit
     * then it won't throw {@linkplain InterruptedException}, but its interrupt status will be set.
     *
     * @throws BulkheadFullException               when the Bulkhead is full and no further calls
     *                                             are permitted.
     * @throws AcquirePermissionCancelledException if thread was interrupted during permission wait
     */
    void acquirePermission();

    /**
     * Releases a permission and increases the number of available permits by one.
     * <p>
     * Should only be used when a permission was acquired but not used. Otherwise use {@link
     * Bulkhead#onComplete()} to signal a completed call and release a permission.
     */
    void releasePermission();

    /**
     * Records a completed call and releases a permission.
     */
    void onComplete();

    /**
     * Returns the name of this bulkhead.
     *
     * @return the name of this bulkhead
     */
    String getName();

    /**
     * Returns the BulkheadConfig of this Bulkhead.
     *
     * @return bulkhead config
     */
    BulkheadConfig getBulkheadConfig();

    /**
     * Get the Metrics of this Bulkhead.
     *
     * @return the Metrics of this Bulkhead
     */
    Metrics getMetrics();

    /**
     * Returns an unmodifiable map with tags assigned to this Bulkhead.
     *
     * @return the tags assigned to this Retry in an unmodifiable map
     */
    Map<String, String> getTags();

    /**
     * Returns an EventPublisher which subscribes to the reactive stream of BulkheadEvent and can be
     * used to register event consumers.
     *
     * @return an EventPublisher
     */
    EventPublisher getEventPublisher();

    /**
     * Decorates and executes the decorated Supplier.
     *
     * @param supplier the original Supplier
     * @param <T>      the type of results supplied by this supplier
     * @return the result of the decorated Supplier.
     */
    default <T> T executeSupplier(Supplier<T> supplier) {
        return decorateSupplier(this, supplier).get();
    }

    /**
     * Decorates and executes the decorated Callable.
     *
     * @param callable the original Callable
     * @param <T>      the result type of callable
     * @return the result of the decorated Callable.
     * @throws Exception if unable to compute a result
     */
    default <T> T executeCallable(Callable<T> callable) throws Exception {
        return decorateCallable(this, callable).call();
    }

    /**
     * Decorates and executes the decorated Runnable.
     *
     * @param runnable the original Runnable
     */
    default void executeRunnable(Runnable runnable) {
        decorateRunnable(this, runnable).run();
    }

    /**
     * Decorates and executes the decorated Supplier.
     *
     * @param checkedSupplier the original Supplier
     * @param <T>             the type of results supplied by this supplier
     * @return the result of the decorated Supplier.
     * @throws Throwable if something goes wrong applying this function to the given arguments
     */
    default <T> T executeCheckedSupplier(CheckedSupplier<T> checkedSupplier) throws Throwable {
        return decorateCheckedSupplier(this, checkedSupplier).get();
    }

    /**
     * Decorates and executes the decorated CompletionStage.
     *
     * @param supplier the original CompletionStage
     * @param <T>      the type of results supplied by this supplier
     * @return the decorated CompletionStage.
     */
    default <T> CompletionStage<T> executeCompletionStage(Supplier<CompletionStage<T>> supplier) {
        return decorateCompletionStage(this, supplier).get();
    }


    interface Metrics {


        /**
         * Returns the number of parallel executions this bulkhead can support at this point in
         * time.
         *
         * @return remaining bulkhead depth
         */
        int getAvailableConcurrentCalls();

        /**
         * Returns the configured max amount of concurrent calls allowed for this bulkhead,
         * basically it's a top inclusive bound for the value returned from {@link
         * #getAvailableConcurrentCalls()}.
         *
         * @return max allowed concurrent calls
         */
        int getMaxAllowedConcurrentCalls();
    }

    /**
     * An EventPublisher which can be used to register event consumers.
     */
    @SuppressWarnings("squid:S2176")
    interface EventPublisher extends io.github.resilience4j.core.EventPublisher<BulkheadEvent> {

        EventPublisher onCallRejected(EventConsumer<BulkheadOnCallRejectedEvent> eventConsumer);

        EventPublisher onCallPermitted(EventConsumer<BulkheadOnCallPermittedEvent> eventConsumer);

        EventPublisher onCallFinished(EventConsumer<BulkheadOnCallFinishedEvent> eventConsumer);
    }

    /**
     * This class decorates future with Bulkhead functionality around invocation.
     *
     * @param <T> of return type
     */
    final class BulkheadFuture<T> implements Future<T> {
        private final Future<T> future;
        private final OnceConsumer<Bulkhead> onceToBulkhead;

        BulkheadFuture(Bulkhead bulkhead, Future<T> future) {
            Objects.requireNonNull(future, "Non null Future is required to decorate");
            this.onceToBulkhead = OnceConsumer.of(bulkhead);
            this.future = future;

        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return future.cancel(mayInterruptIfRunning);
        }

        @Override
        public boolean isCancelled() {
            return future.isCancelled();
        }

        @Override
        public boolean isDone() {
            return future.isDone();
        }

        @Override
        public T get() throws InterruptedException, ExecutionException {
            try {
                return future.get();
            }  finally {
                onceToBulkhead.applyOnce(Bulkhead::onComplete);
            }
        }

        @Override
        public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            try {
                return future.get(timeout, unit);
            } finally {
                onceToBulkhead.applyOnce(Bulkhead::onComplete);
            }
        }
    }
}

READ resilience4j-timelimiter/src/main/java/io/github/resilience4j/timelimiter/internal/TimeLimiterImpl.java
package io.github.resilience4j.timelimiter.internal;

import io.github.resilience4j.timelimiter.TimeLimiter;
import io.github.resilience4j.timelimiter.TimeLimiterConfig;
import io.github.resilience4j.timelimiter.event.TimeLimiterEvent;
import io.github.resilience4j.timelimiter.event.TimeLimiterOnErrorEvent;
import io.github.resilience4j.timelimiter.event.TimeLimiterOnSuccessEvent;
import io.github.resilience4j.timelimiter.event.TimeLimiterOnTimeoutEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.function.Supplier;

import static java.util.Collections.emptyMap;

public class TimeLimiterImpl implements TimeLimiter {

    private static final Logger LOG = LoggerFactory.getLogger(TimeLimiterImpl.class);

    private final String name;
    private final Map<String, String> tags;
    private final TimeLimiterConfig timeLimiterConfig;
    private final TimeLimiterEventProcessor eventProcessor;

    public TimeLimiterImpl(String name, TimeLimiterConfig timeLimiterConfig) {
        this(name, timeLimiterConfig, emptyMap());

    }

    public TimeLimiterImpl(String name, TimeLimiterConfig timeLimiterConfig, Map<String, String> tags) {
        this.name = name;
        this.tags = Objects.requireNonNull(tags, "Tags must not be null");
        this.timeLimiterConfig = timeLimiterConfig;
        this.eventProcessor = new TimeLimiterEventProcessor();
    }


    @Override
    public <T, F extends Future<T>> Callable<T> decorateFutureSupplier(Supplier<F> futureSupplier) {
        return () -> {
            Future<T> future = futureSupplier.get();
            try {
                T result = future.get(getTimeLimiterConfig().getTimeoutDuration().toMillis(),
                    TimeUnit.MILLISECONDS);
                onSuccess();
                return result;
            } catch (TimeoutException e) {
                TimeoutException timeoutException = TimeLimiter.createdTimeoutExceptionWithName(name, e);
                onError(timeoutException);
                if (getTimeLimiterConfig().shouldCancelRunningFuture()) {
                    future.cancel(true);
                }
                throw timeoutException;
            } catch (ExecutionException e) {
                Throwable t = e.getCause();
                if (t == null) {
                    onError(e);
                    throw e;
                }
                onError(t);
                if (t instanceof Error) {
                    throw (Error) t;
                }
                throw (Exception) t;
            }
        };
    }

    @Override
    public <T, F extends CompletionStage<T>> Supplier<CompletionStage<T>> decorateCompletionStage(
        ScheduledExecutorService scheduler, Supplier<F> supplier) {

        return () -> {
            CompletableFuture<T> future = supplier.get().toCompletableFuture();
            ScheduledFuture<?> timeoutFuture =
                Timeout
                    .of(future, scheduler, name, getTimeLimiterConfig().getTimeoutDuration().toMillis(),
                        TimeUnit.MILLISECONDS);

            return future.whenComplete((result, throwable) -> {
                // complete
                if (result != null) {
                    if (!timeoutFuture.isDone()) {
                        timeoutFuture.cancel(false);
                    }
                    onSuccess();
                }

                // exceptionally
                if (throwable != null) {
                    if (throwable instanceof CompletionException) {
                        Throwable cause = throwable.getCause();
                        onError(cause);
                    } else if (throwable instanceof ExecutionException) {
                        Throwable cause = throwable.getCause();
                        if (cause == null) {
                            onError(throwable);
                        } else {
                            onError(cause);
                        }
                    } else {
                        onError(throwable);
                    }
                }
            });
        };
    }

    @Override
    public String getName() {
        return name;
    }

    @Override
    public Map<String, String> getTags() {
        return tags;
    }

    @Override
    public TimeLimiterConfig getTimeLimiterConfig() {
        return timeLimiterConfig;
    }

    @Override
    public EventPublisher getEventPublisher() {
        return eventProcessor;
    }

    @Override
    public void onSuccess() {
        if (!eventProcessor.hasConsumers()) {
            return;
        }
        publishEvent(new TimeLimiterOnSuccessEvent(name));
    }

    @Override
    public void onError(Throwable throwable) {
        if (throwable instanceof TimeoutException) {
            onTimeout();
        } else {
            onFailure(throwable);
        }
    }

    private void onTimeout() {
        if (!eventProcessor.hasConsumers()) {
            return;
        }
        publishEvent(new TimeLimiterOnTimeoutEvent(name));
    }

    private void onFailure(Throwable throwable) {
        if (!eventProcessor.hasConsumers()) {
            return;
        }
        publishEvent(new TimeLimiterOnErrorEvent(name, throwable));
    }

    private void publishEvent(TimeLimiterEvent event) {
        try {
            eventProcessor.consumeEvent(event);
            LOG.debug("Event {} published: {}", event.getEventType(), event);
        } catch (Exception e) {
            LOG.warn("Failed to handle event {}", event.getEventType(), e);
        }
    }

    /**
     * Completes CompletableFuture with {@link TimeoutException}.
     */
    static final class Timeout {

        private Timeout() {
        }

        static ScheduledFuture<?> of(
            CompletableFuture<?> future, ScheduledExecutorService scheduler, String name, long delay,
            TimeUnit unit) {
            return scheduler.schedule(() -> {
                if (future != null && !future.isDone()) {
                    future.completeExceptionally(TimeLimiter.createdTimeoutExceptionWithName(name, null));
                }
            }, delay, unit);
        }
    }
}