onCompleteWithBreaker
Description
Evaluates its parameter of type CompletionStage<T>
protecting it with the specified CircuitBreaker
. Refer to Circuit Breaker for a detailed description of this pattern.
If the CircuitBreaker
is open, the request is rejected with a CircuitBreakerOpenRejection
. Note that in this case the request’s entity databytes stream is cancelled, and the connection is closed as a consequence.
Otherwise, the same behaviour provided by onComplete is to be expected.
Example
// import static scala.compat.java8.JFunction.func;
// import static akka.http.javadsl.server.PathMatchers.*;
final int maxFailures = 1;
final FiniteDuration callTimeout = FiniteDuration.create(5, TimeUnit.SECONDS);
final FiniteDuration resetTimeout = FiniteDuration.create(1, TimeUnit.SECONDS);
final CircuitBreaker breaker = CircuitBreaker.create(system().scheduler(), maxFailures, callTimeout, resetTimeout);
final Route route = path(segment("divide").slash(integerSegment()).slash(integerSegment()),
(a, b) -> onCompleteWithBreaker(breaker,
() -> CompletableFuture.supplyAsync(() -> a / b),
maybeResult -> maybeResult
.map(func(result -> complete("The result was " + result)))
.recover(new PFBuilder<Throwable, Route>()
.matchAny(ex -> complete(StatusCodes.InternalServerError(),
"An error occurred: " + ex.getMessage())
)
.build())
.get()
)
);
testRoute(route).run(HttpRequest.GET("/divide/10/2"))
.assertEntity("The result was 5");
testRoute(route).run(HttpRequest.GET("/divide/10/0"))
.assertStatusCode(StatusCodes.InternalServerError())
.assertEntity("An error occurred: / by zero");
// opened the circuit-breaker
testRoute(route).run(HttpRequest.GET("/divide/10/0"))
.assertStatusCode(StatusCodes.ServiceUnavailable())
.assertEntity("The server is currently unavailable (because it is overloaded or down for maintenance).");
Thread.sleep(resetTimeout.toMillis() + 300);
// circuit breaker resets after this time
testRoute(route).run(HttpRequest.GET("/divide/8/2"))
.assertEntity("The result was 4");