Reactorで閉鎖可能なリソースを適切に管理する方法

Question

すべての作業が完了したら閉じなければならないhttpクライアントとexecutorがあります。 https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/chapter08/ResourceManagement.java

マイリソースの作成方法：：私はその後、使用

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")); 
} 

：

私はそれがRxJava 1.xのためにここで説明していますように、Flux.usingメソッドを使用しようとしている

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

問題は、最初の要求が行われる前にクライアントの接続が閉じられていることです。

[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) 

Reactorのサンプルは見つかりませんでした。

ありがとうございました

Answer 1

もう一度使用するためのマニュアルを読み、私のエラーが見つかりました。 return Flux.just(client);でクライアントを返すことは意味がありません。Fluxが即時終了してクライアントを終了させるためです。

私が実装してしまった：

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 
    ); 
} 

私は、その後に呼び出した：今、すべての操作が適切な順序になっている

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

。