How can I use the try-with-resources patter in reactor, if resource is a Mono?

I have a client, which is being created as a Mono, e.g.

Mono<MyClient> myClientMono = ...

The MyClient class implements Closeable interface, so you can use the standard java try-with-resources pattern.

What's the best way to use resource wrapped in a Mono?

I understand how to use Mono.using() to mimic try-with-resources if the resource is a regular object, like a InputStream.

However, how can I make sure myclient is being closed after it has been used with Reactor?

I could do this:

Mono.using(() -> myClientMono.block(), myClient -> myClient.doSomething(), myClient.close()) .subscribe(System.out::println); 

however, this will be blocking and cannot be used in an asynchronous thread.

I could do something like this and close the client after it has been used:

myClientMono .flatMap( myClient -> { var a = myClient.doSomething(); myClient.close(); return Mono.just(a); }) .subscribe(Sytem.out::println); 

This - however - does not close the client if an error occurs.

I wonder if there's a better solution?

1 Answer

The proper solution is to use Mono.usingWhen or Flux.usingWhen.

In both case, you provide a resource publisher, that will be called each time the user subscribe, to produce a fresh resource.

Let's see that in an example:

import java.time.Duration; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; public class UsingWhen { /** Simulates a closeable resource. Prints a message when opening and closing. */ public static class ReactiveResourceMock implements AutoCloseable { public ReactiveResourceMock() { System.out.println("OPENING"); } Flux<Integer> content() { return Flux.range(1, 3); } @Override public void close() throws Exception { System.out.println("CLOSING"); }; } /** Simulates async value opening */ public static Mono<ReactiveResourceMock> open() { return Mono.delay(Duration.ofMillis(200)) .map(i -> new ReactiveResourceMock()); } public static void main(String[] args) { Mono<ReactiveResourceMock> openResource = open(); Flux<Integer> flow = Flux.usingWhen( openResource, ReactiveResourceMock::content, resource -> Mono.fromCallable(() -> { resource.close(); return 1; })) .doOnCancel(() -> System.out.println("CANCELLED")) .doOnComplete(() -> System.out.println("COMPLETED")); System.out.println("VERIFY FLOW CONTROL:"); flow.doOnNext(System.out::println) .blockLast(); System.out.println("WHAT HAPPENS IN CASE OF ERROR:"); flow.limitRate(1) .map(i -> i / (i % 2)) .onErrorReturn(-1) .doOnNext(System.out::println) .blockLast(); } } 

The program above prints:

VERIFY FLOW CONTROL: OPENING 1 2 3 CLOSING COMPLETED WHAT HAPPENS IN CASE OF ERROR: OPENING 1 CANCELLED -1 CLOSING