Resubscribe on Token Expiration

Before the recent rebase onto modern versions of Reactor Netty ([#984]), when
the Client received a 401 Unauthorized, it would invalidated any existing
token and then resubscribe with the same request causing a new token to be
negotiated the request to be retried, transparently to the user.

During the rebase, this functionality was lost (an easy mistake to make given
that there are no tests for it...) resulting in invalidation of the token, but
no re-subscription and therefore the failure was exposed to the user.

This change re-introduced the previous re-subscribe behavior.

Signed-off-by: Ben Hale <[email protected]>

Author: nebhale

.idea/compiler.xml

@@ -172,8 +172,8 @@ public <T> Mono<T> parseBodyToMono(Function<HttpClientResponseWithBody, Publishe
             return parseBodyToFlux(responseTransformer).singleOrEmpty();
         }
 
-        private static boolean isUnauthorized(HttpClientResponse response) {
-            return response.status() == HttpResponseStatus.UNAUTHORIZED;
+        private static boolean isUnauthorized(HttpClientResponseWithBody response) {
+            return response.getResponse().status() == HttpResponseStatus.UNAUTHORIZED;
         }
 
         private void attachChannelHandlers(HttpClientResponse response, Connection connection) {
@@ -186,18 +186,24 @@ private <T> Mono<T> deserialized(ByteBufFlux body, Class<T> bodyType) {
             return JsonCodec.decode(this.context.getConnectionContext().getObjectMapper(), body, bodyType);
         }
 
-        private HttpClientResponseWithBody invalidateToken(HttpClientResponseWithBody response) {
-            if (isUnauthorized(response.getResponse())) {
-                this.context.getTokenProvider().ifPresent(tokenProvider -> tokenProvider.invalidate(this.context.getConnectionContext()));
-            }
-            return response;
+        private Mono<HttpClientResponseWithBody> invalidateToken(Mono<HttpClientResponseWithBody> inbound) {
+            return inbound
+                .flatMap(response -> {
+                    if (isUnauthorized(response)) {
+                        this.context.getTokenProvider().ifPresent(tokenProvider -> tokenProvider.invalidate(this.context.getConnectionContext()));
+                        return inbound
+                            .transform(this::invalidateToken);
+                    } else {
+                        return Mono.just(response);
+                    }
+                });
         }
 
         private Mono<HttpClientResponseWithBody> processResponse(HttpClientResponse response, ByteBufFlux body) {
             HttpClientResponseWithBody responseWithBody = HttpClientResponseWithBody.of(body, response);
 
             return Mono.just(responseWithBody)
-                .map(this::invalidateToken)
+                .transform(this::invalidateToken)
                 .transform(this.context.getErrorPayloadMapper()
                     .orElse(ErrorPayloadMappers.fallback()));
         }