如何正確管理反應堆中的可關閉資源

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

然後我用調用。