Project Reactor Integration

The fun-reactor module bridges dmx-fun’s Option, Result, and Try with Project Reactor’s Mono, so absence, failure, and exceptions are modeled as values at reactive boundaries instead of being reconstructed with ad-hoc conversion code in every service.

It is an optional module. reactor-core is declared compileOnly — your application already brings Reactor (directly or through Spring WebFlux), so the adapter never forces a version, and the core fun library keeps no dependency on Reactor.

Adding the dependency

// Gradle
implementation("codes.domix:fun-reactor:LATEST_VERSION")
// Project Reactor — bring your own version (3.4.x – 3.8.x)
implementation("io.projectreactor:reactor-core:3.8.6")

<!-- Maven -->
<dependency>
    <groupId>codes.domix</groupId>
    <artifactId>fun-reactor</artifactId>
    <version>LATEST_VERSION</version>
</dependency>
<!-- Project Reactor — bring your own version (3.4.x – 3.8.x) -->
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.8.6</version>
</dependency>

fun-reactor declares reactor-core as compileOnly. Your application already brings Reactor (directly or via Spring WebFlux), so the adapter never forces a version.

The ReactorFun adapters

Every conversion lives on ReactorFun and falls into three groups.

Reactor → dmx-fun (stay reactive)

The toMonoTry and toMonoResult variants absorb the empty and error signals into the value channel, so the resulting Mono never errors for a modeled failure. (toMonoOption is the exception: it maps empty to None, but since Option has no error channel it lets a Reactor error propagate.) Prefer these inside reactive pipelines:

// A flaky lookup that may error or complete empty
Mono<User> lookup = userClient.findById(id);

// Result on the value channel — the pipeline never short-circuits on failure
Mono<Result<User, ApiError>> result =
    ReactorFun.toMonoResult(lookup, ApiError::fromThrowable);

// 404 vs found, without exceptions
Mono<Option<User>> maybeUser = ReactorFun.toMonoOption(lookup);

Empty-Mono policy

A Mono that completes without a value is treated as a missing value:

• toMonoOption → Option.none()
• toMonoTry / toMonoResult → a failure carrying NoSuchElementException (mapped through your errorMapper when you supply one)

toMonoOption is the one conversion that does not absorb errors — since Option has no error channel, an error signal propagates as a Reactor error.

Composing on the Result track

Once you have a Mono<Result<V, E>>, ReactorResult lets you keep composing without unwrapping at every step — Err is threaded through untouched:

Mono<Result<Receipt, ApiError>> receipt =
    ReactorResult.flatMapOk(
        ReactorResult.mapOk(
            ReactorFun.toMonoResult(userClient.findById(id), ApiError::fromThrowable),
            User::accountId),                          // Mono<Result<AccountId, ApiError>>
        accountId -> ReactorFun.toMonoResult(          // dependent reactive step
            billingClient.charge(accountId), ApiError::fromThrowable));

flatMapOk only invokes its mapper on the Ok path; a failure short-circuits the rest of the chain, exactly like a synchronous Result.flatMap.

dmx-fun → Reactor

Going the other way turns a domain outcome back into a Mono:

Mono<User> m1 = ReactorFun.toMono(Option.some(user));           // Mono.just(user)
Mono<User> m2 = ReactorFun.toMono(Option.<User>none());         // Mono.empty()
Mono<User> m3 = ReactorFun.toMono(Try.success(user));           // Mono.just(user)
Mono<User> m4 = ReactorFun.toMono(Result.<User, ApiError>err(e),
                                  ApiError::toException);         // Mono.error(...)

Because Result’s error channel is an arbitrary domain type — not necessarily a Throwable — toMono(result, errorMapper) asks you to say explicitly how a domain error becomes a Reactor error signal. When the error channel is already a Throwable, use the single-argument overload, which sends it straight to Mono.error:

Mono<User> m = ReactorFun.toMono(Result.<User, IllegalStateException>err(ex));

Reading the Reactor context

ReactorContext.getOrNone reads a key from a Reactor ContextView as an Option, so a missing key is None instead of a thrown exception. The ContextView is always passed in — there is no hidden global state:

Mono<String> tenantAware = Mono.deferContextual(ctx ->
    ReactorFun.toMono(ReactorContext.<String>getOrNone(ctx, "tenantId")
        .map(tenant -> resolveFor(tenant)))
);

ContextView was introduced in Reactor 3.4, so the context helpers require Reactor 3.4 or later. The ReactorFun conversions themselves work on any supported 3.x release.

Streaming with Flux

Where ReactorFun handles a single value, ReactorFlux handles a stream. Its centerpiece folds a Flux of outcomes into one Mono<Result<List<…>, …>> — the reactive equivalent of Result.sequence.

// Fail-fast: stop at the first failure
Mono<Result<List<Order>, Failure>> all = ReactorFlux.sequence(orderResultsFlux);

// Accumulate: report every failure at once
Mono<Validated<NonEmptyList<Failure>, List<Order>>> validated =
    ReactorFlux.collectValidated(orderResultsFlux);

// Collect a raw stream, folding a Reactor error into the typed channel
Mono<Result<List<Order>, Failure>> collected =
    ReactorFlux.collectResult(orderFlux, Failure::fromThrowable);

sequence and collectResult treat an empty stream as Ok([]); only sequence short-circuits, cancelling the upstream as soon as it sees an Err. Backpressure and cancellation are preserved throughout.

Migrating from manual conversion code

Hand-rolled Mono → outcome conversions usually look like this — a map, an onErrorResume that rebuilds the error case, and a defaultIfEmpty for the empty case, repeated at every call site:

// Before: bespoke, easy to get subtly wrong (empty vs error handling)
Mono<Result<User, ApiError>> result = userClient.findById(id)
    .map(Result::<User, ApiError>ok)
    .onErrorResume(ex -> Mono.just(Result.err(ApiError.fromThrowable(ex))))
    .defaultIfEmpty(Result.err(ApiError.notFound(id)));

The same behavior is one call with fun-reactor, with the empty policy stated explicitly:

// After: the empty and error policies are named, not reconstructed
Mono<Result<User, ApiError>> result = ReactorFun.toMonoResult(
    userClient.findById(id),
    ApiError::fromThrowable,
    () -> ApiError.notFound(id));

Streaming aggregation collapses the same way. The manual collectList plus error handling:

// Before: collect, then rebuild the error case by hand
Mono<Result<List<Order>, ApiError>> orders = orderClient.findAll(query)
    .collectList()
    .map(Result::<List<Order>, ApiError>ok)
    .onErrorResume(ex -> Mono.just(Result.err(ApiError.fromThrowable(ex))));

becomes a single collectResult:

// After
Mono<Result<List<Order>, ApiError>> orders =
    ReactorFlux.collectResult(orderClient.findAll(query), ApiError::fromThrowable);

Blocking vs reactive

toTry, toResult, and toOption are blocking extractors: they subscribe and wait for the terminal signal via Mono.block(), returning the dmx-fun value directly. They are convenient for bridging into non-reactive code and tests, but must never run on a non-blocking (event-loop) thread:

// Fine in a test or a blocking adapter; never on a Reactor scheduler thread
Result<User, Throwable> result = ReactorFun.toResult(userClient.findById(id));

Inside a reactive pipeline, always prefer the toMono* variants.

Version compatibility

Tested in CI against the following Reactor versions on every pull request that touches the reactor/ module:

To test locally against a specific version:

./gradlew :reactor:test -PreactorVersion=3.7.0