Align Reactive WebClient with web.reactive.function

This commit refactors the web client to be more similar to
web.reactive.function. Changes include:

- Refactor ClientWebRequest to immutable ClientRequest with builder and
   support for BodyInserters.
- Introduce ClientResponse which exposes headers, status, and support
   for reading from the body with BodyExtractors.
- Removed ResponseErrorHandler, in favor of having a ClientResponse
   with "error" status code (i.e. 4xx or 5xx). Also removed
   WebClientException and subclasses.
- Refactored WebClientConfig to WebClientStrategies.
- Refactored ClientHttpRequestInterceptor to ExchangeFilterFunction.
- Removed ClientWebRequestPostProcessor in favor of
   ExchangeFilterFunction, which allows for asynchronous execution.

Issue: SPR-14827

spring-web/src/test/java/org/springframework/http/server/reactive/FlushingIntegrationTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.http.server.reactive;
 
+import java.time.Duration;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
@@ -24,13 +26,11 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.http.codec.BodyExtractors;
 import org.springframework.tests.TestSubscriber;
-import org.springframework.web.client.reactive.ClientWebRequestBuilders;
-import org.springframework.web.client.reactive.ResponseExtractors;
+import org.springframework.web.client.reactive.ClientRequest;
 import org.springframework.web.client.reactive.WebClient;
 
-import java.time.Duration;
-
 /**
  * @author Sebastien Deleuze
  */
@@ -41,17 +41,19 @@ public class FlushingIntegrationTests extends AbstractHttpHandlerIntegrationTest
 	@Before
 	public void setup() throws Exception {
 		super.setup();
-		this.webClient = new WebClient(new ReactorClientHttpConnector());
+		this.webClient = WebClient.create(new ReactorClientHttpConnector());
 	}
 
 	@Test
 	public void testFlushing() throws Exception {
+
+		ClientRequest<Void> request = ClientRequest.GET("http://localhost:" + port).build();
+
+
 		Mono<String> result = this.webClient
-				.perform(ClientWebRequestBuilders.get("http://localhost:" + port))
-				.extract(ResponseExtractors.bodyStream(String.class))
-				.takeUntil(s -> {
-					return s.endsWith("data1");
-				})
+				.exchange(request)
+				.flatMap(response -> response.body(BodyExtractors.toFlux(String.class)))
+				.takeUntil(s -> s.endsWith("data1"))
 				.reduce((s1, s2) -> s1 + s2);
 
 		TestSubscriber

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

@@ -1,165 +0,0 @@
-/*
- * 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) -> { });
-		}
-	}
-}

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

@@ -1,71 +0,0 @@
-/*
- * 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.nio.charset.Charset;
-import java.util.Base64;
-import java.util.Base64.Encoder;
-
-import org.junit.Test;
-import org.springframework.http.HttpHeaders;
-
-
-import static org.junit.Assert.*;
-import static org.springframework.web.client.reactive.ClientWebRequestPostProcessors.*;
-import static org.springframework.web.client.reactive.ClientWebRequestBuilders.*;
-
-/**
- *
- * @author Rob Winch
- * @since 5.0
- */
-public class ClientWebRequestPostProcessorsTests {
-
-	@Test
-	public void httpBasicWhenUsernamePasswordThenHeaderSet() {
-		ClientWebRequest request = get("/").apply(httpBasic("user", "password")).build();
-		assertEquals(request.getHttpHeaders().getFirst(HttpHeaders.AUTHORIZATION), basic("user:password"));
-	}
-
-	@Test
-	public void httpBasicWhenUsernameEmptyThenHeaderSet() {
-		ClientWebRequest request = get("/").apply(httpBasic("", "password")).build();
-		assertEquals(request.getHttpHeaders().getFirst(HttpHeaders.AUTHORIZATION), basic(":password"));
-	}
-
-	@Test
-	public void httpBasicWhenPasswordEmptyThenHeaderSet() {
-		ClientWebRequest request = get("/").apply(httpBasic("user", "")).build();
-		assertEquals(request.getHttpHeaders().getFirst(HttpHeaders.AUTHORIZATION), basic("user:"));
-	}
-
-	@Test(expected = IllegalArgumentException.class)
-	public void httpBasicWhenUsernameNullThenIllegalArgumentException() {
-		httpBasic(null, "password");
-	}
-
-	@Test(expected = IllegalArgumentException.class)
-	public void httpBasicWhenPasswordNullThenIllegalArgumentException() {
-		httpBasic("username", null);
-	}
-
-	private static String basic(String string) {
-		Encoder encoder = Base64.getEncoder();
-		byte[] bytes = string.getBytes(Charset.defaultCharset());
-		return "Basic " + encoder.encodeToString(bytes);
-	}
-}

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

@@ -0,0 +1,215 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.codec.CharSequenceEncoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
+import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.http.codec.BodyInserter;
+import org.springframework.http.codec.EncoderHttpMessageWriter;
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.web.client.reactive.test.MockClientHttpRequest;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class DefaultClientRequestBuilderTests {
+
+	@Test
+	public void from() throws Exception {
+		ClientRequest<Void> other = ClientRequest.GET("http://example.com")
+				.header("foo", "bar")
+				.cookie("baz", "qux").build();
+		ClientRequest<Void> result = ClientRequest.from(other).build();
+		assertEquals(new URI("http://example.com"), result.url());
+		assertEquals(HttpMethod.GET, result.method());
+		assertEquals("bar", result.headers().getFirst("foo"));
+		assertEquals("qux", result.cookies().getFirst("baz"));
+	}
+
+	@Test
+	public void method() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.method(HttpMethod.DELETE, url).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.DELETE, result.method());
+	}
+
+	@Test
+	public void GET() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.GET(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.GET, result.method());
+	}
+
+	@Test
+	public void HEAD() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.HEAD(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.HEAD, result.method());
+	}
+
+	@Test
+	public void POST() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.POST(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.POST, result.method());
+	}
+
+	@Test
+	public void PUT() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.PUT(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.PUT, result.method());
+	}
+
+	@Test
+	public void PATCH() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.PATCH(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.PATCH, result.method());
+	}
+
+	@Test
+	public void DELETE() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.DELETE(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.DELETE, result.method());
+	}
+
+	@Test
+	public void OPTIONS() throws Exception {
+		URI url = new URI("http://example.com");
+		ClientRequest<Void> result = ClientRequest.OPTIONS(url.toString()).build();
+		assertEquals(url, result.url());
+		assertEquals(HttpMethod.OPTIONS, result.method());
+	}
+
+	@Test
+	public void accept() throws Exception {
+		MediaType json = MediaType.APPLICATION_JSON;
+		ClientRequest<Void> result = ClientRequest.GET("http://example.com").accept(json).build();
+		assertEquals(Collections.singletonList(json), result.headers().getAccept());
+	}
+
+	@Test
+	public void acceptCharset() throws Exception {
+		Charset charset = Charset.defaultCharset();
+		ClientRequest<Void> result = ClientRequest.GET("http://example.com")
+				.acceptCharset(charset).build();
+		assertEquals(Collections.singletonList(charset), result.headers().getAcceptCharset());
+	}
+
+	@Test
+	public void ifModifiedSince() throws Exception {
+		ZonedDateTime now = ZonedDateTime.now();
+		ClientRequest<Void> result = ClientRequest.GET("http://example.com")
+				.ifModifiedSince(now).build();
+		assertEquals(now.toInstant().toEpochMilli()/1000, result.headers().getIfModifiedSince()/1000);
+	}
+
+	@Test
+	public void ifNoneMatch() throws Exception {
+		ClientRequest<Void> result = ClientRequest.GET("http://example.com")
+				.ifNoneMatch("\"v2.7\"", "\"v2.8\"").build();
+		assertEquals(Arrays.asList("\"v2.7\"", "\"v2.8\""), result.headers().getIfNoneMatch());
+	}
+
+	@Test
+	public void cookie() throws Exception {
+		ClientRequest<Void> result = ClientRequest.GET("http://example.com")
+				.cookie("foo", "bar").build();
+		assertEquals("bar", result.cookies().getFirst("foo"));
+	}
+
+	@Test
+	public void build() throws Exception {
+		ClientRequest<Void> result = ClientRequest.GET("http://example.com")
+				.header("MyKey", "MyValue")
+				.cookie("foo", "bar")
+				.build();
+
+		MockClientHttpRequest request = new MockClientHttpRequest();
+		WebClientStrategies strategies = mock(WebClientStrategies.class);
+
+		result.writeTo(request, strategies).block();
+
+		assertEquals("MyValue", request.getHeaders().getFirst("MyKey"));
+		assertEquals("bar", request.getCookies().getFirst("foo").getValue());
+		assertNull(request.getBody());
+	}
+
+	@Test
+	public void bodyInserter() throws Exception {
+		String body = "foo";
+		Supplier<String> supplier = () -> body;
+		BiFunction<ClientHttpRequest, BodyInserter.Context, Mono<Void>> writer =
+				(response, strategies) -> {
+					byte[] bodyBytes = body.getBytes(UTF_8);
+					ByteBuffer byteBuffer = ByteBuffer.wrap(bodyBytes);
+					DataBuffer buffer = new DefaultDataBufferFactory().wrap(byteBuffer);
+
+					return response.writeWith(Mono.just(buffer));
+				};
+
+		ClientRequest<String> result = ClientRequest.POST("http://example.com")
+				.body(BodyInserter.of(writer, supplier));
+		assertEquals(body, result.body());
+
+		MockClientHttpRequest request = new MockClientHttpRequest();
+
+		List<HttpMessageWriter<?>> messageWriters = new ArrayList<>();
+		messageWriters.add(new EncoderHttpMessageWriter<CharSequence>(new CharSequenceEncoder()));
+
+		WebClientStrategies strategies = mock(WebClientStrategies.class);
+		when(strategies.messageWriters()).thenReturn(messageWriters::stream);
+
+		result.writeTo(request, strategies).block();
+		assertNotNull(request.getBody());
+	}
+
+}

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

@@ -0,0 +1,119 @@
+/*
+ * 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.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.Set;
+
+import org.junit.Before;
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.codec.StringDecoder;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpRange;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.codec.DecoderHttpMessageReader;
+import org.springframework.http.codec.HttpMessageReader;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.springframework.http.codec.BodyExtractors.toMono;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class DefaultClientResponseTests {
+
+	private ClientHttpResponse mockResponse;
+
+	private WebClientStrategies mockWebClientStrategies;
+
+	private DefaultClientResponse defaultClientResponse;
+
+
+	@Before
+	public void createMocks() {
+		mockResponse = mock(ClientHttpResponse.class);
+		mockWebClientStrategies = mock(WebClientStrategies.class);
+
+		defaultClientResponse = new DefaultClientResponse(mockResponse, mockWebClientStrategies);
+	}
+
+	@Test
+	public void statusCode() throws Exception {
+		HttpStatus status = HttpStatus.CONTINUE;
+		when(mockResponse.getStatusCode()).thenReturn(status);
+
+		assertEquals(status, defaultClientResponse.statusCode());
+	}
+
+	@Test
+	public void header() throws Exception {
+		HttpHeaders httpHeaders = new HttpHeaders();
+		long contentLength = 42L;
+		httpHeaders.setContentLength(contentLength);
+		MediaType contentType = MediaType.TEXT_PLAIN;
+		httpHeaders.setContentType(contentType);
+		InetSocketAddress host = InetSocketAddress.createUnresolved("localhost", 80);
+		httpHeaders.setHost(host);
+		List<HttpRange> range = Collections.singletonList(HttpRange.createByteRange(0, 42));
+		httpHeaders.setRange(range);
+
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+
+		ClientResponse.Headers headers = defaultClientResponse.headers();
+		assertEquals(OptionalLong.of(contentLength), headers.contentLength());
+		assertEquals(Optional.of(contentType), headers.contentType());
+		assertEquals(httpHeaders, headers.asHttpHeaders());
+	}
+
+	@Test
+	public void body() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		Set<HttpMessageReader<?>> messageReaders = Collections
+				.singleton(new DecoderHttpMessageReader<String>(new StringDecoder()));
+		when(mockWebClientStrategies.messageReaders()).thenReturn(messageReaders::stream);
+
+		Mono<String> resultMono = defaultClientResponse.body(toMono(String.class));
+		assertEquals("foo", resultMono.block());
+	}
+
+
+}

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

@@ -1,99 +0,0 @@
-package org.springframework.web.client.reactive;
-
-import java.util.Collections;
-import java.util.List;
-
-import org.junit.Before;
-import org.junit.Test;
-import reactor.core.publisher.Flux;
-
-import org.springframework.core.codec.StringDecoder;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DefaultDataBufferFactory;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.MediaType;
-import org.springframework.http.client.reactive.ClientHttpResponse;
-import org.springframework.http.codec.DecoderHttpMessageReader;
-import org.springframework.http.codec.HttpMessageReader;
-import org.springframework.tests.TestSubscriber;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
-import static org.springframework.web.client.reactive.ResponseExtractors.as;
-
-/**
- * Unit tests for {@link DefaultResponseErrorHandler}.
- *
- * @author Brian Clozel
- */
-public class DefaultResponseErrorHandlerTests {
-
-	private DefaultResponseErrorHandler errorHandler;
-
-	private ClientHttpResponse response;
-
-	private List<HttpMessageReader<?>> messageReaders;
-
-	@Before
-	public void setUp() throws Exception {
-		this.errorHandler = new DefaultResponseErrorHandler();
-		this.response = mock(ClientHttpResponse.class);
-		this.messageReaders = Collections
-				.singletonList(new DecoderHttpMessageReader<>(new StringDecoder()));
-	}
-
-	@Test
-	public void noError() throws Exception {
-		given(this.response.getStatusCode()).willReturn(HttpStatus.OK);
-		this.errorHandler.handleError(this.response, this.messageReaders);
-	}
-
-	@Test
-	public void clientError() throws Exception {
-		HttpHeaders headers = new HttpHeaders();
-		headers.setContentType(MediaType.TEXT_PLAIN);
-		DataBuffer buffer = new DefaultDataBufferFactory().allocateBuffer();
-		buffer.write(new String("Page Not Found").getBytes("UTF-8"));
-		given(this.response.getStatusCode()).willReturn(HttpStatus.NOT_FOUND);
-		given(this.response.getHeaders()).willReturn(headers);
-		given(this.response.getBody()).willReturn(Flux.just(buffer));
-		try {
-			this.errorHandler.handleError(this.response, this.messageReaders);
-			fail("expected HttpClientErrorException");
-		}
-		catch (WebClientErrorException exc) {
-			assertThat(exc.getMessage(), is("404 Not Found"));
-			assertThat(exc.getStatus(), is(HttpStatus.NOT_FOUND));
-			TestSubscriber.subscribe(exc.getResponseBody(as(String.class)))
-					.awaitAndAssertNextValues("Page Not Found")
-					.assertComplete();
-		}
-	}
-
-	@Test
-	public void serverError() throws Exception {
-		HttpHeaders headers = new HttpHeaders();
-		headers.setContentType(MediaType.TEXT_PLAIN);
-		DataBuffer buffer = new DefaultDataBufferFactory().allocateBuffer();
-		buffer.write(new String("Internal Server Error").getBytes("UTF-8"));
-		given(this.response.getStatusCode()).willReturn(HttpStatus.INTERNAL_SERVER_ERROR);
-		given(this.response.getHeaders()).willReturn(headers);
-		given(this.response.getBody()).willReturn(Flux.just(buffer));
-		try {
-			this.errorHandler.handleError(this.response, this.messageReaders);
-			fail("expected HttpServerErrorException");
-		}
-		catch (WebServerErrorException exc) {
-			assertThat(exc.getMessage(), is("500 Internal Server Error"));
-			assertThat(exc.getStatus(), is(HttpStatus.INTERNAL_SERVER_ERROR));
-			TestSubscriber.subscribe(exc.getResponseBody(as(String.class)))
-					.awaitAndAssertNextValues("Internal Server Error")
-					.assertComplete();
-		}
-	}
-
-}

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

@@ -1,60 +0,0 @@
-/*
- * 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.mockito.Matchers.any;
-import static org.mockito.Mockito.*;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import org.springframework.http.HttpMethod;
-
-/**
- *
- * @author Rob Winch
- */
-public class DefaultWebRequestBuilderTests {
-	private DefaultClientWebRequestBuilder builder;
-
-	@Before
-	public void setup() {
-		builder = new DefaultClientWebRequestBuilder(HttpMethod.GET, "https://example.com/foo");
-	}
-
-	@Test
-	public void apply() {
-		ClientWebRequestPostProcessor postProcessor = mock(ClientWebRequestPostProcessor.class);
-		when(postProcessor.postProcess(any(ClientWebRequest.class))).thenAnswer(new Answer<ClientWebRequest>() {
-			@Override
-			public ClientWebRequest answer(InvocationOnMock invocation) throws Throwable {
-				return (ClientWebRequest) invocation.getArguments()[0];
-			}
-		});
-
-		ClientWebRequest webRequest = builder.apply(postProcessor).build();
-
-		verify(postProcessor).postProcess(webRequest);
-	}
-
-	@Test(expected = IllegalArgumentException.class)
-	public void applyNullPostProcessorThrowsIllegalArgumentException() {
-		builder.apply(null);
-	}
-}

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

@@ -0,0 +1,100 @@
+/*
+ * 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 org.junit.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.HttpHeaders;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class ExchangeFilterFunctionsTests {
+
+	@Test
+	public void andThen() throws Exception {
+		ClientRequest<Void> request = ClientRequest.GET("http://example.com").build();
+		ClientResponse response = mock(ClientResponse.class);
+		ExchangeFunction exchange = r -> Mono.just(response);
+
+		boolean[] filtersInvoked = new boolean[2];
+		ExchangeFilterFunction filter1 = (r, n) -> {
+			assertFalse(filtersInvoked[0]);
+			assertFalse(filtersInvoked[1]);
+			filtersInvoked[0] = true;
+			assertFalse(filtersInvoked[1]);
+			return n.exchange(r);
+		};
+		ExchangeFilterFunction filter2 = (r, n) -> {
+			assertTrue(filtersInvoked[0]);
+			assertFalse(filtersInvoked[1]);
+			filtersInvoked[1] = true;
+			return n.exchange(r);
+		};
+		ExchangeFilterFunction filter = filter1.andThen(filter2);
+
+
+		ClientResponse result = filter.filter(request, exchange).block();
+		assertEquals(response, result);
+
+		assertTrue(filtersInvoked[0]);
+		assertTrue(filtersInvoked[1]);
+	}
+
+	@Test
+	public void apply() throws Exception {
+		ClientRequest<Void> request = ClientRequest.GET("http://example.com").build();
+		ClientResponse response = mock(ClientResponse.class);
+		ExchangeFunction exchange = r -> Mono.just(response);
+
+		boolean[] filterInvoked = new boolean[1];
+		ExchangeFilterFunction filter = (r, n) -> {
+			assertFalse(filterInvoked[0]);
+			filterInvoked[0] = true;
+			return n.exchange(r);
+		};
+
+		ExchangeFunction filteredExchange = filter.apply(exchange);
+		ClientResponse result = filteredExchange.exchange(request).block();
+		assertEquals(response, result);
+		assertTrue(filterInvoked[0]);
+	}
+
+	@Test
+	public void basicAuthentication() throws Exception {
+		ClientRequest<Void> request = ClientRequest.GET("http://example.com").build();
+		ClientResponse response = mock(ClientResponse.class);
+
+		ExchangeFunction exchange = r -> {
+			assertTrue(r.headers().containsKey(HttpHeaders.AUTHORIZATION));
+			assertTrue(r.headers().getFirst(HttpHeaders.AUTHORIZATION).startsWith("Basic "));
+			return Mono.just(response);
+		};
+
+		ExchangeFilterFunction auth = ExchangeFilterFunctions.basicAuthentication("foo", "bar");
+		assertFalse(request.headers().containsKey(HttpHeaders.AUTHORIZATION));
+		ClientResponse result = auth.filter(request, exchange).block();
+		assertEquals(response, result);
+	}
+
+}

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

@@ -1,226 +0,0 @@
-/*
- * 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.io.UnsupportedEncodingException;
-import java.util.Arrays;
-import java.util.List;
-
-import org.junit.Before;
-import org.junit.Test;
-import reactor.core.Exceptions;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import org.springframework.core.codec.StringDecoder;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DefaultDataBufferFactory;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
-import org.springframework.http.client.reactive.ClientHttpResponse;
-import org.springframework.http.codec.DecoderHttpMessageReader;
-import org.springframework.http.codec.HttpMessageReader;
-import org.springframework.http.codec.json.Jackson2JsonDecoder;
-import org.springframework.tests.TestSubscriber;
-
-import static org.hamcrest.CoreMatchers.*;
-import static org.junit.Assert.*;
-import static org.mockito.BDDMockito.eq;
-import static org.mockito.BDDMockito.*;
-import static org.mockito.Mockito.mock;
-
-/**
- * Unit tests for {@link ResponseExtractors}.
- *
- * @author Brian Clozel
- */
-public class ResponseExtractorsTests {
-
-	private HttpHeaders headers;
-
-	private ClientHttpResponse response;
-
-	private List<HttpMessageReader<?>> messageReaders;
-
-	private WebClientConfig webClientConfig;
-
-	private ResponseErrorHandler errorHandler;
-
-
-	@Before
-	public void setup() throws Exception {
-		this.headers = new HttpHeaders();
-		this.response = mock(ClientHttpResponse.class);
-		given(this.response.getHeaders()).willReturn(headers);
-		this.messageReaders = Arrays.asList(
-				new DecoderHttpMessageReader<>(new StringDecoder()),
-				new DecoderHttpMessageReader<>(new Jackson2JsonDecoder()));
-		this.webClientConfig = mock(WebClientConfig.class);
-		this.errorHandler = mock(ResponseErrorHandler.class);
-		given(this.webClientConfig.getMessageReaders()).willReturn(this.messageReaders);
-		given(this.webClientConfig.getResponseErrorHandler()).willReturn(this.errorHandler);
-	}
-
-
-	@Test
-	public void shouldExtractResponseEntityMono() throws Exception {
-		this.headers.setContentType(MediaType.TEXT_PLAIN);
-		given(this.response.getStatusCode()).willReturn(HttpStatus.OK);
-		given(this.response.getBody()).willReturn(createFluxBody("test content"));
-
-		Mono<ResponseEntity<String>> result = ResponseExtractors.response(String.class)
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.awaitAndAssertNextValuesWith(entity -> {
-					assertThat(entity.getStatusCode(), is(HttpStatus.OK));
-					assertThat(entity.getHeaders().getContentType(), is(MediaType.TEXT_PLAIN));
-					assertThat(entity.getBody(), is("test content"));
-				})
-				.assertComplete();
-	}
-
-	@Test
-	public void shouldExtractResponseEntityFlux() throws Exception {
-		this.headers.setContentType(MediaType.TEXT_PLAIN);
-		given(this.response.getStatusCode()).willReturn(HttpStatus.OK);
-		given(this.response.getBody()).willReturn(createFluxBody("test", " content"));
-
-		Mono<ResponseEntity<String>> result = ResponseExtractors.response(String.class)
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.awaitAndAssertNextValuesWith(entity -> {
-					assertThat(entity.getStatusCode(), is(HttpStatus.OK));
-					assertThat(entity.getHeaders().getContentType(), is(MediaType.TEXT_PLAIN));
-					assertThat(entity.getBody(), is("test content"));
-				})
-				.assertComplete();
-	}
-
-	@Test
-	public void shouldExtractResponseEntityWithEmptyBody() throws Exception {
-		given(this.response.getStatusCode()).willReturn(HttpStatus.NO_CONTENT);
-		given(this.response.getBody()).willReturn(Flux.empty());
-
-		Mono<ResponseEntity<String>> result = ResponseExtractors.response(String.class)
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.awaitAndAssertNextValuesWith(entity -> {
-					assertThat(entity.getStatusCode(), is(HttpStatus.NO_CONTENT));
-					assertNull(entity.getBody());
-				})
-				.assertComplete();
-	}
-
-	@Test
-	public void shouldExtractResponseEntityAsStream() throws Exception {
-		this.headers.setContentType(MediaType.TEXT_PLAIN);
-		given(this.response.getStatusCode()).willReturn(HttpStatus.OK);
-		given(this.response.getBody()).willReturn(createFluxBody("test", " content"));
-
-		Mono<ResponseEntity<Flux<String>>> result = ResponseExtractors.responseStream(String.class)
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.awaitAndAssertNextValuesWith(entity -> {
-					assertThat(entity.getStatusCode(), is(HttpStatus.OK));
-					assertThat(entity.getHeaders().getContentType(), is(MediaType.TEXT_PLAIN));
-					TestSubscriber.subscribe(entity.getBody())
-							.awaitAndAssertNextValues("test", " content")
-							.assertComplete();
-				})
-				.assertComplete();
-	}
-
-	@Test
-	public void shouldGetErrorWhenExtractingWithMissingConverter() throws Exception {
-		this.headers.setContentType(MediaType.APPLICATION_XML);
-		given(this.response.getStatusCode()).willReturn(HttpStatus.OK);
-		given(this.response.getBody()).willReturn(createFluxBody("test content"));
-
-		Mono<ResponseEntity<SomePojo>> result = ResponseExtractors.response(SomePojo.class)
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.assertErrorWith(t -> {
-					assertThat(t, instanceOf(WebClientException.class));
-					WebClientException exc = (WebClientException) t;
-					assertThat(exc.getMessage(), containsString("Could not decode response body of type 'application/xml'"));
-					assertThat(exc.getMessage(), containsString("$SomePojo"));
-				});
-	}
-
-	@Test
-	public void shouldExtractResponseHeaders() throws Exception {
-		this.headers.setContentType(MediaType.TEXT_PLAIN);
-		this.headers.setETag("\"Spring\"");
-		given(this.response.getStatusCode()).willReturn(HttpStatus.OK);
-
-		Mono<HttpHeaders> result = ResponseExtractors.headers()
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.awaitAndAssertNextValuesWith(headers -> {
-					assertThat(headers.getContentType(), is(MediaType.TEXT_PLAIN));
-					assertThat(headers.getETag(), is("\"Spring\""));
-				})
-				.assertComplete();
-	}
-
-	@Test
-	public void shouldExecuteResponseHandler() throws Exception {
-		this.headers.setContentType(MediaType.TEXT_PLAIN);
-		given(this.response.getStatusCode()).willReturn(HttpStatus.NOT_FOUND);
-		given(this.response.getBody()).willReturn(createFluxBody("test", " content"));
-
-		Mono<String> result = ResponseExtractors.body(String.class)
-				.extract(Mono.just(this.response), this.webClientConfig);
-
-		TestSubscriber.subscribe(result)
-				.assertValueCount(1)
-				.assertComplete();
-
-		then(this.errorHandler).should().handleError(eq(this.response), eq(this.messageReaders));
-	}
-
-
-	private Flux<DataBuffer> createFluxBody(String... items) throws Exception {
-		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
-		return Flux.just(items)
-				.map(item -> {
-					DataBuffer buffer = factory.allocateBuffer();
-					try {
-						buffer.write(new String(item).getBytes("UTF-8"));
-					}
-					catch (UnsupportedEncodingException exc) {
-						Exceptions.propagate(exc);
-					}
-					return buffer;
-				});
-	}
-
-
-	protected class SomePojo {
-
-		public String foo;
-	}
-
-}

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

@@ -16,12 +16,7 @@
 
 package org.springframework.web.client.reactive;
 
-import static org.junit.Assert.*;
-import static org.springframework.web.client.reactive.ClientWebRequestBuilders.*;
-import static org.springframework.web.client.reactive.ResponseExtractors.*;
-
 import java.time.Duration;
-import java.util.function.Consumer;
 
 import okhttp3.HttpUrl;
 import okhttp3.mockwebserver.MockResponse;
@@ -35,12 +30,19 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.http.codec.BodyExtractors;
+import org.springframework.http.codec.BodyInserters;
 import org.springframework.http.codec.Pojo;
 import org.springframework.tests.TestSubscriber;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.springframework.http.codec.BodyExtractors.toFlux;
+import static org.springframework.http.codec.BodyExtractors.toMono;
+
 /**
  * {@link WebClient} integration tests with the {@code Flux} and {@code Mono} API.
  *
@@ -55,18 +57,18 @@ public class WebClientIntegrationTests {
 	@Before
 	public void setup() {
 		this.server = new MockWebServer();
-		this.webClient = new WebClient(new ReactorClientHttpConnector());
+		this.webClient = WebClient.create(new ReactorClientHttpConnector());
 	}
 
 	@Test
-	public void shouldGetHeaders() throws Exception {
-
+	public void headers() throws Exception {
 		HttpUrl baseUrl = server.url("/greeting?name=Spring");
 		this.server.enqueue(new MockResponse().setHeader("Content-Type", "text/plain").setBody("Hello Spring!"));
 
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString()).build();
 		Mono<HttpHeaders> result = this.webClient
-				.perform(get(baseUrl.toString()))
-				.extract(headers());
+				.exchange(request)
+				.map(response -> response.headers().asHttpHeaders());
 
 		TestSubscriber
 				.subscribe(result)
@@ -77,116 +79,101 @@ public class WebClientIntegrationTests {
 						})
 				.assertComplete();
 
-		RecordedRequest request = server.takeRequest();
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
-		assertEquals("/greeting?name=Spring", request.getPath());
+		assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/greeting?name=Spring", recordedRequest.getPath());
 	}
 
 	@Test
-	public void shouldGetPlainTextResponseAsObject() throws Exception {
-
+	public void plainText() throws Exception {
 		HttpUrl baseUrl = server.url("/greeting?name=Spring");
 		this.server.enqueue(new MockResponse().setBody("Hello Spring!"));
 
-		Mono<String> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.header("X-Test-Header", "testvalue"))
-				.extract(body(String.class));
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString())
+				.header("X-Test-Header", "testvalue")
+				.build();
 
+		Mono<String> result = this.webClient
+				.exchange(request)
+				.then(response -> response.body(toMono(String.class)));
 
 		TestSubscriber
 				.subscribe(result)
 				.awaitAndAssertNextValues("Hello Spring!")
 				.assertComplete();
 
-		RecordedRequest request = server.takeRequest();
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("testvalue", request.getHeader("X-Test-Header"));
-		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
-		assertEquals("/greeting?name=Spring", request.getPath());
+		assertEquals("testvalue", recordedRequest.getHeader("X-Test-Header"));
+		assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/greeting?name=Spring", recordedRequest.getPath());
 	}
 
 	@Test
-	public void shouldGetPlainTextResponse() throws Exception {
-
-		HttpUrl baseUrl = server.url("/greeting?name=Spring");
-		this.server.enqueue(new MockResponse().setHeader("Content-Type", "text/plain").setBody("Hello Spring!"));
-
-		Mono<ResponseEntity<String>> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.accept(MediaType.TEXT_PLAIN))
-				.extract(response(String.class));
-
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValuesWith((Consumer<ResponseEntity<String>>) response -> {
-					assertEquals(200, response.getStatusCode().value());
-					assertEquals(MediaType.TEXT_PLAIN, response.getHeaders().getContentType());
-					assertEquals("Hello Spring!", response.getBody());
-				});
-		RecordedRequest request = server.takeRequest();
-		assertEquals(1, server.getRequestCount());
-		assertEquals("/greeting?name=Spring", request.getPath());
-		assertEquals("text/plain", request.getHeader(HttpHeaders.ACCEPT));
-	}
-
-	@Test
-	public void shouldGetJsonAsMonoOfString() throws Exception {
-
+	public void jsonString() throws Exception {
 		HttpUrl baseUrl = server.url("/json");
 		String content = "{\"bar\":\"barbar\",\"foo\":\"foofoo\"}";
 		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
 				.setBody(content));
 
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString())
+				.accept(MediaType.APPLICATION_JSON)
+				.build();
+
 		Mono<String> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.accept(MediaType.APPLICATION_JSON))
-				.extract(body(String.class));
+				.exchange(request)
+				.then(response -> response.body(toMono(String.class)));
 
 		TestSubscriber
 				.subscribe(result)
 				.awaitAndAssertNextValues(content)
 				.assertComplete();
-		RecordedRequest request = server.takeRequest();
+
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("/json", request.getPath());
-		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/json", recordedRequest.getPath());
+		assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
 	}
 
 	@Test
-	public void shouldGetJsonAsMonoOfPojo() throws Exception {
-
+	public void jsonPojoMono() throws Exception {
 		HttpUrl baseUrl = server.url("/pojo");
 		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
 				.setBody("{\"bar\":\"barbar\",\"foo\":\"foofoo\"}"));
 
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString())
+				.accept(MediaType.APPLICATION_JSON)
+				.build();
+
 		Mono<Pojo> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.accept(MediaType.APPLICATION_JSON))
-				.extract(body(Pojo.class));
+				.exchange(request)
+				.then(response -> response.body(toMono(Pojo.class)));
 
 		TestSubscriber
 				.subscribe(result)
 				.awaitAndAssertNextValuesWith(p -> assertEquals("barbar", p.getBar()))
 				.assertComplete();
-		RecordedRequest request = server.takeRequest();
+
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("/pojo", request.getPath());
-		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/pojo", recordedRequest.getPath());
+		assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
 	}
 
 	@Test
-	public void shouldGetJsonAsFluxOfPojos() throws Exception {
-
+	public void jsonPojoFlux() throws Exception {
 		HttpUrl baseUrl = server.url("/pojos");
 		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
 				.setBody("[{\"bar\":\"bar1\",\"foo\":\"foo1\"},{\"bar\":\"bar2\",\"foo\":\"foo2\"}]"));
 
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString())
+				.accept(MediaType.APPLICATION_JSON)
+				.build();
+
 		Flux<Pojo> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.accept(MediaType.APPLICATION_JSON))
-				.extract(bodyStream(Pojo.class));
+				.exchange(request)
+				.flatMap(response -> response.body(toFlux(Pojo.class)));
 
 		TestSubscriber
 				.subscribe(result)
@@ -195,153 +182,124 @@ public class WebClientIntegrationTests {
 						p -> assertThat(p.getBar(), Matchers.is("bar2")))
 				.assertValueCount(2)
 				.assertComplete();
-		RecordedRequest request = server.takeRequest();
+
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("/pojos", request.getPath());
-		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/pojos", recordedRequest.getPath());
+		assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
 	}
 
 	@Test
-	public void shouldGetJsonAsResponseOfPojosStream() throws Exception {
-
-		HttpUrl baseUrl = server.url("/pojos");
-		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
-				.setBody("[{\"bar\":\"bar1\",\"foo\":\"foo1\"},{\"bar\":\"bar2\",\"foo\":\"foo2\"}]"));
-
-		Mono<ResponseEntity<Flux<Pojo>>> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.accept(MediaType.APPLICATION_JSON))
-				.extract(responseStream(Pojo.class));
-
-		TestSubscriber
-				.subscribe(result)
-				.awaitAndAssertNextValuesWith(
-						response -> {
-							assertEquals(200, response.getStatusCode().value());
-							assertEquals(MediaType.APPLICATION_JSON, response.getHeaders().getContentType());
-						})
-				.assertComplete();
-		RecordedRequest request = server.takeRequest();
-		assertEquals(1, server.getRequestCount());
-		assertEquals("/pojos", request.getPath());
-		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
-	}
-
-	@Test
-	public void shouldPostPojoAsJson() throws Exception {
-
+	public void postJsonPojo() throws Exception {
 		HttpUrl baseUrl = server.url("/pojo/capitalize");
 		this.server.enqueue(new MockResponse()
 				.setHeader("Content-Type", "application/json")
 				.setBody("{\"bar\":\"BARBAR\",\"foo\":\"FOOFOO\"}"));
 
 		Pojo spring = new Pojo("foofoo", "barbar");
+		ClientRequest<Pojo> request = ClientRequest.POST(baseUrl.toString())
+				.accept(MediaType.APPLICATION_JSON)
+				.contentType(MediaType.APPLICATION_JSON)
+				.body(BodyInserters.fromObject(spring));
+
 		Mono<Pojo> result = this.webClient
-				.perform(post(baseUrl.toString())
-						.body(spring)
-						.contentType(MediaType.APPLICATION_JSON)
-						.accept(MediaType.APPLICATION_JSON))
-				.extract(body(Pojo.class));
+				.exchange(request)
+				.then(response -> response.body(BodyExtractors.toMono(Pojo.class)));
 
 		TestSubscriber
 				.subscribe(result)
 				.awaitAndAssertNextValuesWith(p -> assertEquals("BARBAR", p.getBar()))
 				.assertComplete();
 
-		RecordedRequest request = server.takeRequest();
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("/pojo/capitalize", request.getPath());
-		assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", request.getBody().readUtf8());
-		assertEquals("chunked", request.getHeader(HttpHeaders.TRANSFER_ENCODING));
-		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
-		assertEquals("application/json", request.getHeader(HttpHeaders.CONTENT_TYPE));
+		assertEquals("/pojo/capitalize", recordedRequest.getPath());
+		assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", recordedRequest.getBody().readUtf8());
+		assertEquals("chunked", recordedRequest.getHeader(HttpHeaders.TRANSFER_ENCODING));
+		assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.CONTENT_TYPE));
 	}
 
 	@Test
-	public void shouldSendCookieHeader() throws Exception {
+	public void cookies() throws Exception {
 		HttpUrl baseUrl = server.url("/test");
 		this.server.enqueue(new MockResponse()
 				.setHeader("Content-Type", "text/plain").setBody("test"));
 
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString())
+				.cookie("testkey", "testvalue")
+				.build();
+
 		Mono<String> result = this.webClient
-				.perform(get(baseUrl.toString())
-						.cookie("testkey", "testvalue"))
-				.extract(body(String.class));
+				.exchange(request)
+				.then(response -> response.body(toMono(String.class)));
 
 		TestSubscriber
 				.subscribe(result)
 				.awaitAndAssertNextValues("test")
 				.assertComplete();
 
-		RecordedRequest request = server.takeRequest();
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("/test", request.getPath());
-		assertEquals("testkey=testvalue", request.getHeader(HttpHeaders.COOKIE));
+		assertEquals("/test", recordedRequest.getPath());
+		assertEquals("testkey=testvalue", recordedRequest.getHeader(HttpHeaders.COOKIE));
 	}
 
 	@Test
-	public void shouldGetErrorWhen404() throws Exception {
-
+	public void notFound() throws Exception {
 		HttpUrl baseUrl = server.url("/greeting?name=Spring");
 		this.server.enqueue(new MockResponse().setResponseCode(404)
 				.setHeader("Content-Type", "text/plain").setBody("Not Found"));
 
-		Mono<String> result = this.webClient
-				.perform(get(baseUrl.toString()))
-				.extract(body(String.class));
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString()).build();
+
+		Mono<ClientResponse> result = this.webClient
+				.exchange(request);
 
 		TestSubscriber
 				.subscribe(result)
 				.await(Duration.ofSeconds(3))
-				.assertErrorWith(t -> {
-					assertThat(t, Matchers.instanceOf(WebClientErrorException.class));
-					WebClientErrorException exc = (WebClientErrorException) t;
-					assertEquals(404, exc.getStatus().value());
-					assertEquals(MediaType.TEXT_PLAIN, exc.getResponseHeaders().getContentType());
-
-					Mono<String> body = exc.getResponseBody(as(String.class));
-
-					TestSubscriber.subscribe(body)
-							.awaitAndAssertNextValues("Not Found")
-							.assertComplete();
+				.assertValuesWith(response -> {
+					assertEquals(HttpStatus.NOT_FOUND, response.statusCode());
 				});
 
-		RecordedRequest request = server.takeRequest();
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
-		assertEquals("/greeting?name=Spring", request.getPath());
+		assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/greeting?name=Spring", recordedRequest.getPath());
 	}
 
 	@Test
-	public void shouldGetErrorWhen500() throws Exception {
-
+	public void filter() throws Exception {
 		HttpUrl baseUrl = server.url("/greeting?name=Spring");
-		this.server.enqueue(new MockResponse().setResponseCode(500)
-				.setHeader("Content-Type", "text/plain").setBody("Server Error"));
+		this.server.enqueue(new MockResponse().setHeader("Content-Type", "text/plain").setBody("Hello Spring!"));
 
-		Mono<String> result = this.webClient
-				.perform(get(baseUrl.toString()))
-				.extract(body(String.class));
+		ExchangeFilterFunction filter = (request, next) -> {
+			ClientRequest<?> filteredRequest = ClientRequest.from(request)
+					.header("foo", "bar").build();
+			return next.exchange(filteredRequest);
+		};
+		WebClient filteredClient = WebClient.builder(new ReactorClientHttpConnector())
+				.filter(filter).build();
+
+		ClientRequest<Void> request = ClientRequest.GET(baseUrl.toString()).build();
+
+		Mono<String> result = filteredClient.exchange(request)
+				.then(response -> response.body(toMono(String.class)));
 
 		TestSubscriber
 				.subscribe(result)
-				.await(Duration.ofSeconds(3))
-				.assertErrorWith(t -> {
-					assertThat(t, Matchers.instanceOf(WebServerErrorException.class));
-					WebServerErrorException exc = (WebServerErrorException) t;
-					assertEquals(500, exc.getStatus().value());
-					assertEquals(MediaType.TEXT_PLAIN, exc.getResponseHeaders().getContentType());
-				});
+				.awaitAndAssertNextValues("Hello Spring!")
+				.assertComplete();
 
-		RecordedRequest request = server.takeRequest();
+		RecordedRequest recordedRequest = server.takeRequest();
 		assertEquals(1, server.getRequestCount());
-		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
-		assertEquals("/greeting?name=Spring", request.getPath());
+		assertEquals("bar", recordedRequest.getHeader("foo"));
+
 	}
 
 	@After
 	public void tearDown() throws Exception {
 		this.server.shutdown();
 	}
-
 }

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

@@ -0,0 +1,132 @@
+/*
+ * 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.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.context.support.StaticApplicationContext;
+import org.springframework.core.ResolvableType;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpInputMessage;
+import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.http.codec.HttpMessageReader;
+import org.springframework.http.codec.HttpMessageWriter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Arjen Poutsma
+ */
+public class WebClientStrategiesTests {
+
+	@Test
+	public void empty() {
+		WebClientStrategies strategies = WebClientStrategies.empty().build();
+		assertEquals(Optional.empty(), strategies.messageReaders().get().findFirst());
+		assertEquals(Optional.empty(), strategies.messageWriters().get().findFirst());
+	}
+
+	@Test
+	public void ofSuppliers() {
+		HttpMessageReader<?> messageReader = new DummyMessageReader();
+		HttpMessageWriter<?> messageWriter = new DummyMessageWriter();
+
+		WebClientStrategies strategies = WebClientStrategies.of(
+				() -> Stream.of(messageReader),
+				() -> Stream.of(messageWriter));
+
+		assertEquals(1L, strategies.messageReaders().get().collect(Collectors.counting()).longValue());
+		assertEquals(Optional.of(messageReader), strategies.messageReaders().get().findFirst());
+
+		assertEquals(1L, strategies.messageWriters().get().collect(Collectors.counting()).longValue());
+		assertEquals(Optional.of(messageWriter), strategies.messageWriters().get().findFirst());
+	}
+
+	@Test
+	public void toConfiguration() throws Exception {
+		StaticApplicationContext applicationContext = new StaticApplicationContext();
+		applicationContext.registerSingleton("messageWriter", DummyMessageWriter.class);
+		applicationContext.registerSingleton("messageReader", DummyMessageReader.class);
+		applicationContext.refresh();
+
+		WebClientStrategies strategies = WebClientStrategies.of(applicationContext);
+		assertTrue(strategies.messageReaders().get()
+				.allMatch(r -> r instanceof DummyMessageReader));
+		assertTrue(strategies.messageWriters().get()
+				.allMatch(r -> r instanceof DummyMessageWriter));
+
+	}
+
+
+	private static class DummyMessageWriter implements HttpMessageWriter<Object> {
+
+		@Override
+		public boolean canWrite(ResolvableType type, MediaType mediaType) {
+			return false;
+		}
+
+		@Override
+		public List<MediaType> getWritableMediaTypes() {
+			return Collections.emptyList();
+		}
+
+		@Override
+		public Mono<Void> write(Publisher<?> inputStream, ResolvableType type,
+				MediaType contentType,
+				ReactiveHttpOutputMessage outputMessage,
+				Map<String, Object> hints) {
+			return Mono.empty();
+		}
+	}
+
+	private static class DummyMessageReader implements HttpMessageReader<Object> {
+
+		@Override
+		public boolean canRead(ResolvableType type, MediaType mediaType) {
+			return false;
+		}
+
+		@Override
+		public List<MediaType> getReadableMediaTypes() {
+			return Collections.emptyList();
+		}
+
+		@Override
+		public Flux<Object> read(ResolvableType type, ReactiveHttpInputMessage inputMessage,
+				Map<String, Object> hints) {
+			return Flux.empty();
+		}
+
+		@Override
+		public Mono<Object> readMono(ResolvableType type, ReactiveHttpInputMessage inputMessage,
+				Map<String, Object> hints) {
+			return Mono.empty();
+		}
+	}
+}
+