如何正确管理反应堆中的可关闭资源

Question

我有一个http客户端和执行程序，应在所有工作完成后关闭。

我试图用Flux.using方法，它是在这里描述RxJava 1.x的一种方式： 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")); 
} 

然后我用：

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) 

没有找到任何反应堆的例子。

谢谢

Answer 1

我读了再次使用的文档，发现我的错误。通过return Flux.just(client);返回客户端没有意义，因为Flux立即终止触发客户端关闭。

我最终实现：

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

现在，所有的操作都在适当的顺序：

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

然后我用调用。