Add ClientHttpRequestInterceptor in WebClient

This commit adds a new chain-based, interception contract to be used
with `WebClient`. This is the HTTP client equivalent of the `WebFilter`
contract already implemented in web reactive server.

A `ClientHttpRequestInterceptor` implementation can transform the
outgoing HTTP request (method, URI or headers) before delegating it to
the next interceptor in the chain, or bypass the request processing
altogether and return a (cached) HTTP response.

Issue: SPR-14502

spring-web/src/main/java/org/springframework/web/client/reactive/ClientHttpRequestInterceptionChain.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.client.reactive;
+
+import java.net.URI;
+import java.util.function.Consumer;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.HttpMessage;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+
+/**
+ * Delegate to the next {@link ClientHttpRequestInterceptor} in the chain.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+public interface ClientHttpRequestInterceptionChain {
+
+	/**
+	 * Delegate to the next {@link ClientHttpRequestInterceptor} in the chain.
+	 *
+	 * @param method the HTTP request method
+	 * @param uri the HTTP request URI
+	 * @param requestCallback a function that can customize the request
+	 * by changing the HTTP request headers with {@code HttpMessage.getHeaders()}.
+	 * @return a publisher of the resulting {@link ClientHttpResponse}
+	 */
+	Mono<ClientHttpResponse> intercept(HttpMethod method, URI uri, Consumer<? super HttpMessage> requestCallback);
+
+}

spring-web/src/main/java/org/springframework/web/client/reactive/ClientHttpRequestInterceptor.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.client.reactive;
+
+import java.net.URI;
+import java.util.List;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.HttpMethod;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+
+/**
+ * Contract for chain-based, interception processing of client http requests
+ * that may be used to implement cross-cutting requirements such
+ * as security, timeouts, caching, and others.
+ *
+ * <p>Implementations of this interface can be
+ * {@link WebClient#setInterceptors(List) registered} with the {@link WebClient}.
+ *
+ * @author Brian Clozel
+ * @see org.springframework.web.client.reactive.WebClient
+ * @since 5.0
+ */
+@FunctionalInterface
+public interface ClientHttpRequestInterceptor {
+
+	/**
+	 * Intercept the client HTTP request
+	 *
+	 * <p>The provided {@link ClientHttpRequestInterceptionChain}
+	 * instance allows the interceptor to delegate the request
+	 * to the next interceptor in the chain.
+	 *
+	 * <p>An implementation might follow this pattern:
+	 * <ol>
+	 * <li>Examine the {@link HttpMethod method} and {@link URI uri}</li>
+	 * <li>Optionally change those when delegating to the next interceptor
+	 * with the {@code ClientHttpRequestInterceptionChain}.</li>
+	 * <li>Optionally transform the HTTP message given as an
+	 * argument of the request callback in
+	 * {@code chain.intercept(method, uri, requestCallback)}.</li>
+	 * <li>Optionally transform the response before returning it.</li>
+	 * </ol>
+	 *
+	 * @param method the HTTP request method
+	 * @param uri the HTTP request URI
+	 * @param chain the request interception chain
+	 * @return a publisher of the {@link ClientHttpResponse}
+	 */
+	Mono<ClientHttpResponse> intercept(HttpMethod method, URI uri, ClientHttpRequestInterceptionChain chain);
+}

spring-web/src/main/java/org/springframework/web/client/reactive/WebClient.java

@@ -16,6 +16,7 @@
 
 package org.springframework.web.client.reactive;
 
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -33,6 +34,8 @@ import org.springframework.core.codec.ByteBufferEncoder;
 import org.springframework.core.codec.CharSequenceEncoder;
 import org.springframework.core.codec.ResourceDecoder;
 import org.springframework.core.codec.StringDecoder;
+import org.springframework.http.HttpMessage;
+import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.client.reactive.ClientHttpConnector;
@@ -47,6 +50,7 @@ import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.http.codec.xml.Jaxb2XmlDecoder;
 import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
+import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -94,6 +98,8 @@ public final class WebClient {
 
 	private ClientHttpConnector clientHttpConnector;
 
+	private List<ClientHttpRequestInterceptor> interceptors;
+
 	private final DefaultWebClientConfig webClientConfig;
 
 
@@ -172,6 +178,15 @@ public final class WebClient {
 		this.webClientConfig.setResponseErrorHandler(responseErrorHandler);
 	}
 
+	/**
+	 * Set the list of {@link ClientHttpRequestInterceptor} to use
+	 * for intercepting client HTTP requests
+	 */
+	public void setInterceptors(List<ClientHttpRequestInterceptor> interceptors) {
+		this.interceptors = (interceptors != null ?
+				Collections.unmodifiableList(interceptors) : Collections.emptyList());
+	}
+
 	/**
 	 * Perform the actual HTTP request/response exchange
 	 *
@@ -186,10 +201,12 @@ public final class WebClient {
 	public WebResponseActions perform(ClientWebRequestBuilder builder) {
 
 		ClientWebRequest clientWebRequest = builder.build();
+		DefaultClientHttpRequestInterceptionChain interception =
+				new DefaultClientHttpRequestInterceptionChain(this.clientHttpConnector,
+						this.interceptors, clientWebRequest);
 
-		final Mono<ClientHttpResponse> clientResponse = this.clientHttpConnector
-				.connect(clientWebRequest.getMethod(), clientWebRequest.getUrl(),
-						new DefaultRequestCallback(clientWebRequest))
+		final Mono<ClientHttpResponse> clientResponse = interception
+				.intercept(clientWebRequest.getMethod(), clientWebRequest.getUrl(), null)
 				.log("org.springframework.web.client.reactive", Level.FINE);
 
 		return new WebResponseActions() {
@@ -253,12 +270,15 @@ public final class WebClient {
 
 		private final ClientWebRequest clientWebRequest;
 
+		private final List<Consumer<? super HttpMessage>> requestCustomizers;
+
 
-		public DefaultRequestCallback(ClientWebRequest clientWebRequest) {
+		public DefaultRequestCallback(ClientWebRequest clientWebRequest,
+				List<Consumer<? super HttpMessage>> requestCustomizers) {
 			this.clientWebRequest = clientWebRequest;
+			this.requestCustomizers = requestCustomizers;
 		}
 
-
 		@Override
 		public Mono<Void> apply(ClientHttpRequest clientHttpRequest) {
 			clientHttpRequest.getHeaders().putAll(this.clientWebRequest.getHttpHeaders());
@@ -269,6 +289,9 @@ public final class WebClient {
 			this.clientWebRequest.getCookies().values()
 					.stream().flatMap(cookies -> cookies.stream())
 					.forEach(cookie -> clientHttpRequest.getCookies().add(cookie.getName(), cookie));
+
+			this.requestCustomizers.forEach(customizer -> customizer.accept(clientHttpRequest));
+
 			if (this.clientWebRequest.getBody() != null) {
 				return writeRequestBody(this.clientWebRequest.getBody(),
 						this.clientWebRequest.getElementType(),
@@ -279,7 +302,7 @@ public final class WebClient {
 			}
 		}
 
-		@SuppressWarnings({ "unchecked", "rawtypes" })
+		@SuppressWarnings({"unchecked", "rawtypes"})
 		protected Mono<Void> writeRequestBody(Publisher<?> content,
 				ResolvableType requestType, ClientHttpRequest request,
 				List<HttpMessageWriter<?>> messageWriters) {
@@ -301,4 +324,47 @@ public final class WebClient {
 		}
 	}
 
+	protected class DefaultClientHttpRequestInterceptionChain implements ClientHttpRequestInterceptionChain {
+
+		private final ClientHttpConnector connector;
+
+		private final List<ClientHttpRequestInterceptor> interceptors;
+
+		private final ClientWebRequest clientWebRequest;
+
+		private final List<Consumer<? super HttpMessage>> requestCustomizers;
+
+		private int index;
+
+		public DefaultClientHttpRequestInterceptionChain(ClientHttpConnector connector,
+				List<ClientHttpRequestInterceptor> interceptors,
+				ClientWebRequest clientWebRequest) {
+
+			Assert.notNull(connector, "'connector' should not be null");
+			this.connector = connector;
+			this.interceptors = interceptors;
+			this.clientWebRequest = clientWebRequest;
+			this.requestCustomizers = new ArrayList<>();
+			this.index = 0;
+		}
+
+		@Override
+		public Mono<ClientHttpResponse> intercept(HttpMethod method, URI uri,
+				Consumer<? super HttpMessage> requestCustomizer) {
+
+			if (requestCustomizer != null) {
+				this.requestCustomizers.add(requestCustomizer);
+			}
+			if (this.interceptors != null && this.index < this.interceptors.size()) {
+				ClientHttpRequestInterceptor interceptor = this.interceptors.get(this.index++);
+				return interceptor.intercept(method, uri, this);
+			}
+			else {
+				return this.connector.connect(method, uri,
+						new DefaultRequestCallback(this.clientWebRequest, this.requestCustomizers));
+			}
+		}
+
+	}
+
 }

spring-web/src/test/java/org/springframework/web/client/reactive/ClientHttpRequestInterceptorTests.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright 2002-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.client.reactive;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+import static org.springframework.web.client.reactive.ClientWebRequestBuilders.*;
+import static org.springframework.web.client.reactive.ResponseExtractors.*;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.client.reactive.ClientHttpConnector;
+import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.tests.TestSubscriber;
+import org.springframework.web.client.reactive.test.MockClientHttpRequest;
+import org.springframework.web.client.reactive.test.MockClientHttpResponse;
+
+/**
+ * @author Brian Clozel
+ */
+public class ClientHttpRequestInterceptorTests {
+
+	private MockClientHttpRequest mockRequest;
+
+	private MockClientHttpResponse mockResponse;
+
+	private MockClientHttpConnector mockClientHttpConnector;
+
+	private WebClient webClient;
+
+
+	@Before
+	public void setUp() throws Exception {
+		this.mockClientHttpConnector = new MockClientHttpConnector();
+		this.webClient = new WebClient(this.mockClientHttpConnector);
+		this.mockResponse = new MockClientHttpResponse();
+		this.mockResponse.setStatus(HttpStatus.OK);
+		this.mockResponse.getHeaders().setContentType(MediaType.TEXT_PLAIN);
+		this.mockResponse.setBody("Spring Framework");
+	}
+
+	@Test
+	public void shouldExecuteInterceptors() throws Exception {
+		List<ClientHttpRequestInterceptor> interceptors = new ArrayList<>();
+		interceptors.add(new NoOpInterceptor());
+		interceptors.add(new NoOpInterceptor());
+		interceptors.add(new NoOpInterceptor());
+		this.webClient.setInterceptors(interceptors);
+
+		Mono<String> result = this.webClient.perform(get("http://example.org/resource"))
+				.extract(body(String.class));
+
+		TestSubscriber.subscribe(result)
+				.assertNoError()
+				.assertValues("Spring Framework")
+				.assertComplete();
+		interceptors.stream().forEach(interceptor -> {
+			Assert.assertTrue(((NoOpInterceptor) interceptor).invoked);
+		});
+	}
+
+	@Test
+	public void shouldChangeRequest() throws Exception {
+		ClientHttpRequestInterceptor interceptor = new ClientHttpRequestInterceptor() {
+			@Override
+			public Mono<ClientHttpResponse> intercept(HttpMethod method, URI uri,
+					ClientHttpRequestInterceptionChain interception) {
+
+				return interception.intercept(HttpMethod.POST, URI.create("http://example.org/other"),
+						(request) -> {
+							request.getHeaders().set("X-Custom", "Spring Framework");
+						});
+			}
+		};
+		this.webClient.setInterceptors(Collections.singletonList(interceptor));
+
+		Mono<String> result = this.webClient.perform(get("http://example.org/resource"))
+				.extract(body(String.class));
+
+		TestSubscriber.subscribe(result)
+				.assertNoError()
+				.assertValues("Spring Framework")
+				.assertComplete();
+
+		assertThat(this.mockRequest.getMethod(), is(HttpMethod.POST));
+		assertThat(this.mockRequest.getURI().toString(), is("http://example.org/other"));
+		assertThat(this.mockRequest.getHeaders().getFirst("X-Custom"), is("Spring Framework"));
+	}
+
+	@Test
+	public void shouldShortCircuitConnector() throws Exception {
+
+		MockClientHttpResponse otherResponse = new MockClientHttpResponse();
+		otherResponse.setStatus(HttpStatus.OK);
+		otherResponse.setBody("Other content");
+
+		List<ClientHttpRequestInterceptor> interceptors = new ArrayList<>();
+		interceptors.add((method, uri, interception) -> Mono.just(otherResponse));
+		interceptors.add(new NoOpInterceptor());
+		this.webClient.setInterceptors(interceptors);
+
+		Mono<String> result = this.webClient.perform(get("http://example.org/resource"))
+				.extract(body(String.class));
+
+		TestSubscriber.subscribe(result)
+				.assertNoError()
+				.assertValues("Other content")
+				.assertComplete();
+
+		assertFalse(((NoOpInterceptor) interceptors.get(1)).invoked);
+	}
+
+	private class MockClientHttpConnector implements ClientHttpConnector {
+
+		@Override
+		public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
+				Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+
+			mockRequest = new MockClientHttpRequest(method, uri);
+			return requestCallback.apply(mockRequest).then(Mono.just(mockResponse));
+		}
+	}
+
+
+	private static class NoOpInterceptor implements ClientHttpRequestInterceptor {
+
+		public boolean invoked = false;
+
+		@Override
+		public Mono<ClientHttpResponse> intercept(HttpMethod method, URI uri,
+				ClientHttpRequestInterceptionChain interception) {
+
+			this.invoked = true;
+			return interception.intercept(method, uri, (request) -> { });
+		}
+	}
+}