Add WebClient and its WebResponseExtractor API

This commit adds the `WebClient`, which relies on several parts of our
infrastructure:
* a `ClientHttpRequestFactory` to drive the underlying client library
* a `HttpRequestBuilder` builder API to create the client request
* a `WebResponseExtractor` to extract the "low-level"
`ClientHttpResponse` into a higher level representation such as a
decoded body

The `WebResponseExtractors` helper class contains many extractor
implementations all based on the `Flux`/`Mono` composition API.

spring-web-reactive/build.gradle

@@ -102,14 +102,15 @@ dependencies {
 	testCompile "org.springframework:spring-test:${springVersion}"
 	testCompile "org.slf4j:slf4j-jcl:1.7.12"
 	testCompile "org.slf4j:jul-to-slf4j:1.7.12"
-	testCompile("log4j:log4j:1.2.16")
+	testCompile "log4j:log4j:1.2.16"
 	testCompile("org.mockito:mockito-core:1.10.19") {
 		exclude group: 'org.hamcrest', module: 'hamcrest-core'
 	}
-	testCompile("org.hamcrest:hamcrest-all:1.3")
+	testCompile "org.hamcrest:hamcrest-all:1.3"
+	testCompile "com.squareup.okhttp3:mockwebserver:3.0.1"
 
 	// Needed to run Javadoc without error
-	optional("org.apache.httpcomponents:httpclient:4.5.1")
+	optional "org.apache.httpcomponents:httpclient:4.5.1"
 }
 
 

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

@@ -0,0 +1,52 @@
+/*
+ * 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.List;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.codec.Decoder;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+
+/**
+ * Default implementation of the {@link WebResponse} interface
+ *
+ * @author Brian Clozel
+ */
+public class DefaultWebResponse implements WebResponse {
+
+	private final Mono<ClientHttpResponse> clientResponse;
+
+	private final List<Decoder<?>> messageDecoders;
+
+
+	public DefaultWebResponse(Mono<ClientHttpResponse> clientResponse, List<Decoder<?>> messageDecoders) {
+		this.clientResponse = clientResponse;
+		this.messageDecoders = messageDecoders;
+	}
+
+	@Override
+	public Mono<ClientHttpResponse> getClientResponse() {
+		return this.clientResponse;
+	}
+
+	@Override
+	public List<Decoder<?>> getMessageDecoders() {
+		return this.messageDecoders;
+	}
+}

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

@@ -0,0 +1,139 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.codec.Decoder;
+import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.support.ByteBufferDecoder;
+import org.springframework.core.codec.support.ByteBufferEncoder;
+import org.springframework.core.codec.support.JacksonJsonDecoder;
+import org.springframework.core.codec.support.JacksonJsonEncoder;
+import org.springframework.core.codec.support.JsonObjectDecoder;
+import org.springframework.core.codec.support.StringDecoder;
+import org.springframework.core.codec.support.StringEncoder;
+import org.springframework.core.io.buffer.DataBufferAllocator;
+import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.http.client.reactive.ClientHttpRequestFactory;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+
+/**
+ * Reactive Web client supporting the HTTP/1.1 protocol
+ *
+ * <p>Here is a simple example of a GET request:
+ * <pre class="code">
+ * WebClient client = new WebClient(new ReactorHttpClientRequestFactory());
+ * Mono&lt;String&gt; result = client
+ * 		.perform(HttpRequestBuilders.get("http://example.org/resource")
+ * 			.accept(MediaType.TEXT_PLAIN))
+ * 		.extract(WebResponseExtractors.body(String.class));
+ * </pre>
+ *
+ * <p>This Web client relies on
+ * <ul>
+ *     <li>a {@link ClientHttpRequestFactory} that drives the underlying library (e.g. Reactor-Net, RxNetty...)</li>
+ *     <li>an {@link HttpRequestBuilder} which create a Web request with a builder API (see {@link HttpRequestBuilders})</li>
+ *     <li>an {@link WebResponseExtractor} which extracts the relevant part of the server response
+ *     with the composition API of choice (see {@link WebResponseExtractors}</li>
+ * </ul>
+ *
+ * @author Brian Clozel
+ * @see HttpRequestBuilders
+ * @see WebResponseExtractors
+ */
+public final class WebClient {
+
+	private ClientHttpRequestFactory requestFactory;
+
+	private List<Encoder<?>> messageEncoders;
+
+	private List<Decoder<?>> messageDecoders;
+
+	/**
+	 * Create a {@code ReactiveRestClient} instance, using the {@link ClientHttpRequestFactory}
+	 * implementation given as an argument to drive the underlying HTTP client implementation.
+	 *
+	 * Register by default the following Encoders and Decoders:
+	 * <ul>
+	 *     <li>{@link ByteBufferEncoder} / {@link ByteBufferDecoder}</li>
+	 *     <li>{@link StringEncoder} / {@link StringDecoder}</li>
+	 *     <li>{@link JacksonJsonEncoder} / {@link JacksonJsonDecoder}</li>
+	 * </ul>
+	 *
+	 * @param requestFactory the {@code ClientHttpRequestFactory} to use
+	 */
+	public WebClient(ClientHttpRequestFactory requestFactory) {
+		this.requestFactory = requestFactory;
+		DataBufferAllocator allocator = new DefaultDataBufferAllocator();
+		this.messageEncoders = Arrays.asList(new ByteBufferEncoder(allocator), new StringEncoder(allocator),
+				new JacksonJsonEncoder(allocator));
+		this.messageDecoders = Arrays.asList(new ByteBufferDecoder(), new StringDecoder(allocator),
+				new JacksonJsonDecoder(new JsonObjectDecoder(allocator)));
+	}
+
+	/**
+	 * Set the list of {@link Encoder}s to use for encoding messages
+	 */
+	public void setMessageEncoders(List<Encoder<?>> messageEncoders) {
+		this.messageEncoders = messageEncoders;
+	}
+
+	/**
+	 * Set the list of {@link Decoder}s to use for decoding messages
+	 */
+	public void setMessageDecoders(List<Decoder<?>> messageDecoders) {
+		this.messageDecoders = messageDecoders;
+	}
+
+	/**
+	 * Perform the actual HTTP request/response exchange
+	 *
+	 * <p>Pulling demand from the exposed {@code Flux} will result in:
+	 * <ul>
+	 *     <li>building the actual HTTP request using the provided {@code RequestBuilder}</li>
+	 *     <li>encoding the HTTP request body with the configured {@code Encoder}s</li>
+	 *     <li>returning the response with a publisher of the body</li>
+	 * </ul>
+	 */
+	public WebResponseActions perform(DefaultHttpRequestBuilder builder) {
+
+		ClientHttpRequest request = builder.setMessageEncoders(messageEncoders).build(requestFactory);
+		final Mono<ClientHttpResponse> clientResponse = request.execute()
+				.log("org.springframework.http.client.reactive");
+
+		return new WebResponseActions() {
+			@Override
+			public void doWithStatus(Consumer<HttpStatus> consumer) {
+				// TODO: implement
+			}
+
+			@Override
+			public <T> T extract(WebResponseExtractor<T> extractor) {
+				return extractor.extract(new DefaultWebResponse(clientResponse, messageDecoders));
+			}
+
+		};
+	}
+
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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.List;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.codec.Decoder;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+
+/**
+ * Result of a {@code ClientHttpRequest} sent to a remote server by the {@code WebClient}
+ *
+ * <p>Contains all the required information to extract relevant information from the raw response.
+ *
+ * @author Brian Clozel
+ */
+public interface WebResponse {
+
+	/**
+	 * Return the raw response received by the {@code WebClient}
+	 */
+	Mono<ClientHttpResponse> getClientResponse();
+
+	/**
+	 * Return the configured list of {@link Decoder}s that can be used to decode the raw response body
+	 */
+	List<Decoder<?>> getMessageDecoders();
+}

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

@@ -0,0 +1,49 @@
+/*
+ * 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.function.Consumer;
+
+import org.springframework.http.HttpStatus;
+
+/**
+ * Allows applying actions, such as extractors, on the result of an executed
+ * {@link WebClient} request.
+ *
+ * @author Brian Clozel
+ */
+public interface WebResponseActions {
+
+	/**
+	 * Apply synchronous operations once the HTTP response status
+	 * has been received.
+	 */
+	void doWithStatus(Consumer<HttpStatus> consumer);
+
+	/**
+	 * Perform an extraction of the response body into a higher level representation.
+	 *
+	 * <pre class="code">
+	 * static imports: HttpRequestBuilders.*, HttpResponseExtractors.*
+	 *
+	 * webClient
+	 *   .perform(get(baseUrl.toString()).accept(MediaType.TEXT_PLAIN))
+	 *   .extract(response(String.class));
+	 * </pre>
+	 */
+	<T> T extract(WebResponseExtractor<T> extractor);
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+/**
+ * A {@code WebResponseExtractor} extracts the relevant part of a
+ * raw {@link org.springframework.http.client.reactive.ClientHttpResponse},
+ * optionally decoding the response body and using a target composition API.
+ *
+ * <p>See static factory methods in {@link WebResponseExtractors}.
+ *
+ * @author Brian Clozel
+ */
+public interface WebResponseExtractor<T> {
+
+	T extract(WebResponse webResponse);
+}

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

@@ -0,0 +1,123 @@
+/*
+ * 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.List;
+import java.util.Optional;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.Decoder;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+
+/**
+ * Static factory methods for {@link WebResponseExtractor}
+ * based on the {@link Flux} and {@link Mono} API.
+ *
+ * @author Brian Clozel
+ */
+public class WebResponseExtractors {
+
+	private static final Charset UTF_8 = Charset.forName("UTF-8");
+
+	private static final Object[] HINTS = new Object[] {UTF_8};
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Mono<T>}
+	 */
+	public static <T> WebResponseExtractor<Mono<T>> body(Class<T> sourceClass) {
+
+		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
+		//noinspection unchecked
+		return webResponse -> (Mono<T>) webResponse.getClientResponse()
+				.flatMap(resp -> decodeResponseBody(resp, resolvableType, webResponse.getMessageDecoders()))
+				.next();
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Flux<T>}
+	 */
+	public static <T> WebResponseExtractor<Flux<T>> bodyStream(Class<T> sourceClass) {
+
+		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
+		return webResponse -> webResponse.getClientResponse()
+				.flatMap(resp -> decodeResponseBody(resp, resolvableType, webResponse.getMessageDecoders()));
+	}
+
+	/**
+	 * Extract the full response body as a {@code ResponseEntity}
+	 * with its body decoded as a single type {@code T}
+	 */
+	public static <T> WebResponseExtractor<Mono<ResponseEntity<T>>> response(Class<T> sourceClass) {
+
+		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
+		return webResponse -> webResponse.getClientResponse()
+				.then(response ->
+						Mono.when(
+								decodeResponseBody(response, resolvableType, webResponse.getMessageDecoders()).next(),
+								Mono.just(response.getHeaders()),
+								Mono.just(response.getStatusCode())))
+				.map(tuple -> {
+					//noinspection unchecked
+					return new ResponseEntity<>((T) tuple.getT1(), tuple.getT2(), tuple.getT3());
+				});
+	}
+
+	/**
+	 * Extract the full response body as a {@code ResponseEntity}
+	 * with its body decoded as a {@code Flux<T>}
+	 */
+	public static <T> WebResponseExtractor<Mono<ResponseEntity<Flux<T>>>> responseStream(Class<T> sourceClass) {
+		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
+		return webResponse -> webResponse.getClientResponse()
+				.map(response -> new ResponseEntity<>(
+						decodeResponseBody(response, resolvableType, webResponse.getMessageDecoders()),
+						response.getHeaders(), response.getStatusCode()));
+	}
+
+	/**
+	 * Extract the response headers as an {@code HttpHeaders} instance
+	 */
+	public static WebResponseExtractor<Mono<HttpHeaders>> headers() {
+		return webResponse -> webResponse.getClientResponse().map(resp -> resp.getHeaders());
+	}
+
+	protected static <T> Flux<T> decodeResponseBody(ClientHttpResponse response, ResolvableType responseType,
+			List<Decoder<?>> messageDecoders) {
+
+		MediaType contentType = response.getHeaders().getContentType();
+		Optional<Decoder<?>> decoder = resolveDecoder(messageDecoders, responseType, contentType);
+		if (!decoder.isPresent()) {
+			return Flux.error(new IllegalStateException("Could not decode response body of type '" + contentType +
+					"' with target type '" + responseType.toString() + "'"));
+		}
+		//noinspection unchecked
+		return (Flux<T>) decoder.get().decode(response.getBody(), responseType, contentType, HINTS);
+	}
+
+
+	protected static Optional<Decoder<?>> resolveDecoder(List<Decoder<?>> messageDecoders, ResolvableType type,
+			MediaType mediaType) {
+		return messageDecoders.stream().filter(e -> e.canDecode(type, mediaType)).findFirst();
+	}
+}

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

@@ -0,0 +1,249 @@
+/*
+ * 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.junit.Assert.*;
+import static org.springframework.web.client.reactive.HttpRequestBuilders.*;
+import static org.springframework.web.client.reactive.WebResponseExtractors.*;
+
+import okhttp3.HttpUrl;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.test.TestSubscriber;
+import reactor.fn.Consumer;
+
+import org.springframework.core.codec.support.Pojo;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.reactive.ReactorHttpClientRequestFactory;
+
+/**
+ * @author Brian Clozel
+ */
+public class WebClientIntegrationTests {
+
+	private MockWebServer server;
+
+	private WebClient webClient;
+
+	@Before
+	public void setup() {
+		this.server = new MockWebServer();
+		this.webClient = new WebClient(new ReactorHttpClientRequestFactory());
+	}
+
+	@Test
+	public void shouldGetHeaders() throws Exception {
+
+		HttpUrl baseUrl = server.url("/greeting?name=Spring");
+		this.server.enqueue(new MockResponse().setHeader("Content-Type", "text/plain").setBody("Hello Spring!"));
+
+		Mono<HttpHeaders> result = this.webClient
+				.perform(get(baseUrl.toString()))
+				.extract(headers());
+
+		TestSubscriber<HttpHeaders> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValuesWith(
+				httpHeaders -> {
+					assertEquals(MediaType.TEXT_PLAIN, httpHeaders.getContentType());
+					assertEquals(13L, httpHeaders.getContentLength());
+				}
+		).assertComplete();
+
+		RecordedRequest request = server.takeRequest();
+		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/greeting?name=Spring", request.getPath());
+	}
+
+	@Test
+	public void shouldGetPlainTextResponseAsObject() 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));
+
+
+		TestSubscriber<String> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValues("Hello Spring!").assertComplete();
+
+		RecordedRequest request = server.takeRequest();
+		assertEquals("testvalue", request.getHeader("X-Test-Header"));
+		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/greeting?name=Spring", request.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<ResponseEntity<String>> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValuesWith(new Consumer<ResponseEntity<String>>() {
+			@Override
+			public void accept(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("/greeting?name=Spring", request.getPath());
+		assertEquals("text/plain", request.getHeader(HttpHeaders.ACCEPT));
+	}
+
+	@Test
+	public void shouldGetJsonAsMonoOfPojo() throws Exception {
+
+		HttpUrl baseUrl = server.url("/pojo");
+		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
+				.setBody("{\"bar\":\"barbar\",\"foo\":\"foofoo\"}"));
+
+		Mono<Pojo> result = this.webClient
+				.perform(get(baseUrl.toString())
+						.accept(MediaType.APPLICATION_JSON))
+				.extract(body(Pojo.class));
+
+		TestSubscriber<Pojo> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValuesWith(p -> assertEquals("barbar", p.getBar())).assertComplete();
+		RecordedRequest request = server.takeRequest();
+		assertEquals("/pojo", request.getPath());
+		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
+	}
+
+	@Test
+	public void shouldGetJsonAsFluxOfPojos() 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\"}]"));
+
+		Flux<Pojo> result = this.webClient
+				.perform(get(baseUrl.toString())
+						.accept(MediaType.APPLICATION_JSON))
+				.extract(bodyStream(Pojo.class));
+
+		TestSubscriber<Pojo> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValuesWith(
+				p -> assertThat(p.getBar(), Matchers.is("bar1")),
+				p -> assertThat(p.getBar(), Matchers.is("bar2"))
+		).assertValueCount(2).assertComplete();
+		RecordedRequest request = server.takeRequest();
+		assertEquals("/pojos", request.getPath());
+		assertEquals("application/json", request.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<ResponseEntity<Flux<Pojo>>> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValuesWith(
+				response -> {
+					assertEquals(200, response.getStatusCode().value());
+					assertEquals(MediaType.APPLICATION_JSON, response.getHeaders().getContentType());
+				}
+		).assertComplete();
+		RecordedRequest request = server.takeRequest();
+		assertEquals("/pojos", request.getPath());
+		assertEquals("application/json", request.getHeader(HttpHeaders.ACCEPT));
+	}
+
+	@Test
+	public void shouldPostPojoAsJson() throws Exception {
+
+		HttpUrl baseUrl = server.url("/pojo/capitalize");
+		this.server.enqueue(new MockResponse().setBody("{\"bar\":\"BARBAR\",\"foo\":\"FOOFOO\"}"));
+
+		Pojo spring = new Pojo("foofoo", "barbar");
+		Mono<Pojo> result = this.webClient
+				.perform(post(baseUrl.toString())
+						.content(spring)
+						.contentType(MediaType.APPLICATION_JSON)
+						.accept(MediaType.APPLICATION_JSON))
+				.extract(body(Pojo.class));
+
+		TestSubscriber<Pojo> ts = new TestSubscriber();
+		result.subscribe(ts);
+		ts.awaitAndAssertValuesWith(p -> assertEquals("BARBAR", p.getBar())).assertComplete();
+
+		RecordedRequest request = server.takeRequest();
+		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));
+	}
+
+	@Test
+	public void shouldGetErrorWhen404() throws Exception {
+
+		HttpUrl baseUrl = server.url("/greeting?name=Spring");
+		this.server.enqueue(new MockResponse().setResponseCode(404));
+
+		Mono<String> result = this.webClient
+				.perform(get(baseUrl.toString()))
+				.extract(body(String.class));
+
+
+		TestSubscriber<String> ts = new TestSubscriber();
+		result.subscribe(ts);
+		// TODO: error message should be converted to a ClientException
+		ts.await().assertError();
+
+		RecordedRequest request = server.takeRequest();
+		assertEquals("*/*", request.getHeader(HttpHeaders.ACCEPT));
+		assertEquals("/greeting?name=Spring", request.getPath());
+	}
+
+	@After
+	public void tearDown() throws Exception {
+		this.server.shutdown();
+	}
+
+}