javaspring-webfluxproject-reactorspring-webclientretrywhen

How to retry with delay based on response header


I've been learning Spring Webflux and reactive programming and have gotten stuck on a problem I'm trying to solve around retry logic using Spring Webclient. I've created a client and made successful calls to an external web-service GET endpoint that returns some JSON data.

When the external service responds with a "503 - Service Unavailable" status, the response includes a Retry-After header with a value that indicates how long I should wait before retrying the request. I want to find a way within Spring Webflux/Reactor to tell the webClient to retry it's request after X period, where X is the difference between now and the DateTime that I parse out of the response header.

Simple WebClient GET request

public <T> Mono<T> get(final String url, Class<T> clazz) {
    return webClient
            .get().uri(url)
            .retrieve()
            .bodyToMono(clazz);
}

WebClient Builder

I use a builder to create the webClient variable used in the above method, and it's stored as an instance variable in the class.

webClientBuilder = WebClient.builder();
webClientBuilder.codecs(clientCodecConfigurer -> {
    clientCodecConfigurer.defaultCodecs();
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonDecoder());
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonEncoder());
});
webClient = webClientBuilder.build();

Retry When

I've tried to understand and use the retryWhen method with the Retry class, but can't figure out if I can access or pass through the response header value there.

public <T> Mono<T> get(final String url, Class<T> clazz) {
    return webClient
            .get().uri(url)
            .retrieve()
            .bodyToMono(clazz);
            .retryWhen(new Retry() {
                @Override
                public Publisher<?> generateCompanion(final Flux<RetrySignal> retrySignals) {
                    // Can I use retrySignals or retryContext to find the response header somehow?
                    // If I can find the response header, how to return a "yes-retry" response?
                }
            })
}

Filter(s) with Extra Logic and DB Interaction

I've also tried to do some extra logic and use filters with the WebClient.Builder, but that only gets me to a point of halting a new request (call to #get) until a previously established Retry-After value has elapsed.

webClientBuilder = WebClient.builder();
webClientBuilder.codecs(clientCodecConfigurer -> {
    clientCodecConfigurer.defaultCodecs();
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonDecoder());
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonEncoder());
});
webClientBuilder.filter(ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
            final Clock clock = Clock.systemUTC();
            final int id = (int) clientRequest.attribute("id"); // id is saved as an attribute for the request, pull it out here
            final long retryAfterEpochMillis = // get epoch millisecond from DB for id
            if(epoch is in the past) {
                return Mono.just(clientRequest);
            } else { // have to wait until epoch passes to send request
                return Mono.just(clientRequest).delayElement(Duration.between(clock.instant(), Instant.ofEpochMilli(retryAfterEpochMillis)));
            }
        })
);
webClient = webClientBuilder.build();
.onStatus(HttpStatus::isError, response -> {
            final List<String> retryAfterHeaders = response.headers().header("Retry-After");
            if(retryAfterHeaders.size() > 0) {
                final long retryAfterEpochMillis = // parse millisecond epoch time from header
                // Save millisecond time to DB associated to specific id
            }
            return response.bodyToMono(String.class).flatMap(body ->
                    Mono.error(new RuntimeException(
                                    String.format("Request url {%s} failed with status {%s} and reason {%s}",
                                            url,
                                            response.rawStatusCode(),
                                            body))));
        })

Solution

  • 1. Retrieve header in retry builder

    public class WebClientStatefulRetry3 {
        public static void main(String[] args) {
            WebClient webClient = WebClient.create();
    
            call(webClient)
                    .retryWhen(Retry.indefinitely()
                            .filter(ex -> ex instanceof WebClientResponseException.ServiceUnavailable)
                            .doBeforeRetryAsync(signal -> Mono.delay(calculateDelay(signal.failure())).then()))
                    .block();
        }
    
        private static Mono<String> call(WebClient webClient) {
            return webClient.get()
                    .uri("http://mockbin.org/bin/b2a26614-0219-4018-9446-c03bc1868ebf")
                    .retrieve()
                    .bodyToMono(String.class);
        }
    
        private static Duration calculateDelay(Throwable failure) {
            String headerValue = ((WebClientResponseException.ServiceUnavailable) failure).getHeaders().get("Retry-After").get(0);
    
            return // calculate delay here from header and current time;
        }
    }
    

    2. Use expand operator to access the previous response and generate the next one

    public class WebClientRetryWithExpand {
        public static void main(String[] args) {
            WebClient webClient = WebClient.create();
    
            call(webClient)
                    .expand(prevResponse -> {
                        List<String> header = prevResponse.headers.header("Retry-After");
                        if (header.isEmpty()) {
                            return Mono.empty();
                        }
                        long delayInMillis = // calculate delay from header and current time
    
                        return Mono.delay(Duration.ofMillis(delayInMillis))
                                .then(call(webClient));
                    })
                    .last()
                    .block();
        }
    
        private static Mono<ResponseWithHeaders> call(WebClient webClient) {
            return webClient.get()
                    .uri("https://example.com")
                    .exchangeToMono(response -> response.bodyToMono(String.class)
                            .map(rawResponse -> new ResponseWithHeaders(rawResponse, response.headers())));
        }
    
        @Data
        static class ResponseWithHeaders {
            private final String rawResponse;
            private final ClientResponse.Headers headers;
        }
    }