Defer ExchangeFilterFunction to subscription time

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

Closes gh-23909

Author: rstoyanchev

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

@@ -317,7 +317,8 @@ public Mono<ClientResponse> exchange() {
 			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() {

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

@@ -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 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

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

@@ -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.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 void basic() {
 	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 void uriBuilder() {
 	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 void uriBuilderWithPathOverride() {
 	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 void defaultHeaderAndCookie() {
 				.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 void defaultHeaderAndCookieOverrides() {
 				.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 void defaultRequest() {
 
 		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 void withStringAttribute() {
 		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 void withNullAttribute() {
 		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 void apply() {
 						.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 void apply() {
 
 	@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);