Ensure connections are closed in the event of a 401 & retry. (#1118)

Previously when a 401 would occur, we would throw the InvalidTokenException which in turn would trigger a retry. The retry would allow the TokenProvider to fetch a new, possibly valid token. When this happened, you could get into a state where reactor-netty was waiting for the library to finish using the connection (either consume the body or dispose it). Since we never did that, you could accumulate connections in the ESTABLISHED state. This code change will read and throw away the body when there's a 401, which signals to reactor-netty that it can reuse the connection.

Signed-off-by: Daniel Mikusa <[email protected]>

Author: Daniel Mikusa

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/Operator.java

@@ -245,6 +245,9 @@ private Flux<HttpClientResponseWithConnection> invalidateToken(Flux<HttpClientRe
                 .doOnNext(response -> {
                     if (isUnauthorized(response)) {
                         this.context.getTokenProvider().ifPresent(tokenProvider -> tokenProvider.invalidate(this.context.getConnectionContext()));
+                        // we don't need the body, but we need to consume the body so reactor-netty can reuse the connection
+                        //   if not, this will result in connections that don't close & the pool will fill up
+                        response.getConnection().inbound().receive().doOnNext(byteBuf -> {}).subscribe(byteBuf -> {}, ex -> {});
                         throw new InvalidTokenException();
                     }
                 });