CODEWITHSUNDEEP
javaproject-reactor
Mavo
Mavo

Reputation: 625

How to properly manage closable resources in Reactor

I have a http client and executor which should be closed when all work is done.

I'm trying to use Flux.using method in a way it's described here for RxJava 1.x: https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/chapter08/ResourceManagement.java

My resource creating method:

public static Flux<GithubClient> createResource(String token,
                                                int connectionCount) {

    return Flux.using(
            () -> {
                logger.info(Thread.currentThread().getName() + " : Created and started the client.");
                return new GithubClient(token, connectionCount);
            },
            client -> {
                logger.info(Thread.currentThread().getName() + " : About to create Observable.");
                return Flux.just(client);
            },
            client -> {
                logger.info(Thread.currentThread().getName() + " : Closing the client.");
                client.close();
            },
            false
    ).doOnSubscribe(subscription -> logger.info("subscribed"));
}

Which I then use:

Flux<StateMutator> dataMutators = GithubClient.createResource(
            config.getAccessToken(),
            config.getConnectionCount())
            .flatMap(client -> client.loadRepository(organization, repository)

The problem is that the client connection is closed even before first request is made.

[main] INFO com.sapho.services.githubpublic.client.GithubClient - main : Created and started the client.
[main] INFO com.sapho.services.githubpublic.client.GithubClient - main : About to create Observable.
[main] INFO com.sapho.services.githubpublic.client.GithubClient - subscribed
[main] INFO com.sapho.services.githubpublic.client.GithubClient - main : Closing the client.

java.lang.IllegalStateException: Client instance has been closed.

at jersey.repackaged.com.google.common.base.Preconditions.checkState(Preconditions.java:173)
at org.glassfish.jersey.client.JerseyClient.checkNotClosed(JerseyClient.java:273)

Did not find any example for Reactor.

Upvotes: 2

Views: 2131

Answers (1)

Mavo
Mavo

Reputation: 625

I read the documentation for using again and found my error. Returning the client with return Flux.just(client); does not makes sense as the Flux terminates immediatelly which triggers client closing.

I ended up implementing: 

public static Flux<StateMutator> createAndExecute(GithubPublicConfiguration config,
                                                  Function<GithubClient, Flux<StateMutator>> toExecute) {

    return Flux.using(
            () -> {
                logger.debug(Thread.currentThread().getName() + " : Created and started the client.");
                return new GithubClient(entityModelHandler, config.getAccessToken(), config.getConnectionCount());
            },
            client -> toExecute.apply(client),
            client -> {
                logger.debug(Thread.currentThread().getName() + " : Closing the client.");
                client.close();
            },
            false
    );
}

which I then call with:

GithubClient.createAndExecute(config,
            client -> client.loadRepository(organization, repository))

Now all operations are in appropriate order.

Upvotes: 1

Related Questions