Skip to content

Commit

Permalink
allow specification of recordFailurePredicate in ratpack configs, closes
Browse files Browse the repository at this point in the history 
 ReactiveX#237, ReactiveX#222
  • Loading branch information
@RobWin
drmaas authored and RobWin committed Jun 13, 2018
1 parent 2b75bc1 commit d5cfb90
Show file tree
Hide file tree
Showing 11 changed files with 233 additions and 94 deletions.
94 changes: 61 additions & 33 deletions resilience4j-documentation/src/docs/asciidoc/addon_guides/ratpack.adoc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,9 +36,18 @@ public class MyModule extends AbstractModule {
@Override
protected void configure() {
CircuitBreakerConfig config = CircuitBreakerConfig.custom();
bind(CircuitBreakerRegistry.class).toInstance(CircuitBreakerRegistry.of(config);
install(Resilience4jModule.class);
CircuitBreakerConfig config = CircuitBreakerConfig.custom().build();
bind(CircuitBreakerRegistry.class).toInstance(CircuitBreakerRegistry.of(config));
Resilience4jModule module = new Resilience4jModule();
module.configure(c -> {
c.circuitBreaker("test", circuitBreakerConfig ->
circuitBreakerConfig
.defaults(false)
.automaticTransitionFromOpenToHalfOpen(false)
.failureRateThreshold(25)
);
});
install(new Resilience4jModule());
}
}
----
Expand All @@ -51,40 +60,40 @@ This will only rate limit the `/` endpoint.
[source,groovy]
----
ratpack {
bindings {
bindInstance(RateLimiterRegistry, RateLimiterRegistry.ofDefaults())
module(Resilience4jModule)
}
handlers {
get(new RateLimiterHandler(registry, 'test'))
get {
render 'success'
}
get('a') {
render 'success'
}
}
bindings {
bindInstance(RateLimiterRegistry, RateLimiterRegistry.ofDefaults())
module(Resilience4jModule)
}
handlers {
get(new RateLimiterHandler(registry, 'test'))
get {
render 'success'
}
get('a') {
render 'success'
}
}
}
----

This will rate limit all endpoints against the same `RateLimiter`.
[source,groovy]
----
ratpack {
bindings {
bindInstance(RateLimiterRegistry, RateLimiterRegistry.ofDefaults())
module(Resilience4jModule)
}
handlers {
all(new RateLimiterHandler(registry, 'test'))
get {
render 'success'
}
get('a') {
render 'success'
}
}
bindings {
bindInstance(RateLimiterRegistry, RateLimiterRegistry.ofDefaults())
module(Resilience4jModule)
}
handlers {
all(new RateLimiterHandler(registry, 'test'))
get {
render 'success'
}
get('a') {
render 'success'
}
}
}
----

==== Promises
Expand All @@ -95,7 +104,7 @@ that is coming from some sort of I/O source.

===== Bulkhead

You can easily apply a Bulkhead to any Ratpack Promise.
You can easily apply a Bulkhead to any Ratpack Promise, given an existing `Bulkhead` instance called `bulkhead`.

[source,java]
----
Expand All @@ -107,7 +116,7 @@ public Promise<String> methodWhichReturnsAPromise() {

===== CircuitBreaker

You can easily apply a CircuitBreaker to any Ratpack Promise.
You can easily apply a CircuitBreaker to any Ratpack Promise, given an existing `CircuitBreaker` instance called `circuitBreaker`.

[source,java]
----
Expand All @@ -117,9 +126,26 @@ public Promise<String> methodWhichReturnsAPromise() {
}
----

You can also specify in-line which exception conditions should be recorded as a failure. In this example when
`MyException` is thrown, it will be recorded as a circuitbreaker failure. Other exceptions will be ignored by
the `circuitBreaker`.

[source,java]
----
public Promise<String> methodWhichReturnsAPromise() {
return backendBConnector.methodWhichReturnsAPromise()
.transform(
CircuitBreakerTransformer
.of(circuitBreaker)
.recover(t -> "recovered")
.recordFailurePredicate(e -> e instanceof MyException)
);
}
----

===== Retry

You can easily apply a Retry to any Ratpack Promise.
You can easily apply a Retry to any Ratpack Promise, given an existing `Retry` instance called `retry`.

[source,java]
----
Expand All @@ -131,7 +157,7 @@ public Promise<String> methodWhichReturnsAPromise() {

===== RateLimiter

You can easily apply a RateLimiter to any Ratpack Promise.
You can easily apply a RateLimiter to any Ratpack Promise, given an existing `RateLimiter` instance called `rateLimiter`.

[source,java]
----
Expand Down Expand Up @@ -306,6 +332,8 @@ ratpack {
}
----

Note that `recordFailurePredicate` cannot be specified via `yaml` configuration.

[source,yaml]
----
resilience4j:
Expand Down
8 changes: 6 additions & 2 deletions resilience4j-ratpack/src/main/java/io/github/resilience4j/ratpack/Resilience4jModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -177,12 +177,16 @@ public void onStart(StartEvent event) throws Exception {
if (circuitBreakerConfig.getDefaults()) {
circuitBreaker = circuitBreakerRegistry.circuitBreaker(name);
} else {
circuitBreaker = circuitBreakerRegistry.circuitBreaker(name, CircuitBreakerConfig.custom()
CircuitBreakerConfig.Builder builder = CircuitBreakerConfig.custom()
.failureRateThreshold(circuitBreakerConfig.getFailureRateThreshold())
.ringBufferSizeInClosedState(circuitBreakerConfig.getRingBufferSizeInClosedState())
.ringBufferSizeInHalfOpenState(circuitBreakerConfig.getRingBufferSizeInHalfOpenState())
.waitDurationInOpenState(Duration.ofMillis(circuitBreakerConfig.getWaitIntervalInMillis()))
.build());
.recordFailure(circuitBreakerConfig.getRecordFailurePredicate());
if (circuitBreakerConfig.isAutomaticTransitionFromOpenToHalfOpen()) {
builder.enableAutomaticTransitionFromOpenToHalfOpen();
}
circuitBreaker = circuitBreakerRegistry.circuitBreaker(name, builder.build());
}
if (endpointsConfig.getCircuitBreakers().isEnabled()) {
circuitBreaker.getEventPublisher().onEvent(cbConsumerRegistry.createEventConsumer(name, endpointsConfig.getCircuitBreakers().getEventConsumerBufferSize()));
Expand Down
30 changes: 7 additions & 23 deletions ...4j-ratpack/src/main/java/io/github/resilience4j/ratpack/bulkhead/BulkheadTransformer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,15 @@

import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadFullException;
import io.github.resilience4j.ratpack.internal.AbstractTransformer;
import ratpack.exec.Downstream;
import ratpack.exec.Upstream;
import ratpack.func.Function;


public class BulkheadTransformer <T> implements Function<Upstream<? extends T>, Upstream<T>> {
public class BulkheadTransformer<T> extends AbstractTransformer<T> {

private final Bulkhead bulkhead;
private Function<Throwable, ? extends T> recover;

private BulkheadTransformer(Bulkhead bulkhead) {
this.bulkhead = bulkhead;
Expand All @@ -47,11 +47,11 @@ public static <T> BulkheadTransformer<T> of(Bulkhead bulkhead) {
/**
* Set a recovery function that will execute when the rateLimiter limit is exceeded.
*
* @param recover the recovery function
* @param recoverer the recovery function
* @return the transformer
*/
public BulkheadTransformer<T> recover(Function<Throwable, ? extends T> recover) {
this.recover = recover;
public BulkheadTransformer<T> recover(Function<Throwable, ? extends T> recoverer) {
this.recoverer = recoverer;
return this;
}

Expand All @@ -71,15 +71,7 @@ public void success(T value) {
@Override
public void error(Throwable throwable) {
bulkhead.onComplete();
try {
if (recover != null) {
down.success(recover.apply(throwable));
} else {
down.error(throwable);
}
} catch (Throwable t) {
down.error(t);
}
handleRecovery(down, throwable);
}

@Override
Expand All @@ -90,15 +82,7 @@ public void complete() {
});
} else {
Throwable t = new BulkheadFullException(String.format("Bulkhead '%s' is full", bulkhead.getName()));
if (recover != null) {
try {
down.success(recover.apply(t));
} catch (Throwable t2) {
down.error(t2);
}
} else {
down.error(t);
}
handleRecovery(down, t);
}
};
}
Expand Down
27 changes: 23 additions & 4 deletions ...ack/src/main/java/io/github/resilience4j/ratpack/circuitbreaker/CircuitBreakerConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,17 +16,18 @@

package io.github.resilience4j.ratpack.circuitbreaker;

import static io.github.resilience4j.circuitbreaker.CircuitBreakerConfig.DEFAULT_MAX_FAILURE_THRESHOLD;
import static io.github.resilience4j.circuitbreaker.CircuitBreakerConfig.DEFAULT_RING_BUFFER_SIZE_IN_CLOSED_STATE;
import static io.github.resilience4j.circuitbreaker.CircuitBreakerConfig.DEFAULT_RING_BUFFER_SIZE_IN_HALF_OPEN_STATE;
import static io.github.resilience4j.circuitbreaker.CircuitBreakerConfig.DEFAULT_WAIT_DURATION_IN_OPEN_STATE;
import java.util.function.Predicate;

import static io.github.resilience4j.circuitbreaker.CircuitBreakerConfig.*;

public class CircuitBreakerConfig {
private boolean defaults = false;
private Integer waitIntervalInMillis = DEFAULT_WAIT_DURATION_IN_OPEN_STATE * 1000;
private Integer failureRateThreshold = DEFAULT_MAX_FAILURE_THRESHOLD;
private Integer ringBufferSizeInClosedState = DEFAULT_RING_BUFFER_SIZE_IN_CLOSED_STATE;
private Integer ringBufferSizeInHalfOpenState = DEFAULT_RING_BUFFER_SIZE_IN_HALF_OPEN_STATE;
private Predicate<Throwable> recordFailurePredicate = DEFAULT_RECORD_FAILURE_PREDICATE;
private boolean automaticTransitionFromOpenToHalfOpen = false;

/**
* Use config provided by circuitbreaker registry instead of these config values.
Expand Down Expand Up @@ -59,6 +60,16 @@ public CircuitBreakerConfig ringBufferSizeInHalfOpenState(Integer ringBufferSize
return this;
}

public CircuitBreakerConfig recordFailurePredicate(Predicate<Throwable> recordFailurePredicate) {
this.recordFailurePredicate = recordFailurePredicate;
return this;
}

public CircuitBreakerConfig automaticTransitionFromOpenToHalfOpen(boolean automaticTransitionFromOpenToHalfOpen) {
this.automaticTransitionFromOpenToHalfOpen = automaticTransitionFromOpenToHalfOpen;
return this;
}

public boolean getDefaults() {
return defaults;
}
Expand All @@ -79,4 +90,12 @@ public Integer getRingBufferSizeInHalfOpenState() {
return ringBufferSizeInHalfOpenState;
}

public Predicate<Throwable> getRecordFailurePredicate() {
return recordFailurePredicate;
}

public boolean isAutomaticTransitionFromOpenToHalfOpen() {
return automaticTransitionFromOpenToHalfOpen;
}

}
41 changes: 21 additions & 20 deletions ...rc/main/java/io/github/resilience4j/ratpack/circuitbreaker/CircuitBreakerTransformer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,17 +17,20 @@

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerOpenException;
import io.github.resilience4j.ratpack.internal.AbstractTransformer;
import ratpack.exec.Downstream;
import ratpack.exec.Upstream;
import ratpack.func.Function;

public class CircuitBreakerTransformer<T> implements Function<Upstream<? extends T>, Upstream<T>> {
import java.util.function.Predicate;

public class CircuitBreakerTransformer<T> extends AbstractTransformer<T> {
private final CircuitBreaker circuitBreaker;
private Function<Throwable, ? extends T> recoverer;
private Predicate<Throwable> recordFailurePredicate;

private CircuitBreakerTransformer(CircuitBreaker circuitBreaker) {
this.circuitBreaker = circuitBreaker;
this.recordFailurePredicate = circuitBreaker.getCircuitBreakerConfig().getRecordFailurePredicate();
}

/**
Expand All @@ -43,6 +46,18 @@ public static <T> CircuitBreakerTransformer<T> of(CircuitBreaker circuitBreaker)
return new CircuitBreakerTransformer<>(circuitBreaker);
}

/**
* Set predicate for which exceptions should record circuitbreaker failure.
* This will override any values configured in {@link CircuitBreakerConfig}.
*
* @param recordFailurePredicate the predicate. When it evaluates to true, the throwable will record an error.
* @return the transformer
*/
public CircuitBreakerTransformer<T> recordFailurePredicate(Predicate<Throwable> recordFailurePredicate) {
this.recordFailurePredicate = recordFailurePredicate;
return this;
}

/**
* Set a recovery function that will execute when the circuit breaker is open.
*
Expand Down Expand Up @@ -72,16 +87,10 @@ public void success(T value) {
@Override
public void error(Throwable throwable) {
long durationInNanos = System.nanoTime() - start;
circuitBreaker.onError(durationInNanos, throwable);
try {
if (recoverer != null) {
down.success(recoverer.apply(throwable));
} else {
down.error(throwable);
}
} catch (Throwable t) {
down.error(t);
if (recordFailurePredicate.test(throwable)) {
circuitBreaker.onError(durationInNanos, throwable);
}
handleRecovery(down, throwable);
}

@Override
Expand All @@ -91,15 +100,7 @@ public void complete() {
});
} else {
Throwable t = new CircuitBreakerOpenException(String.format("CircuitBreaker '%s' is open", circuitBreaker.getName()));
if (recoverer != null) {
try {
down.success(recoverer.apply(t));
} catch (Throwable t2) {
down.error(t2);
}
} else {
down.error(t);
}
handleRecovery(down, t);
}
};
}
Expand Down
Loading

0 comments on commit d5cfb90

Please sign in to comment.