ConnectionContext Cleanup

This change adds a dispose() method on the DefaultConnectionContext in order
to cleanup any cached resources like connection and thread pools. This method
is also marked as @PreDestroy so any DefaultConnectionContext created as a
Spring (or EJB) bean will be automatically disposed as part of the standard
application lifecycle.  This implementation is a temporary solution until
reactor/reactor-netty#88 is implemented and
available.

[resolves #745]

Author: nebhale

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/_DefaultConnectionContext.java

@@ -37,6 +37,7 @@
 import reactor.ipc.netty.resources.LoopResources;
 import reactor.ipc.netty.resources.PoolResources;
 
+import javax.annotation.PreDestroy;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
@@ -67,6 +68,15 @@ abstract class _DefaultConnectionContext implements ConnectionContext {
 
     private static final int UNDEFINED_PORT = -1;
 
+    /**
+     * Disposes resources created to service this connection context
+     */
+    @PreDestroy
+    public final void dispose() {
+        getConnectionPool().ifPresent(PoolResources::dispose);
+        getThreadPool().dispose();
+    }
+
     /**
      * The number of connections to use when processing requests and responses.  Setting this to `null` disables connection pooling.
      */
@@ -81,12 +91,12 @@ public Integer getConnectionPoolSize() {
     public HttpClient getHttpClient() {
         return HttpClient.create(options -> {
             options
-                .loopResources(LoopResources.create("cloudfoundry-client", getThreadPoolSize(), true))
+                .loopResources(getThreadPool())
                 .option(SO_SNDBUF, SEND_BUFFER_SIZE)
                 .option(SO_RCVBUF, RECEIVE_BUFFER_SIZE)
                 .disablePool();
 
-            Optional.ofNullable(getConnectionPoolSize()).ifPresent(connectionPoolSize -> options.poolResources(PoolResources.fixed("cloudfoundry-client", connectionPoolSize)));
+            getConnectionPool().ifPresent(options::poolResources);
             getKeepAlive().ifPresent(keepAlive -> options.option(SO_KEEPALIVE, keepAlive));
             getProxyConfiguration().ifPresent(c -> options.proxy(ClientOptions.Proxy.HTTP, c.getHost(), c.getPort().orElse(null), c.getUsername().orElse(null), u -> c.getPassword().orElse(null)));
             getConnectTimeout().ifPresent(socketTimeout -> options.option(CONNECT_TIMEOUT_MILLIS, (int) socketTimeout.toMillis()));
@@ -115,15 +125,6 @@ public Integer getPort() {
         return DEFAULT_PORT;
     }
 
-    @Override
-    public Mono<String> getRoot(String key) {
-        return getInfo()
-            .map(info -> normalize(UriComponentsBuilder.fromUriString(info.get(key)), getScheme()))
-            .doOnNext(components -> trust(components, getSslCertificateTruster()))
-            .map(UriComponents::toUriString)
-            .cache();
-    }
-
     @Value.Derived
     public Mono<String> getRoot() {
         Integer port = getPort();
@@ -138,6 +139,15 @@ public Mono<String> getRoot() {
         return Mono.just(components.toUriString());
     }
 
+    @Override
+    public Mono<String> getRoot(String key) {
+        return getInfo()
+            .map(info -> normalize(UriComponentsBuilder.fromUriString(info.get(key)), getScheme()))
+            .doOnNext(components -> trust(components, getSslCertificateTruster()))
+            .map(UriComponents::toUriString)
+            .cache();
+    }
+
     /**
      * The number of worker threads to use when processing requests and responses
      */
@@ -165,6 +175,12 @@ void checkForValidApiHost() {
      */
     abstract Optional<Duration> getConnectTimeout();
 
+    @Value.Derived
+    Optional<PoolResources> getConnectionPool() {
+        return Optional.ofNullable(getConnectionPoolSize())
+            .map(connectionPoolSize -> PoolResources.fixed("cloudfoundry-client", connectionPoolSize));
+    }
+
     @SuppressWarnings("unchecked")
     @Value.Derived
     Mono<Map<String, String>> getInfo() {
@@ -229,6 +245,11 @@ Optional<SslCertificateTruster> getSslCertificateTruster() {
      */
     abstract Optional<Duration> getSslHandshakeTimeout();
 
+    @Value.Derived
+    LoopResources getThreadPool() {
+        return LoopResources.create("cloudfoundry-client", getThreadPoolSize(), true);
+    }
+
     private static void trust(UriComponents components, Optional<SslCertificateTruster> sslCertificateTruster) {
         sslCertificateTruster.ifPresent(t -> t.trust(components.getHost(), components.getPort(), Duration.ofSeconds(30)));
     }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/_ReactorCloudFoundryClient.java

@@ -95,6 +95,8 @@
 import org.immutables.value.Value;
 import reactor.core.publisher.Mono;
 
+import javax.annotation.PostConstruct;
+
 /**
  * The Reactor-based implementation of {@link CloudFoundryClient}
  */
@@ -125,6 +127,7 @@ public Buildpacks buildpacks() {
         return new ReactorBuildpacks(getConnectionContext(), getRoot(), getTokenProvider());
     }
 
+    @PostConstruct
     public void checkCompatibility() {
         new CloudFoundryClientCompatibilityChecker(info()).check();
     }

integration-test/src/test/java/org/cloudfoundry/ExampleConfiguration.java

@@ -39,7 +39,7 @@
 @EnableAutoConfiguration
 public class ExampleConfiguration {
 
-    @Bean(initMethod = "checkCompatibility")
+    @Bean
     @Lazy
     ReactorCloudFoundryClient cloudFoundryClient(ConnectionContext connectionContext, TokenProvider tokenProvider) {
         return ReactorCloudFoundryClient.builder()

integration-test/src/test/java/org/cloudfoundry/IntegrationTestConfiguration.java

@@ -208,7 +208,7 @@ CloudFoundryCleaner cloudFoundryCleaner(@Qualifier("admin") CloudFoundryClient c
         return new CloudFoundryCleaner(cloudFoundryClient, nameFactory, uaaClient);
     }
 
-    @Bean(initMethod = "checkCompatibility")
+    @Bean
     ReactorCloudFoundryClient cloudFoundryClient(ConnectionContext connectionContext, TokenProvider tokenProvider) {
         return ReactorCloudFoundryClient.builder()
             .connectionContext(connectionContext)