Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Defer ExchangeFilterFunction to subscription time

Browse Source 
Previously fixed in 5.2 via d46359. Now also backported to 5.1.x.

Closes gh-23909
This commit is contained in:
Rossen Stoyanchev
2019-11-06 21:31:42 +00:00
parent ca3440cb42
commit 1a057654b2
3 changed files with 54 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java
View File

@@ -317,7 +317,8 @@ class DefaultWebClient implements WebClient {
ClientRequest request = (this.inserter != null ?
initRequestBuilder().body(this.inserter).build() :
initRequestBuilder().build());
return exchangeFunction.exchange(request).switchIfEmpty(NO_HTTP_CLIENT_RESPONSE_ERROR);
return Mono.defer(() -> exchangeFunction.exchange(request))
.switchIfEmpty(NO_HTTP_CLIENT_RESPONSE_ERROR);
}
private ClientRequest.Builder initRequestBuilder() {

6
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ExchangeFilterFunction.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,7 +23,9 @@ import reactor.core.publisher.Mono;
import org.springframework.util.Assert;
/**
* Represents a function that filters an{@linkplain ExchangeFunction exchange function}.
* Represents a function that filters an {@linkplain ExchangeFunction exchange function}.
* <p>The filter is executed when a {@code Subscriber} subscribes to the
* {@code Publisher} returned by the {@code WebClient}.
*
* @author Arjen Poutsma
* @since 5.0

62
spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -34,13 +34,20 @@ import org.springframework.core.NamedThreadLocal;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.verifyZeroInteractions;
import static org.mockito.Mockito.when;
/**
* Unit tests for {@link DefaultWebClient}.
*
* @author Rossen Stoyanchev
* @author Brian Clozel
*/
public class DefaultWebClientTests {
@@ -56,14 +63,15 @@ public class DefaultWebClientTests {
public void setup() {
MockitoAnnotations.initMocks(this);
this.exchangeFunction = mock(ExchangeFunction.class);
when(this.exchangeFunction.exchange(this.captor.capture())).thenReturn(Mono.empty());
ClientResponse mockResponse = mock(ClientResponse.class);
when(this.exchangeFunction.exchange(this.captor.capture())).thenReturn(Mono.just(mockResponse));
this.builder = WebClient.builder().baseUrl("/base").exchangeFunction(this.exchangeFunction);
}
@Test
public void basic() {
this.builder.build().get().uri("/path").exchange();
this.builder.build().get().uri("/path").exchange().block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("/base/path", request.url().toString());
@@ -75,7 +83,8 @@ public class DefaultWebClientTests {
public void uriBuilder() {
this.builder.build().get()
.uri(builder -> builder.path("/path").queryParam("q", "12").build())
.exchange();
.exchange()
.block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("/base/path?q=12", request.url().toString());
@@ -86,7 +95,8 @@ public class DefaultWebClientTests {
public void uriBuilderWithPathOverride() {
this.builder.build().get()
.uri(builder -> builder.replacePath("/path").build())
.exchange();
.exchange()
.block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("/path", request.url().toString());
@@ -97,7 +107,8 @@ public class DefaultWebClientTests {
public void requestHeaderAndCookie() {
this.builder.build().get().uri("/path").accept(MediaType.APPLICATION_JSON)
.cookies(cookies -> cookies.add("id", "123")) // SPR-16178
.exchange();
.exchange()
.block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("application/json", request.headers().getFirst("Accept"));
@@ -111,7 +122,7 @@ public class DefaultWebClientTests {
.defaultHeader("Accept", "application/json").defaultCookie("id", "123")
.build();
client.get().uri("/path").exchange();
client.get().uri("/path").exchange().block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("application/json", request.headers().getFirst("Accept"));
@@ -126,7 +137,8 @@ public class DefaultWebClientTests {
.defaultCookie("id", "123")
.build();
client.get().uri("/path").header("Accept", "application/xml").cookie("id", "456").exchange();
client.get().uri("/path").header("Accept", "application/xml").cookie("id", "456")
.exchange().block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("application/xml", request.headers().getFirst("Accept"));
@@ -151,7 +163,7 @@ public class DefaultWebClientTests {
try {
context.set("bar");
client.get().uri("/path").attribute("foo", "bar").exchange();
client.get().uri("/path").attribute("foo", "bar").exchange().block(Duration.ofSeconds(10));
}
finally {
context.remove();
@@ -219,7 +231,8 @@ public class DefaultWebClientTests {
this.builder.filter(filter).build()
.get().uri("/path")
.attribute("foo", "bar")
.exchange();
.exchange()
.block(Duration.ofSeconds(10));
assertEquals("bar", actual.get("foo"));
@@ -238,7 +251,8 @@ public class DefaultWebClientTests {
this.builder.filter(filter).build()
.get().uri("/path")
.attribute("foo", null)
.exchange();
.exchange()
.block(Duration.ofSeconds(10));
assertNull(actual.get("foo"));
@@ -254,7 +268,7 @@ public class DefaultWebClientTests {
.defaultCookie("id", "123"))
.build();
client.get().uri("/path").exchange();
client.get().uri("/path").exchange().block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("application/json", request.headers().getFirst("Accept"));
@@ -264,11 +278,31 @@ public class DefaultWebClientTests {
@Test
public void switchToErrorOnEmptyClientResponseMono() {
ExchangeFunction exchangeFunction = mock(ExchangeFunction.class);
when(exchangeFunction.exchange(any())).thenReturn(Mono.empty());
WebClient.Builder builder = WebClient.builder().baseUrl("/base").exchangeFunction(exchangeFunction);
StepVerifier.create(builder.build().get().uri("/path").exchange())
.expectErrorMessage("The underlying HTTP client completed without emitting a response.")
.verify(Duration.ofSeconds(5));
}
@Test // gh-23909
public void shouldApplyFiltersAtSubscription() {
WebClient client = this.builder
.filter((request, next) ->
next.exchange(ClientRequest
.from(request)
.header("Custom", "value")
.build()))
.build();
Mono<ClientResponse> exchange = client.get().uri("/path").exchange();
verifyZeroInteractions(this.exchangeFunction);
exchange.block(Duration.ofSeconds(10));
ClientRequest request = verifyAndGetRequest();
assertEquals("value", request.headers().getFirst("Custom"));
}
private ClientRequest verifyAndGetRequest() {
ClientRequest request = this.captor.getValue();
Mockito.verify(this.exchangeFunction).exchange(request);