Extract body population logic in w.r.f

This commit extracts the response body insertion logic into a separate
strategy interface: BodyPopulator. Standard populators can be found in
BodyPopulators.

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/BodyPopulator.java

@@ -0,0 +1,64 @@
+/*
+ * 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.reactive.function;
+
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.Assert;
+
+/**
+ * A combination of functions that can populate {@link Response#body()}.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ * @see Response.BodyBuilder#body(BodyPopulator)
+ * @see BodyPopulators
+ */
+public interface BodyPopulator<T> {
+
+	/**
+	 * Return a new {@code BodyPopulator} described by the given writer and supplier functions.
+
+	 * @param writer  the writer function for the new populator
+	 * @param supplier the supplier function for the new populator
+	 * @param <T> the type supplied and written by the populator
+	 * @return the new {@code BodyPopulator}
+	 */
+	static <T> BodyPopulator<T> of(BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer,
+			Supplier<T> supplier) {
+
+		Assert.notNull(writer, "'writer' must not be null");
+		Assert.notNull(supplier, "'supplier' must not be null");
+
+		return new BodyPopulators.DefaultBodyPopulator<T>(writer, supplier);
+	}
+
+	/**
+	 * Return a function that writes to the given response body.
+	 */
+	BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer();
+
+	/**
+	 * Return a function that supplies the type contained in the body.
+	 */
+	Supplier<T> supplier();
+
+}

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/BodyPopulators.java

@@ -0,0 +1,254 @@
+/*
+ * 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.reactive.function;
+
+import java.util.Collections;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.Resource;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.codec.ResourceHttpMessageWriter;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.http.codec.ServerSentEventHttpMessageWriter;
+import org.springframework.http.codec.json.Jackson2JsonEncoder;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Implementations of {@link BodyPopulator} that write various bodies, such a reactive streams,
+ * server-sent events, resources, etc.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+public abstract class BodyPopulators {
+
+	private static final ResolvableType RESOURCE_TYPE = ResolvableType.forClass(Resource.class);
+
+	private static final ResolvableType SERVER_SIDE_EVENT_TYPE =
+			ResolvableType.forClass(ServerSentEvent.class);
+
+	private static final boolean jackson2Present =
+			ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper",
+					BodyPopulators.class.getClassLoader()) &&
+					ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator",
+							BodyPopulators.class.getClassLoader());
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given single object.
+	 * @param body the body of the response
+	 * @return a {@code BodyPopulator} that writes a single object
+	 */
+	public static <T> BodyPopulator<T> ofObject(T body) {
+		Assert.notNull(body, "'body' must not be null");
+		return BodyPopulator.of(
+				(response, configuration) -> writeWithMessageWriters(response, configuration,
+						Mono.just(body), ResolvableType.forInstance(body)),
+				() -> body);
+	}
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given {@link Publisher}.
+	 * @param publisher the publisher to stream to the response body
+	 * @param elementClass the class of elements contained in the publisher
+	 * @param <T> the type of the elements contained in the publisher
+	 * @param <S> the type of the {@code Publisher}.
+	 * @return a {@code BodyPopulator} that writes a {@code Publisher}
+	 */
+	public static <S extends Publisher<T>, T> BodyPopulator<S> ofPublisher(S publisher,
+			Class<T> elementClass) {
+
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+		return ofPublisher(publisher, ResolvableType.forClass(elementClass));
+	}
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given {@link Publisher}.
+	 * @param publisher the publisher to stream to the response body
+	 * @param elementType the type of elements contained in the publisher
+	 * @param <T> the type of the elements contained in the publisher
+	 * @param <S> the type of the {@code Publisher}.
+	 * @return a {@code BodyPopulator} that writes a {@code Publisher}
+	 */
+	public static <S extends Publisher<T>, T> BodyPopulator<S> ofPublisher(S publisher,
+			ResolvableType elementType) {
+
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementType, "'elementType' must not be null");
+		return BodyPopulator.of(
+				(response, configuration) -> writeWithMessageWriters(response, configuration,
+						publisher, elementType),
+				() -> publisher
+		);
+	}
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given {@code Resource}.
+	 * If the resource can be resolved to a {@linkplain Resource#getFile() file}, it will be copied
+	 * using
+	 * <a href="https://en.wikipedia.org/wiki/Zero-copy">zero-copy</a>
+	 * @param resource the resource to write to the response
+	 * @param <T> the type of the {@code Resource}
+	 * @return a {@code BodyPopulator} that writes a {@code Publisher}
+	 */
+	public static <T extends Resource> BodyPopulator<T> ofResource(T resource) {
+		Assert.notNull(resource, "'resource' must not be null");
+		return BodyPopulator.of(
+				(response, configuration) -> {
+					ResourceHttpMessageWriter messageWriter = new ResourceHttpMessageWriter();
+					MediaType contentType = response.getHeaders().getContentType();
+					return messageWriter.write(Mono.just(resource), RESOURCE_TYPE, contentType,
+							response, Collections.emptyMap());
+				},
+				() -> resource
+		);
+	}
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given {@code ServerSentEvent} publisher.
+	 * @param eventsPublisher the {@code ServerSentEvent} publisher to write to the response body
+	 * @param <T> the type of the elements contained in the {@link ServerSentEvent}
+	 * @return a {@code BodyPopulator} that writes a {@code ServerSentEvent} publisher
+	 * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
+	 */
+	public static <T, S extends Publisher<ServerSentEvent<T>>> BodyPopulator<S> ofServerSentEvents(
+			S eventsPublisher) {
+
+		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
+		return BodyPopulator.of(
+				(response, configuration) -> {
+					ServerSentEventHttpMessageWriter messageWriter = sseMessageWriter();
+					MediaType contentType = response.getHeaders().getContentType();
+					return messageWriter.write(eventsPublisher, SERVER_SIDE_EVENT_TYPE,
+							contentType, response, Collections.emptyMap());
+				},
+				() -> eventsPublisher
+		);
+	}
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given {@code Publisher} publisher as
+	 * Server-Sent Events.
+	 * @param eventsPublisher the publisher to write to the response body as Server-Sent Events
+	 * @param eventClass the class of event contained in the publisher
+	 * @param <T> the type of the elements contained in the publisher
+	 * @return a {@code BodyPopulator} that writes the given {@code Publisher} publisher as
+	 * Server-Sent Events
+	 * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
+	 */
+	public static <T, S extends Publisher<T>> BodyPopulator<S> ofServerSentEvents(S eventsPublisher,
+			Class<T> eventClass) {
+
+		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
+		Assert.notNull(eventClass, "'eventClass' must not be null");
+		return ofServerSentEvents(eventsPublisher, ResolvableType.forClass(eventClass));
+	}
+
+	/**
+	 * Return a {@code BodyPopulator} that writes the given {@code Publisher} publisher as
+	 * Server-Sent Events.
+	 * @param eventsPublisher the publisher to write to the response body as Server-Sent Events
+	 * @param eventType the type of event contained in the publisher
+	 * @param <T> the type of the elements contained in the publisher
+	 * @return a {@code BodyPopulator} that writes the given {@code Publisher} publisher as
+	 * Server-Sent Events
+	 * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
+	 */
+	public static <T, S extends Publisher<T>> BodyPopulator<S> ofServerSentEvents(S eventsPublisher,
+			ResolvableType eventType) {
+
+		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
+		Assert.notNull(eventType, "'eventType' must not be null");
+		return BodyPopulator.of(
+				(response, configuration) -> {
+					ServerSentEventHttpMessageWriter messageWriter = sseMessageWriter();
+					MediaType contentType = response.getHeaders().getContentType();
+					return messageWriter.write(eventsPublisher, eventType, contentType, response,
+							Collections.emptyMap());
+
+				},
+				() -> eventsPublisher
+		);
+	}
+
+	private static ServerSentEventHttpMessageWriter sseMessageWriter() {
+		return jackson2Present ? new ServerSentEventHttpMessageWriter(
+				Collections.singletonList(new Jackson2JsonEncoder())) :
+				new ServerSentEventHttpMessageWriter();
+	}
+
+	private static <T> Mono<Void> writeWithMessageWriters(ServerHttpResponse response,
+			Configuration configuration,
+			Publisher<T> body,
+			ResolvableType bodyType) {
+
+		// TODO: use ContentNegotiatingResultHandlerSupport
+		MediaType contentType = response.getHeaders().getContentType();
+		return configuration.messageWriters().get()
+				.filter(messageWriter -> messageWriter.canWrite(bodyType, contentType, Collections
+						.emptyMap()))
+				.findFirst()
+				.map(BodyPopulators::cast)
+				.map(messageWriter -> messageWriter
+						.write(body, bodyType, contentType, response, Collections
+								.emptyMap()))
+				.orElseGet(() -> {
+					response.setStatusCode(HttpStatus.NOT_ACCEPTABLE);
+					return response.setComplete();
+				});
+	}
+
+	@SuppressWarnings("unchecked")
+	public static <T> HttpMessageWriter<T> cast(HttpMessageWriter<?> messageWriter) {
+		return (HttpMessageWriter<T>) messageWriter;
+	}
+
+	static class DefaultBodyPopulator<T> implements BodyPopulator<T> {
+
+		private final BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer;
+
+		private final Supplier<T> supplier;
+
+		public DefaultBodyPopulator(
+				BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer,
+				Supplier<T> supplier) {
+			this.writer = writer;
+			this.supplier = supplier;
+		}
+
+		@Override
+		public BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer() {
+			return this.writer;
+		}
+
+		@Override
+		public Supplier<T> supplier() {
+			return this.supplier;
+		}
+	}
+
+
+}

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/CastingUtils.java

@@ -17,7 +17,6 @@
 package org.springframework.web.reactive.function;
 
 import org.springframework.http.codec.HttpMessageReader;
-import org.springframework.http.codec.HttpMessageWriter;
 
 /**
  * @author Arjen Poutsma
@@ -29,10 +28,6 @@ abstract class CastingUtils {
 		return (HttpMessageReader<T>) messageReader;
 	}
 
-	public static <T> HttpMessageWriter<T> cast(HttpMessageWriter<?> messageWriter) {
-		return (HttpMessageWriter<T>) messageWriter;
-	}
-
 	public static <T> HandlerFunction<T> cast(HandlerFunction<?> handlerFunction) {
 		return (HandlerFunction<T>) handlerFunction;
 	}

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultResponseBuilder.java

@@ -24,19 +24,28 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
+import java.util.Locale;
 import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.Conventions;
-import org.springframework.core.io.Resource;
 import org.springframework.http.CacheControl;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
-import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+import org.springframework.web.reactive.result.view.ViewResolver;
+import org.springframework.web.server.ServerWebExchange;
 
 /**
  * Default {@link Response.BodyBuilder} implementation.
@@ -132,53 +141,46 @@ class DefaultResponseBuilder implements Response.BodyBuilder {
 
 	@Override
 	public Response<Void> build() {
-		return DefaultResponses.empty(this.statusCode, this.headers);
+		return body(BodyPopulator.of(
+				(response, configuration) -> response.setComplete(),
+				() -> null));
 	}
 
 	@Override
 	public <T extends Publisher<Void>> Response<T> build(T voidPublisher) {
 		Assert.notNull(voidPublisher, "'voidPublisher' must not be null");
-		return DefaultResponses.empty(this.statusCode, this.headers, voidPublisher);
+		return body(BodyPopulator.of(
+				(response, configuration) -> Flux.from(voidPublisher).then(response.setComplete()),
+				() -> null));
 	}
 
 	@Override
-	public <T> Response<T> body(T body) {
+	public <T> Response<T> body(BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer,
-		Assert.notNull(body, "'body' must not be null");
+			Supplier<T> supplier) {
-		return DefaultResponses.body(this.statusCode, this.headers, body);
+		return body(BodyPopulator.of(writer, supplier));
 	}
 
 	@Override
-	public <T, S extends Publisher<T>> Response<S> stream(S publisher, Class<T> elementClass) {
+	public <T> Response<T> body(BodyPopulator<T> populator) {
-		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(populator, "'populator' must not be null");
-		Assert.notNull(elementClass, "'elementClass' must not be null");
+		return new BodyPopulatorResponse<T>(this.statusCode, this.headers, populator);
-		return DefaultResponses.stream(this.statusCode, this.headers, publisher, elementClass);
-	}
-
-	@Override
-	public Response<Resource> resource(Resource resource) {
-		Assert.notNull(resource, "'resource' must not be null");
-		return DefaultResponses.resource(this.statusCode, this.headers, resource);
-	}
-
-	@Override
-	public <T, S extends Publisher<ServerSentEvent<T>>> Response<S> sse(S eventsPublisher) {
-		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
-		return DefaultResponses.sse(this.statusCode, this.headers, eventsPublisher);
-	}
-
-	@Override
-	public <T, S extends Publisher<T>> Response<S> sse(S eventsPublisher, Class<T> eventClass) {
-		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
-		Assert.notNull(eventClass, "'eventClass' must not be null");
-		return DefaultResponses.sse(this.statusCode, this.headers, eventsPublisher, eventClass);
 	}
 
 	@Override
 	public Response<Rendering> render(String name, Object... modelAttributes) {
-		Map<String, Object> modelMap = Arrays.stream(modelAttributes)
+		Assert.hasLength(name, "'name' must not be empty");
-				.filter(o -> !isEmptyCollection(o))
+		return render(name, toModelMap(modelAttributes));
-				.collect(Collectors.toMap(Conventions::getVariableName, o -> o));
+	}
-		return DefaultResponses.render(this.statusCode, this.headers, name, modelMap);
+
+	private static Map<String, Object> toModelMap(Object[] modelAttributes) {
+		if (!ObjectUtils.isEmpty(modelAttributes)) {
+			return Arrays.stream(modelAttributes)
+					.filter(o -> !isEmptyCollection(o))
+					.collect(Collectors.toMap(Conventions::getVariableName, o -> o));
+		}
+		else {
+			return null;
+		}
 	}
 
 	private static boolean isEmptyCollection(Object o) {
@@ -192,7 +194,122 @@ class DefaultResponseBuilder implements Response.BodyBuilder {
 		if (model != null) {
 			modelMap.putAll(model);
 		}
-		return DefaultResponses.render(this.statusCode, this.headers, name, modelMap);
+		return new RenderingResponse(this.statusCode, this.headers, name, modelMap);
 	}
 
+
+	private static abstract class AbstractResponse<T> implements Response<T> {
+
+		private final int statusCode;
+
+		private final HttpHeaders headers;
+
+
+		protected AbstractResponse(int statusCode, HttpHeaders headers) {
+			this.statusCode = statusCode;
+			this.headers = HttpHeaders.readOnlyHttpHeaders(headers);
+		}
+
+		@Override
+		public final HttpStatus statusCode() {
+			return HttpStatus.valueOf(this.statusCode);
+		}
+
+		@Override
+		public final HttpHeaders headers() {
+			return this.headers;
+		}
+
+		protected void writeStatusAndHeaders(ServerHttpResponse response) {
+			response.setStatusCode(HttpStatus.valueOf(this.statusCode));
+			HttpHeaders responseHeaders = response.getHeaders();
+
+			if (!this.headers.isEmpty()) {
+				this.headers.entrySet().stream()
+						.filter(entry -> !responseHeaders.containsKey(entry.getKey()))
+						.forEach(entry -> responseHeaders
+								.put(entry.getKey(), entry.getValue()));
+			}
+		}
+	}
+
+	private static final class BodyPopulatorResponse<T> extends AbstractResponse<T> {
+
+		private final BodyPopulator<T> populator;
+
+
+		public BodyPopulatorResponse(
+				int statusCode, HttpHeaders headers, BodyPopulator<T> populator) {
+			super(statusCode, headers);
+			this.populator = populator;
+		}
+
+		@Override
+		public T body() {
+			return this.populator.supplier().get();
+		}
+
+		@Override
+		public Mono<Void> writeTo(ServerWebExchange exchange, Configuration configuration) {
+			ServerHttpResponse response = exchange.getResponse();
+			writeStatusAndHeaders(response);
+			return this.populator.writer().apply(response, configuration);
+		}
+
+	}
+
+
+	private static final class RenderingResponse extends AbstractResponse<Rendering> {
+
+		private final String name;
+
+		private final Map<String, Object> model;
+
+		private final Rendering rendering;
+
+		public RenderingResponse(int statusCode, HttpHeaders headers, String name,
+				Map<String, Object> model) {
+			super(statusCode, headers);
+			this.name = name;
+			this.model = model;
+			this.rendering = new DefaultRendering();
+		}
+
+		@Override
+		public Rendering body() {
+			return this.rendering;
+		}
+
+		@Override
+		public Mono<Void> writeTo(ServerWebExchange exchange, Configuration configuration) {
+			ServerHttpResponse response = exchange.getResponse();
+			writeStatusAndHeaders(response);
+			MediaType contentType = exchange.getResponse().getHeaders().getContentType();
+			Locale locale = Locale.ENGLISH; // TODO: resolve locale
+			Stream<ViewResolver> viewResolverStream = configuration.viewResolvers().get();
+			return Flux.fromStream(viewResolverStream)
+					.concatMap(viewResolver -> viewResolver.resolveViewName(this.name, locale))
+					.next()
+					.otherwiseIfEmpty(Mono.error(new IllegalArgumentException("Could not resolve view with name '" +
+							this.name +"'")))
+					.then(view -> view.render(this.model, contentType, exchange));
+		}
+
+		private class DefaultRendering implements Rendering {
+
+			@Override
+			public String name() {
+				return name;
+			}
+
+			@Override
+			public Map<String, Object> model() {
+				return model;
+			}
+		}
+
+
+	}
+
+
 }

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/DefaultResponses.java

@@ -1,260 +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.reactive.function;
-
-import java.util.Collections;
-import java.util.Locale;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.function.Supplier;
-import java.util.stream.Stream;
-
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import org.springframework.core.ResolvableType;
-import org.springframework.core.io.Resource;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.MediaType;
-import org.springframework.http.codec.HttpMessageWriter;
-import org.springframework.http.codec.ResourceHttpMessageWriter;
-import org.springframework.http.codec.ServerSentEvent;
-import org.springframework.http.codec.ServerSentEventHttpMessageWriter;
-import org.springframework.http.codec.json.Jackson2JsonEncoder;
-import org.springframework.http.server.reactive.ServerHttpResponse;
-import org.springframework.util.ClassUtils;
-import org.springframework.web.reactive.result.view.ViewResolver;
-import org.springframework.web.server.ServerWebExchange;
-
-/**
- * @author Arjen Poutsma
- * @since 5.0
- */
-abstract class DefaultResponses {
-
-	private static final ResolvableType RESOURCE_TYPE = ResolvableType.forClass(Resource.class);
-
-	private static final ResolvableType SERVER_SIDE_EVENT_TYPE =
-			ResolvableType.forClass(ServerSentEvent.class);
-
-	private static final boolean jackson2Present =
-			ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper",
-					DefaultResponses.class.getClassLoader()) &&
-					ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator",
-							DefaultResponses.class.getClassLoader());
-
-
-	public static Response<Void> empty(int statusCode, HttpHeaders headers) {
-		return new DefaultResponse<>(statusCode, headers, null,
-				exchange -> exchange.getResponse().setComplete()
-		);
-	}
-
-	public static <T extends Publisher<Void>> Response<T> empty(int statusCode, HttpHeaders headers,
-			T voidPublisher) {
-		return new DefaultResponse<T>(statusCode, headers, voidPublisher,
-				exchange -> Flux.from(voidPublisher)
-						.then(exchange.getResponse().setComplete()));
-	}
-
-	public static Response<Resource> resource(int statusCode, HttpHeaders headers,
-			Resource resource) {
-		return new DefaultResponse<>(statusCode, headers, resource,
-				exchange -> {
-					ResourceHttpMessageWriter messageWriter = new ResourceHttpMessageWriter();
-					MediaType contentType = exchange.getResponse().getHeaders().getContentType();
-					return messageWriter
-							.write(Mono.just(resource), RESOURCE_TYPE, contentType,
-									exchange.getResponse(), Collections.emptyMap());
-
-				});
-	}
-
-	public static <T> Response<T> body(int statusCode, HttpHeaders headers, T body) {
-		return new DefaultResponse<T>(statusCode, headers, body,
-				exchange -> writeWithMessageWriters(exchange, Mono.just(body),
-						ResolvableType.forInstance(body)));
-	}
-
-	public static <S extends Publisher<T>, T> Response<S> stream(int statusCode,
-			HttpHeaders headers, S publisher,
-			Class<T> elementClass) {
-		return new DefaultResponse<S>(statusCode, headers, publisher,
-				exchange -> writeWithMessageWriters(exchange, publisher,
-						ResolvableType.forClass(elementClass)));
-	}
-
-	public static <T, S extends Publisher<ServerSentEvent<T>>> Response<S> sse(int statusCode,
-			HttpHeaders headers, S eventsPublisher) {
-		return new DefaultResponse<S>(statusCode, headers, eventsPublisher,
-				exchange -> {
-					ServerSentEventHttpMessageWriter messageWriter =
-							serverSentEventHttpMessageWriter();
-					MediaType contentType = exchange.getResponse().getHeaders().getContentType();
-					return messageWriter
-							.write(eventsPublisher, SERVER_SIDE_EVENT_TYPE, contentType, exchange.getResponse(), Collections
-									.emptyMap());
-				});
-	}
-
-	public static <S extends Publisher<T>, T> Response<S> sse(int statusCode, HttpHeaders headers,
-			S eventsPublisher,
-			Class<T> eventClass) {
-		return new DefaultResponse<S>(statusCode, headers, eventsPublisher,
-				exchange -> {
-					ServerSentEventHttpMessageWriter messageWriter =
-							serverSentEventHttpMessageWriter();
-					MediaType contentType = exchange.getResponse().getHeaders().getContentType();
-					return messageWriter
-							.write(eventsPublisher, ResolvableType.forClass(eventClass), contentType,
-									exchange.getResponse(), Collections.emptyMap());
-				});
-	}
-
-	public static Response<Rendering> render(int statusCode, HttpHeaders headers, String name,
-			Map<String, Object> modelMap) {
-		Rendering defaultRendering = new DefaultRendering(name, modelMap);
-		return new DefaultResponse<>(statusCode, headers, defaultRendering,
-				exchange -> {
-					MediaType contentType = exchange.getResponse().getHeaders().getContentType();
-					Locale locale = Locale.ENGLISH; // TODO: resolve locale
-					Stream<ViewResolver> viewResolverStream = configuration(exchange).viewResolvers().get();
-					return Flux.fromStream(viewResolverStream)
-							.concatMap(viewResolver -> viewResolver.resolveViewName(name, locale))
-							.next()
-							.otherwiseIfEmpty(Mono.error(new IllegalArgumentException("Could not resolve view with name '" + name +"'")))
-							.then(view -> view.render(modelMap, contentType, exchange));
-
-
-				});
-	}
-
-	private static ServerSentEventHttpMessageWriter serverSentEventHttpMessageWriter() {
-		return jackson2Present ? new ServerSentEventHttpMessageWriter(
-				Collections.singletonList(new Jackson2JsonEncoder())) :
-				new ServerSentEventHttpMessageWriter();
-	}
-
-	private static <T> Mono<Void> writeWithMessageWriters(ServerWebExchange exchange,
-			Publisher<T> body,
-			ResolvableType bodyType) {
-
-		// TODO: use ContentNegotiatingResultHandlerSupport
-		MediaType contentType = exchange.getResponse().getHeaders().getContentType();
-		ServerHttpResponse response = exchange.getResponse();
-		Stream<HttpMessageWriter<?>> messageWriterStream = configuration(exchange).messageWriters().get();
-		return messageWriterStream
-				.filter(messageWriter -> messageWriter.canWrite(bodyType, contentType, Collections
-						.emptyMap()))
-				.findFirst()
-				.map(CastingUtils::cast)
-				.map(messageWriter -> messageWriter.write(body, bodyType, contentType, response, Collections
-						.emptyMap()))
-				.orElseGet(() -> {
-					response.setStatusCode(HttpStatus.NOT_ACCEPTABLE);
-					return response.setComplete();
-				});
-	}
-
-	private static Configuration configuration(ServerWebExchange exchange) {
-		return exchange.<Configuration>getAttribute(
-				RoutingFunctions.CONFIGURATION_ATTRIBUTE)
-				.orElseThrow(() -> new IllegalStateException(
-						"Could not find Configuration in ServerWebExchange"));
-	}
-
-
-	private static final class DefaultResponse<T> implements Response<T> {
-
-		private final int statusCode;
-
-		private final HttpHeaders headers;
-
-		private final T body;
-
-		private final Function<ServerWebExchange, Mono<Void>> writingFunction;
-
-
-		public DefaultResponse(
-				int statusCode, HttpHeaders headers, T body,
-				Function<ServerWebExchange, Mono<Void>> writingFunction) {
-			this.statusCode = statusCode;
-			this.headers = HttpHeaders.readOnlyHttpHeaders(headers);
-			this.body = body;
-			this.writingFunction = writingFunction;
-		}
-
-		@Override
-		public HttpStatus statusCode() {
-			return HttpStatus.valueOf(this.statusCode);
-		}
-
-		@Override
-		public HttpHeaders headers() {
-			return this.headers;
-		}
-
-		@Override
-		public T body() {
-			return this.body;
-		}
-
-		@Override
-		public Mono<Void> writeTo(ServerWebExchange exchange) {
-			writeStatusAndHeaders(exchange);
-			return this.writingFunction.apply(exchange);
-		}
-
-		private void writeStatusAndHeaders(ServerWebExchange exchange) {
-			ServerHttpResponse response = exchange.getResponse();
-			response.setStatusCode(HttpStatus.valueOf(this.statusCode));
-			HttpHeaders responseHeaders = response.getHeaders();
-
-			if (!this.headers.isEmpty()) {
-				this.headers.entrySet().stream()
-						.filter(entry -> !responseHeaders.containsKey(entry.getKey()))
-						.forEach(entry -> responseHeaders
-								.put(entry.getKey(), entry.getValue()));
-			}
-		}
-	}
-
-	private static class DefaultRendering implements Rendering {
-
-		private final String name;
-
-		private final Map<String, Object> model;
-
-
-		public DefaultRendering(String name, Map<String, Object> model) {
-			this.name = name;
-			this.model = model;
-		}
-
-		@Override
-		public String name() {
-			return this.name;
-		}
-
-		@Override
-		public Map<String, Object> model() {
-			return this.model;
-		}
-	}
-}

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/HttpMessageExtractor.java

@@ -0,0 +1,39 @@
+/*
+ * 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.reactive.function;
+
+import org.springframework.http.ReactiveHttpInputMessage;
+
+/**
+ * Contract to extract the content of a raw {@link ReactiveHttpInputMessage} decoding
+ * the request body and using a target composition API.
+ *
+ * @author Brian Clozel
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+@FunctionalInterface
+public interface HttpMessageExtractor<T, R extends ReactiveHttpInputMessage> {
+
+	/**
+	 * Extract content from the response body
+	 * @param message the raw HTTP message
+	 * @return the extracted content
+	 */
+	T extract(R message);
+
+}

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/Response.java

@@ -21,17 +21,18 @@ import java.time.ZonedDateTime;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 
-import org.springframework.core.io.Resource;
 import org.springframework.http.CacheControl;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
-import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.util.Assert;
 import org.springframework.web.server.ServerWebExchange;
 
@@ -170,7 +171,7 @@ public interface Response<T> {
 	 * @param exchange the web exchange to write to
 	 * @return {@code Mono<Void>} to indicate when request handling is complete
 	 */
-	Mono<Void> writeTo(ServerWebExchange exchange);
+	Mono<Void> writeTo(ServerWebExchange exchange, Configuration configuration);
 
 
 	/**
@@ -306,55 +307,23 @@ public interface Response<T> {
 		 */
 		BodyBuilder contentType(MediaType contentType);
 
-//		<T> Response<T> body(BodyPopulator<T> populator);
-
 		/**
-		 * Set the body of the response to the given object and return it.
+		 * Write the body to the given {@code BodyPopulator} and return it.
-		 *
+		 * @param writer a function that writes the body to the {@code ServerHttpResponse}
-		 * @param body the body of the response
+		 * @param supplier a function that returns the body instance
+		 * @param <T> the type contained in the body
 		 * @return the built response
 		 */
-		<T> Response<T> body(T body);
+		<T> Response<T> body(BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer,
+				Supplier<T> supplier);
 
 		/**
-		 * Set the body of the response to the given {@link Publisher} and return it.
+		 * Set the body of the response to the given {@code BodyPopulator} and return it.
-		 * @param publisher the publisher to stream to the response body
+		 * @param populator the {@code BodyPopulator} that writes to the response
-		 * @param elementClass the class of elements contained in the publisher
+		 * @param <T> the type contained in the body
-		 * @param <T> the type of the elements contained in the publisher
 		 * @return the built response
 		 */
-		<T, S extends Publisher<T>> Response<S> stream(S publisher, Class<T> elementClass);
+		<T> Response<T> body(BodyPopulator<T> populator);
-		// ResolvableType
-
-		/**
-		 * Set the body of the response to the given {@link Resource} and return it.
-		 * If the resource can be resolved to a {@linkplain Resource#getFile() file}, it will be copied using
-		 * <a href="https://en.wikipedia.org/wiki/Zero-copy">zero-copy</a>
-		 *
-		 * @param resource the resource to write to the response
-		 * @return the built response
-		 */
-		Response<Resource> resource(Resource resource);
-
-		/**
-		 * Set the body of the response to the given {@link ServerSentEvent} publisher and return it.
-		 * @param eventsPublisher the {@link ServerSentEvent} publisher to stream to the response body
-		 * @param <T> the type of the elements contained in the {@link ServerSentEvent}
-		 * @return the built response
-		 * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
-		 */
-		<T, S extends Publisher<ServerSentEvent<T>>> Response<S> sse(S eventsPublisher);
-
-		/**
-		 * Set the body of the response to the given Server-Sent Event {@link Publisher} and return it.
-		 * @param eventsPublisher the publisher to stream to the response body as Server-Sent Events
-		 * @param eventClass the class of event contained in the publisher
-		 * @param <T> the type of the elements contained in the publisher
-		 * @return the built response
-		 * @see <a href="https://www.w3.org/TR/eventsource/">Server-Sent Events W3C recommendation</a>
-		 */
-		// remove?
-		<T, S extends Publisher<T>> Response<S> sse(S eventsPublisher, Class<T> eventClass);
 
 		/**
 		 * Render the template with the given {@code name} using the given {@code modelAttributes}.

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/RoutingFunctions.java

@@ -151,7 +151,7 @@ public abstract class RoutingFunctions {
 
 			HandlerFunction<?> handlerFunction = routingFunction.route(request).orElse(notFound());
 			Response<?> response = handlerFunction.handle(request);
-			return response.writeTo(exchange);
+			return response.writeTo(exchange, configuration);
 		});
 	}
 

spring-web-reactive/src/main/java/org/springframework/web/reactive/function/support/ResponseResultHandler.java

@@ -18,8 +18,10 @@ package org.springframework.web.reactive.function.support;
 
 import reactor.core.publisher.Mono;
 
+import org.springframework.util.Assert;
 import org.springframework.web.reactive.HandlerResult;
 import org.springframework.web.reactive.HandlerResultHandler;
+import org.springframework.web.reactive.function.Configuration;
 import org.springframework.web.reactive.function.Response;
 import org.springframework.web.server.ServerWebExchange;
 
@@ -31,6 +33,17 @@ import org.springframework.web.server.ServerWebExchange;
  */
 public class ResponseResultHandler implements HandlerResultHandler {
 
+	private final Configuration configuration;
+
+	public ResponseResultHandler() {
+		this(Configuration.defaultBuilder().build());
+	}
+
+	public ResponseResultHandler(Configuration configuration) {
+		Assert.notNull(configuration, "'configuration' must not be null");
+		this.configuration = configuration;
+	}
+
 	@Override
 	public boolean supports(HandlerResult result) {
 		Object returnValue = result.getReturnValue().orElse(null);
@@ -41,6 +54,6 @@ public class ResponseResultHandler implements HandlerResultHandler {
 	public Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result) {
 		Response<?> response = (Response<?>) result.getReturnValue().orElseThrow(
 				IllegalStateException::new);
-		return response.writeTo(exchange);
+		return response.writeTo(exchange, this.configuration);
 	}
 }

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/ConfigurationTests.java

@@ -47,7 +47,7 @@ public class ConfigurationTests {
 		applicationContext.registerSingleton("messageReader", DummyMessageReader.class);
 		applicationContext.refresh();
 
-		Configuration configuration = Configuration.toConfiguration(applicationContext);
+		Configuration configuration = Configuration.applicationContext(applicationContext).build();
 		assertTrue(configuration.messageReaders().get()
 				.allMatch(r -> r instanceof DummyMessageReader));
 		assertTrue(configuration.messageWriters().get()

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/DefaultResponseBuilderTests.java

@@ -17,24 +17,22 @@
 package org.springframework.web.reactive.function;
 
 import java.net.URI;
+import java.nio.ByteBuffer;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Set;
+import java.util.function.BiFunction;
 import java.util.function.Supplier;
-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.core.codec.CharSequenceEncoder;
-import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.Resource;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.CacheControl;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
@@ -42,7 +40,7 @@ import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.EncoderHttpMessageWriter;
 import org.springframework.http.codec.HttpMessageWriter;
-import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
 import org.springframework.web.reactive.result.view.View;
@@ -51,7 +49,11 @@ import org.springframework.web.server.ServerWebExchange;
 import org.springframework.web.server.adapter.DefaultServerWebExchange;
 import org.springframework.web.server.session.MockWebSessionManager;
 
-import static org.junit.Assert.*;
+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.junit.Assert.assertSame;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -188,8 +190,9 @@ public class DefaultResponseBuilderTests {
 		ServerWebExchange exchange = mock(ServerWebExchange.class);
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		when(exchange.getResponse()).thenReturn(response);
+		Configuration configuration = mock(Configuration.class);
 
-		result.writeTo(exchange).block();
+		result.writeTo(exchange, configuration).block();
 		assertEquals(201, response.getStatusCode().value());
 		assertEquals("MyValue", response.getHeaders().getFirst("MyKey"));
 		assertNull(response.getBody());
@@ -204,15 +207,27 @@ public class DefaultResponseBuilderTests {
 		ServerWebExchange exchange = mock(ServerWebExchange.class);
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		when(exchange.getResponse()).thenReturn(response);
+		Configuration configuration = mock(Configuration.class);
 
-		result.writeTo(exchange).block();
+		result.writeTo(exchange, configuration).block();
 		assertNull(response.getBody());
 	}
 
+	@SuppressWarnings("rawtypes")
 	@Test
-	public void body() throws Exception {
+	public void bodyPopulator() throws Exception {
 		String body = "foo";
-		Response<String> result = Response.ok().body(body);
+		Supplier<String> supplier = () -> body;
+		BiFunction<ServerHttpResponse, Configuration, Mono<Void>> writer =
+				(response, configuration) -> {
+					byte[] bodyBytes = body.getBytes(UTF_8);
+					ByteBuffer byteBuffer = ByteBuffer.wrap(bodyBytes);
+					DataBuffer buffer = new DefaultDataBufferFactory().wrap(byteBuffer);
+
+					return response.writeWith(Mono.just(buffer));
+				};
+
+		Response<String> result = Response.ok().body(writer, supplier);
 		assertEquals(body, result.body());
 
 		MockServerHttpRequest request =
@@ -224,14 +239,15 @@ public class DefaultResponseBuilderTests {
 		List<HttpMessageWriter<?>> messageWriters = new ArrayList<>();
 		messageWriters.add(new EncoderHttpMessageWriter<CharSequence>(new CharSequenceEncoder()));
 
-		Configuration mockConfig = mock(Configuration.class);
+		Configuration configuration = mock(Configuration.class);
-		when(mockConfig.messageWriters()).thenReturn(messageWriters::stream);
+		when(configuration.messageWriters()).thenReturn(messageWriters::stream);
-		exchange.getAttributes().put(RoutingFunctions.CONFIGURATION_ATTRIBUTE, mockConfig);
 
-		result.writeTo(exchange).block();
+		result.writeTo(exchange, configuration).block();
 		assertNotNull(response.getBody());
 	}
 
+	/*
+
 	@Test
 	public void bodyNotAcceptable() throws Exception {
 		String body = "foo";
@@ -247,11 +263,10 @@ public class DefaultResponseBuilderTests {
 		List<HttpMessageWriter<?>> messageWriters = new ArrayList<>();
 		messageWriters.add(new EncoderHttpMessageWriter<CharSequence>(new CharSequenceEncoder()));
 
-		Configuration mockConfig = mock(Configuration.class);
+		Configuration configuration = mock(Configuration.class);
-		when(mockConfig.messageWriters()).thenReturn(messageWriters::stream);
+		when(configuration.messageWriters()).thenReturn(messageWriters::stream);
-		exchange.getAttributes().put(RoutingFunctions.CONFIGURATION_ATTRIBUTE, mockConfig);
 
-		result.writeTo(exchange).block();
+		result.writeTo(exchange, configuration).block();
 		assertEquals(HttpStatus.NOT_ACCEPTABLE, response.getStatusCode());
 	}
 
@@ -304,6 +319,7 @@ public class DefaultResponseBuilderTests {
 		result.writeTo(exchange).block();
 		assertNotNull(response.getBodyWithFlush());
 	}
+	*/
 
 	@Test
 	public void render() throws Exception {
@@ -326,9 +342,8 @@ public class DefaultResponseBuilderTests {
 
 		Configuration mockConfig = mock(Configuration.class);
 		when(mockConfig.viewResolvers()).thenReturn(viewResolvers::stream);
-		exchange.getAttributes().put(RoutingFunctions.CONFIGURATION_ATTRIBUTE, mockConfig);
 
-		result.writeTo(exchange).block();
+		result.writeTo(exchange, mockConfig).block();
 	}
 
 	@Test

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/DispatcherHandlerIntegrationTests.java

@@ -50,6 +50,7 @@ import org.springframework.web.reactive.result.view.ViewResolver;
 import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
 
 import static org.junit.Assert.assertEquals;
+import static org.springframework.web.reactive.function.BodyPopulators.ofPublisher;
 import static org.springframework.web.reactive.function.RoutingFunctions.route;
 
 /**
@@ -155,13 +156,13 @@ public class DispatcherHandlerIntegrationTests extends AbstractHttpHandlerIntegr
 
 		public Response<Publisher<Person>> mono(Request request) {
 			Person person = new Person("John");
-			return Response.ok().stream(Mono.just(person), Person.class);
+			return Response.ok().body(ofPublisher(Mono.just(person), Person.class));
 		}
 
 		public Response<Publisher<Person>> flux(Request request) {
 			Person person1 = new Person("John");
 			Person person2 = new Person("Jane");
-			return Response.ok().stream(Flux.just(person1, person2), Person.class);
+			return Response.ok().body(ofPublisher(Flux.just(person1, person2), Person.class));
 		}
 
 	}

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/PublisherHandlerFunctionIntegrationTests.java

@@ -33,6 +33,7 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.web.client.RestTemplate;
 
 import static org.junit.Assert.assertEquals;
+import static org.springframework.web.reactive.function.BodyPopulators.ofPublisher;
 import static org.springframework.web.reactive.function.RequestPredicates.GET;
 import static org.springframework.web.reactive.function.RequestPredicates.POST;
 import static org.springframework.web.reactive.function.RoutingFunctions.route;
@@ -97,18 +98,18 @@ public class PublisherHandlerFunctionIntegrationTests
 
 		public Response<Publisher<Person>> mono(Request request) {
 			Person person = new Person("John");
-			return Response.ok().stream(Mono.just(person), Person.class);
+			return Response.ok().body(ofPublisher(Mono.just(person), Person.class));
 		}
 
 		public Response<Publisher<Person>> postMono(Request request) {
 			Mono<Person> personMono = request.body().convertToMono(Person.class);
-			return Response.ok().stream(personMono, Person.class);
+			return Response.ok().body(ofPublisher(personMono, Person.class));
 		}
 
 		public Response<Publisher<Person>> flux(Request request) {
 			Person person1 = new Person("John");
 			Person person2 = new Person("Jane");
-			return Response.ok().stream(Flux.just(person1, person2), Person.class);
+			return Response.ok().body(ofPublisher(Flux.just(person1, person2), Person.class));
 		}
 
 	}

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/RoutingFunctionTests.java

@@ -20,7 +20,10 @@ import java.util.Optional;
 
 import org.junit.Test;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.web.reactive.function.BodyPopulators.ofObject;
 
 /**
  * @author Arjen Poutsma
@@ -45,7 +48,7 @@ public class RoutingFunctionTests {
 
 	@Test
 	public void and() throws Exception {
-		HandlerFunction<String> handlerFunction = request -> Response.ok().body("42");
+		HandlerFunction<String> handlerFunction = request -> Response.ok().body(ofObject("42"));
 		RoutingFunction<Void> routingFunction1 = request -> Optional.empty();
 		RoutingFunction<String> routingFunction2 = request -> Optional.of(handlerFunction);
 
@@ -60,13 +63,13 @@ public class RoutingFunctionTests {
 
 	@Test
 	public void filter() throws Exception {
-		HandlerFunction<String> handlerFunction = request -> Response.ok().body("42");
+		HandlerFunction<String> handlerFunction = request -> Response.ok().body(ofObject("42"));
 		RoutingFunction<String> routingFunction = request -> Optional.of(handlerFunction);
 
 		FilterFunction<String, Integer> filterFunction = (request, next) -> {
 			Response<String> response = next.handle(request);
 			int i = Integer.parseInt(response.body());
-			return Response.ok().body(i);
+			return Response.ok().body(ofObject(i));
 		};
 		RoutingFunction<Integer> result = routingFunction.filter(filterFunction);
 		assertNotNull(result);

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/RoutingFunctionsTests.java

@@ -17,21 +17,12 @@
 package org.springframework.web.reactive.function;
 
 import java.util.Collections;
-import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 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.HttpMethod;
-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 org.springframework.http.server.reactive.HttpHandler;
@@ -40,14 +31,20 @@ import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse
 import org.springframework.web.reactive.result.view.ViewResolver;
 import org.springframework.web.server.ServerWebExchange;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
-import static org.mockito.Mockito.*;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * @author Arjen Poutsma
  */
 @SuppressWarnings("unchecked")
-public class RouterTests {
+public class RoutingFunctionsTests {
 
 	@Test
 	public void routeMatch() throws Exception {
@@ -115,9 +112,17 @@ public class RouterTests {
 
 	@Test
 	public void toHttpHandler() throws Exception {
+		Configuration configuration = mock(Configuration.class);
+		when(configuration.messageReaders()).thenReturn(
+				() -> Collections.<HttpMessageReader<?>>emptyList().stream());
+		when(configuration.messageWriters()).thenReturn(
+				() -> Collections.<HttpMessageWriter<?>>emptyList().stream());
+		when(configuration.viewResolvers()).thenReturn(
+				() -> Collections.<ViewResolver>emptyList().stream());
+
 		Request request = mock(Request.class);
 		Response response = mock(Response.class);
-		when(response.writeTo(any(ServerWebExchange.class))).thenReturn(Mono.empty());
+		when(response.writeTo(any(ServerWebExchange.class), eq(configuration))).thenReturn(Mono.empty());
 
 		HandlerFunction handlerFunction = mock(HandlerFunction.class);
 		when(handlerFunction.handle(any(Request.class))).thenReturn(response);
@@ -128,13 +133,6 @@ public class RouterTests {
 		RequestPredicate requestPredicate = mock(RequestPredicate.class);
 		when(requestPredicate.test(request)).thenReturn(false);
 
-		Configuration configuration = mock(Configuration.class);
-		when(configuration.messageReaders()).thenReturn(
-				() -> Collections.<HttpMessageReader<?>>emptyList().stream());
-		when(configuration.messageWriters()).thenReturn(
-				() -> Collections.<HttpMessageWriter<?>>emptyList().stream());
-		when(configuration.viewResolvers()).thenReturn(
-				() -> Collections.<ViewResolver>emptyList().stream());
 
 		HttpHandler result = RoutingFunctions.toHttpHandler(routingFunction, configuration);
 		assertNotNull(result);

spring-web-reactive/src/test/java/org/springframework/web/reactive/function/SseHandlerFunctionIntegrationTests.java

@@ -32,6 +32,7 @@ import org.springframework.web.client.reactive.WebClient;
 
 import static org.springframework.web.client.reactive.ClientWebRequestBuilders.get;
 import static org.springframework.web.client.reactive.ResponseExtractors.bodyStream;
+import static org.springframework.web.reactive.function.BodyPopulators.ofServerSentEvents;
 import static org.springframework.web.reactive.function.RoutingFunctions.route;
 
 /**
@@ -111,13 +112,13 @@ public class SseHandlerFunctionIntegrationTests
 
 		public Response<Publisher<String>> string(Request request) {
 			Flux<String> flux = Flux.interval(Duration.ofMillis(100)).map(l -> "foo " + l).take(2);
-			return Response.ok().sse(flux, String.class);
+			return Response.ok().body(ofServerSentEvents(flux, String.class));
 		}
 
 		public Response<Publisher<Person>> person(Request request) {
 			Flux<Person> flux = Flux.interval(Duration.ofMillis(100))
 					.map(l -> new Person("foo " + l)).take(2);
-			return Response.ok().sse(flux, Person.class);
+			return Response.ok().body(ofServerSentEvents(flux, Person.class));
 		}
 
 		public Response<Publisher<ServerSentEvent<String>>> sse(Request request) {
@@ -127,7 +128,7 @@ public class SseHandlerFunctionIntegrationTests
 							.comment("bar")
 							.build()).take(2);
 
-			return Response.ok().sse(flux);
+			return Response.ok().body(ofServerSentEvents(flux));
 		}
 	}