Project Reactor

A method that returns Reactor's Mono<T> gets the asynchronous form of every policy, once per subscription, with the subscriber's context, and is cancelled through the policies when the subscriber cancels.

Install

Add the module next to the others; the processor needs nothing new.

Gradle

dependencies {
    implementation 'dev.resilience4cata:resilience4cata-all:0.11.0'
    implementation 'dev.resilience4cata:resilience4cata-reactor:0.11.0'
    annotationProcessor 'dev.resilience4cata:resilience4cata-processor:0.11.0'
}

Gradle (Kotlin)

dependencies {
    implementation("dev.resilience4cata:resilience4cata-all:0.11.0")
    implementation("dev.resilience4cata:resilience4cata-reactor:0.11.0")
    annotationProcessor("dev.resilience4cata:resilience4cata-processor:0.11.0")
}

Maven

<dependency>
  <groupId>dev.resilience4cata</groupId>
  <artifactId>resilience4cata-reactor</artifactId>
  <version>0.11.0</version>
</dependency>

The module depends on resilience4cata-core and reactor-core; your application's own, newer Reactor wins. A Mono method compiled without it is the error R4C-ASY-003.

A Mono interface

The annotations are the same as on any other method; the return type decides.

public interface PriceClient {

    @Retry(maxAttempts = 3, delay = "50ms")
    @CircuitBreaker(name = "prices")
    @Timeout("2s")
    @Fallback(method = "unknown")
    Mono<BigDecimal> price(String sku);

    default Mono<BigDecimal> unknown(String sku, Throwable failure) {
        return Mono.just(BigDecimal.ZERO);
    }
}

ResilienceRegistry registry = ResilienceRegistry.builder().build();
PriceClient prices = Resilient.of(PriceClient.class, http, registry);

Mono<BigDecimal> price = prices.price("sku-1");   // nothing runs yet
BigDecimal value = price.block();                 // subscribes: two failed attempts, then 42

Annotated classes work the same way (annotated classes), and so do Spring, Micronaut and Quarkus beans: in a WebFlux controller, return the protected bean's Mono as it is.

What a Mono method does

• Nothing runs before a subscription. Calling price("sku-1") builds a Mono; each subscription is one call through the policies, so subscribing twice (or repeat()) calls twice.
• Each attempt calls the method again. A retry or a hedge calls your implementation for a new Mono and subscribes to it, as for a CompletionStage method. A method that throws, or returns null, is a failed attempt.
• The subscriber's context reaches every attempt, and the fallback's Mono too, so tracing, Spring Security's ReactiveSecurityContextHolder and your own keys work inside the policies:

Mono<BigDecimal> price = prices.price("sku-1")
        .contextWrite(Context.of("tenant", "acme"));   // the delegate's Mono, and every retry, see it

• An empty Mono is a success with no value (null for a classifier), and the result is empty too.
• Cancelling the subscription cancels the call through every policy: the current attempt's Mono is cancelled, a retry waiting for its delay stops, the breaker records nothing, the bulkhead and adaptive limit free their slot. A client that disconnects from a WebFlux endpoint does this.
• Failures arrive without demand, values on demand, as Reactive Streams requires; the generated Mono passes the Reactive Streams TCK.
• No policy blocks the thread, which BlockHound checks in the build: a rate limiter or a bulkhead with maxWait makes the call wait without parking a thread, as for any async method.

Which thread emits: the first attempt subscribes on the subscribing thread. Retry delays, hedges, timeouts and rate-limiter waits fire on the library's timer, which hands the work to a new virtual thread, so the value of a later attempt arrives there. Add publishOn(...) after the method if the next operators need a particular scheduler.

Measured On AWS Graviton3, a Mono method with a circuit breaker costs 130 ns per call against 246 ns for resilience4j-reactor 2.4.0's operator (119 vs 1379 ns with four threads), and a retry, breaker, timeout and bulkhead stack 227 ns against 4338 ns, including the 52 ns of subscribing to the Mono. See Performance.

Fallbacks

A fallback of a Mono method returns the value (BigDecimal) or a Mono of it (Mono<BigDecimal>), as a sibling method or a FallbackHandler; a CompletionStage is not accepted. Its Mono is subscribed with the caller's context.

Code you cannot annotate

PipelineOperator applies a pipeline to a Mono, for a WebClient call or a third-party client:

PipelineOperator<String> resilient = PipelineOperator.of(Pipeline.builder(registry, "rates", String.class)
        .retry(r -> r.maxAttempts(3).delay(Duration.ofMillis(50)))
        .timeout(t -> t.timeout(Duration.ofSeconds(2)))
        .build());

Mono<String> rates = webClientCall.transform(resilient);   // each attempt subscribes to webClientCall again

Every attempt subscribes to the source again, which runs a WebClient request again. A source that gives the same outcome to every subscriber (cache(), a hot source) is not retried meaningfully; build it per attempt with resilient.defer(() -> ...) instead.

Testing

On a ManualClock, delays, timeouts and hedges happen when the test advances the clock, and StepVerifier checks the outcome:

ManualClock clock = new ManualClock();
ResilienceRegistry registry = ResilienceTesting.registry(clock).build();
PriceClient prices = Resilient.of(PriceClient.class, slow, registry);

StepVerifier.create(prices.price("sku-1"))
        .then(() -> clock.advance(Duration.ofSeconds(7)))   // three 2 s timeouts and the delays between them
        .expectNext(BigDecimal.ZERO)                        // the fallback
        .verifyComplete();

The manual clock is the library's; StepVerifier.withVirtualTime moves Reactor's schedulers, not the policies' timers.

Limits

• Flux is not supported yet: a method that returns one is R4C-ASY-002. Return Mono<List<T>> (collectList()) when the items fit in memory.
• Exactly Mono: a subclass of it, Mono<? super T> or a raw Mono is not a Mono method (R4C-ASY-001, or R4C-ASY-002 for a subclass).
• Transactions: a reactive @Transactional on a retried or hedged Mono method of an annotated class would run every attempt in one transaction, which Spring, Micronaut and Quarkus report at startup, as for other methods (R4C-SPR-004).
• Mutiny and RxJava are not supported yet; convert to a CompletionStage (uni.subscribeAsCompletionStage()).