Как повторно вызвать функцию ExchangeFilterFunction WebClient при повторной попытке - PullRequest
Новая
       33

Как повторно вызвать функцию ExchangeFilterFunction WebClient при повторной попытке

1 голос
/ 29 октября 2019

При использовании оператора retry(..) реактора WebClient функции обменного фильтра не запускаются при повторных попытках. Я понимаю, почему, но проблема в том, что функция (как показано ниже) генерирует токен аутентификации со временем истечения. Это может произойти, пока запрос «повторяется», токен истекает, потому что функция Exchange не вызывается повторно при повторной попытке. Есть ли способ, как заново сгенерировать токен для каждой повторной попытки?

После AuthClientExchangeFunction генерируется токен аутентификации (JWT) с истечением срока действия. 

public class AuthClientExchangeFunction implements ExchangeFilterFunction {


    private final TokenProvider tokenProvider;

    public IntraAuthWebClientExchangeFunction(TokenProvider tokenProvider) {
        this.tokenProvider = tokenProvider;
    }

    @Override
    public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
        String jwt = tokenProvider.getToken();
        return next.exchange(withBearer(request, jwt));

    }

    private ClientRequest withBearer(ClientRequest request, String jwt){
        return ClientRequest.from(request)
                            .headers(headers -> headers.set(HttpHeaders.AUTHORIZATION, "Bearer "+ jwt))
                            .build();
    }
}

Допустим, токен действителен в течение 2999 мс -> Каждый запрос повторной попытки не выполняется из-за 401. 

WebClient client = WebClient.builder()
                            .filter(new AuthClientExchangeFunction(tokenProvider))
                            .build();        

 client.get()
       .uri("/api")
       .retrieve()
       .bodyToMono(String.class)
       .retryBackoff(1, Duration.ofMillis(3000)) ;

Редактировать Вот исполняемый пример 

@SpringBootTest
@RunWith(SpringRunner.class)
public class RetryApplicationTests {


    private static final MockWebServer server  = new MockWebServer();

    private final RquestCountingFilterFunction requestCounter = new RquestCountingFilterFunction();

    @AfterClass
    public static void shutdown() throws IOException {
        server.shutdown();
    }

    @Test
    public void test() {

        server.enqueue(new MockResponse().setResponseCode(500).setBody("{}"));
        server.enqueue(new MockResponse().setResponseCode(500).setBody("{}"));
        server.enqueue(new MockResponse().setResponseCode(500).setBody("{}"));
        server.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));

        WebClient webClient = WebClient.builder()
                                       .baseUrl(server.url("/api").toString())
                                       .filter(requestCounter)
                                       .build();

        Mono<String> responseMono1 = webClient.get()
                                              .uri("/api")
                                              .retrieve()
                                              .bodyToMono(String.class)
                                              .retryBackoff(3, Duration.ofMillis(1000)) ;

        StepVerifier.create(responseMono1).expectNextCount(1).verifyComplete();

        assertThat(requestCounter.count()).isEqualTo(4);
    }



    static class RquestCountingFilterFunction implements ExchangeFilterFunction {

        final Logger log = LoggerFactory.getLogger(getClass());
        final AtomicInteger counter = new AtomicInteger();

        @Override
        public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
            log.info("Sending {} request to {} {}", counter.incrementAndGet(), request.method(), request.url());
            return next.exchange(request);
        }

        int count() {
            return counter.get();
        }
    }

}

output 

MockWebServer[44855] starting to accept connections
Sending 1 request to GET http://localhost:44855/api/api
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 500 Server Error
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 500 Server Error
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 500 Server Error
MockWebServer[44855] received request: GET /api/api HTTP/1.1 and responded: HTTP/1.1 200 OK

org.junit.ComparisonFailure: 
Expected :4
Actual   :1

1 Ответ

0 голосов
/ 29 октября 2019

Вам необходимо обновить версию весенней загрузки до 2.2.0.RELEASE. retry() не будет вызывать функцию обмена в предыдущей версии.

Я проверил это с помощью простого кода (In Kotlin). 

@Component
class AnswerPub {

    val webClient = WebClient.builder()
        .filter(PrintExchangeFunction())
        .baseUrl("https://jsonplaceholder.typicode.com").build()

    fun productInfo(): Mono<User> {
        return webClient
            .get()
            .uri("/todos2/1")
            .retrieve()
            .bodyToMono(User::class.java)
            .retry(2) { it is Exception }
    }

    data class User(
        val id: String,
        val userId: String,
        val title: String,
        val completed: Boolean
    )

}

class PrintExchangeFunction : ExchangeFilterFunction {
    override fun filter(request: ClientRequest, next: ExchangeFunction): Mono<ClientResponse> {
        println("Filtered")
        return next.exchange(request)
    }

}

И вывод консоли выглядел так: 

2019-10-29 09:31:55.912  INFO 12206 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port(s): 8080
2019-10-29 09:31:55.917  INFO 12206 --- [           main] c.e.s.SpringWfDemoApplicationKt          : Started SpringWfDemoApplicationKt in 3.19 seconds (JVM running for 4.234)
Filtered
Filtered
Filtered

Так что в моем случае функция exchange вызывается каждый раз.

...