Moved BodyExtractor and BodyInserter to http.codec

This commit moves the web.reactive.function.[BodyInserter|BodyExtractor]
to http.codec, so that they can be used from the client as well.

Furthermore, it parameterized both inserter and extractor over
ReactiveHttpOutputMessage and ReactiveHttpInputMessage respectively, so
that they can be limited to only be used on the client or server.

spring-web/src/main/java/org/springframework/http/codec/BodyExtractor.java

@@ -0,0 +1,56 @@
+/*
+ * 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.http.codec;
+
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+import org.springframework.http.ReactiveHttpInputMessage;
+
+/**
+ * A function that can extract data from a {@link ReactiveHttpInputMessage} body.
+ *
+ * @param <T> the type of data to extract
+ * @author Arjen Poutsma
+ * @since 5.0
+ * @see BodyExtractors
+ */
+@FunctionalInterface
+public interface BodyExtractor<T, M extends ReactiveHttpInputMessage> {
+
+	/**
+	 * Extract from the given request.
+	 * @param inputMessage request to extract from
+	 * @param context the configuration to use
+	 * @return the extracted data
+	 */
+	T extract(M inputMessage, Context context);
+
+	/**
+	 * Defines the context used during the extraction.
+	 */
+	interface Context {
+
+		/**
+		 * Supply a {@linkplain Stream stream} of {@link HttpMessageReader}s to be used for body
+		 * extraction.
+		 * @return the stream of message readers
+		 */
+		Supplier<Stream<HttpMessageReader<?>>> messageReaders();
+	}
+
+}

spring-web/src/main/java/org/springframework/http/codec/BodyExtractors.java

@@ -0,0 +1,128 @@
+/*
+ * 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.http.codec;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+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.ResolvableType;
+import org.springframework.http.HttpMessage;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpInputMessage;
+import org.springframework.util.Assert;
+
+/**
+ * Implementations of {@link BodyExtractor} that read various bodies, such a reactive streams.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+public abstract class BodyExtractors {
+
+	/**
+	 * Return a {@code BodyExtractor} that reads into a Reactor {@link Mono}.
+	 * @param elementClass the class of element in the {@code Mono}
+	 * @param <T> the element type
+	 * @return a {@code BodyExtractor} that reads a mono
+	 */
+	public static <T> BodyExtractor<Mono<T>, ReactiveHttpInputMessage> toMono(Class<? extends T> elementClass) {
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+		return toMono(ResolvableType.forClass(elementClass));
+	}
+
+	/**
+	 * Return a {@code BodyExtractor} that reads into a Reactor {@link Mono}.
+	 * @param elementType the type of element in the {@code Mono}
+	 * @param <T> the element type
+	 * @return a {@code BodyExtractor} that reads a mono
+	 */
+	public static <T> BodyExtractor<Mono<T>, ReactiveHttpInputMessage> toMono(ResolvableType elementType) {
+		Assert.notNull(elementType, "'elementType' must not be null");
+		return (request, context) -> readWithMessageReaders(request, context,
+				elementType,
+				reader -> reader.readMono(elementType, request, Collections.emptyMap()),
+				Mono::error);
+	}
+
+	/**
+	 * Return a {@code BodyExtractor} that reads into a Reactor {@link Flux}.
+	 * @param elementClass the class of element in the {@code Flux}
+	 * @param <T> the element type
+	 * @return a {@code BodyExtractor} that reads a mono
+	 */
+	public static <T> BodyExtractor<Flux<T>, ReactiveHttpInputMessage> toFlux(Class<? extends T> elementClass) {
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+		return toFlux(ResolvableType.forClass(elementClass));
+	}
+
+	/**
+	 * Return a {@code BodyExtractor} that reads into a Reactor {@link Flux}.
+	 * @param elementType the type of element in the {@code Flux}
+	 * @param <T> the element type
+	 * @return a {@code BodyExtractor} that reads a mono
+	 */
+	public static <T> BodyExtractor<Flux<T>, ReactiveHttpInputMessage> toFlux(ResolvableType elementType) {
+		Assert.notNull(elementType, "'elementType' must not be null");
+		return (request, context) -> readWithMessageReaders(request, context,
+				elementType,
+				reader -> reader.read(elementType, request, Collections.emptyMap()),
+				Flux::error);
+	}
+
+	private static <T, S extends Publisher<T>> S readWithMessageReaders(
+			ReactiveHttpInputMessage inputMessage,
+			BodyExtractor.Context context,
+			ResolvableType elementType,
+			Function<HttpMessageReader<T>, S> readerFunction,
+			Function<Throwable, S> unsupportedError) {
+
+		MediaType contentType = contentType(inputMessage);
+		Supplier<Stream<HttpMessageReader<?>>> messageReaders = context.messageReaders();
+		return messageReaders.get()
+				.filter(r -> r.canRead(elementType, contentType))
+				.findFirst()
+				.map(BodyExtractors::<T>cast)
+				.map(readerFunction)
+				.orElseGet(() -> {
+					List<MediaType> supportedMediaTypes = messageReaders.get()
+							.flatMap(reader -> reader.getReadableMediaTypes().stream())
+							.collect(Collectors.toList());
+					UnsupportedMediaTypeException error =
+							new UnsupportedMediaTypeException(contentType, supportedMediaTypes);
+					return unsupportedError.apply(error);
+				});
+	}
+
+	private static MediaType contentType(HttpMessage message) {
+		MediaType result = message.getHeaders().getContentType();
+		return result != null ? result : MediaType.APPLICATION_OCTET_STREAM;
+	}
+
+	@SuppressWarnings("unchecked")
+	private static <T> HttpMessageReader<T> cast(HttpMessageReader<?> messageReader) {
+		return (HttpMessageReader<T>) messageReader;
+	}
+
+}

spring-web/src/main/java/org/springframework/http/codec/BodyInserter.java

@@ -0,0 +1,82 @@
+/*
+ * 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.http.codec;
+
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.util.Assert;
+
+/**
+ * A combination of functions that can populate a {@link ReactiveHttpOutputMessage} body.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+public interface BodyInserter<T, M extends ReactiveHttpOutputMessage> {
+
+	/**
+	 * Insert into the given response.
+	 * @param outputMessage the response to insert into
+	 * @param context the context to use
+	 * @return a {@code Mono} that indicates completion or error
+	 */
+	Mono<Void> insert(M outputMessage, Context context);
+
+	/**
+	 * Return the type contained in the body.
+	 * @return the type contained in the body
+	 */
+	T t();
+
+	/**
+	 * Return a new {@code BodyInserter} described by the given writer and supplier functions.
+	 * @param writer  the writer function for the new inserter
+	 * @param supplier the supplier function for the new inserter
+	 * @param <T> the type supplied and written by the inserter
+	 * @return the new {@code BodyInserter}
+	 */
+	static <T, M extends ReactiveHttpOutputMessage> BodyInserter<T, M> of(
+			BiFunction<M, Context, Mono<Void>> writer,
+			Supplier<T> supplier) {
+
+		Assert.notNull(writer, "'writer' must not be null");
+		Assert.notNull(supplier, "'supplier' must not be null");
+
+		return new BodyInserters.DefaultBodyInserter<T, M>(writer, supplier);
+	}
+
+	/**
+	 * Defines the context used during the insertion.
+	 */
+	interface Context {
+
+		/**
+		 * Supply a {@linkplain Stream stream} of {@link HttpMessageWriter}s to be used for response
+		 * body conversion.
+		 * @return the stream of message writers
+		 */
+		Supplier<Stream<HttpMessageWriter<?>>> messageWriters();
+
+	}
+
+
+}

spring-web/src/main/java/org/springframework/http/codec/BodyInserters.java

@@ -0,0 +1,258 @@
+/*
+ * 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.http.codec;
+
+import java.util.Collections;
+import java.util.List;
+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.Mono;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.Resource;
+import org.springframework.http.MediaType;
+import org.springframework.http.ReactiveHttpOutputMessage;
+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 BodyInserter} that write various bodies, such a reactive streams,
+ * server-sent events, resources, etc.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+public abstract class BodyInserters {
+
+	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",
+					BodyInserters.class.getClassLoader()) &&
+					ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator",
+							BodyInserters.class.getClassLoader());
+
+	/**
+	 * Return a {@code BodyInserter} that writes the given single object.
+	 * @param body the body of the response
+	 * @return a {@code BodyInserter} that writes a single object
+	 */
+	public static <T> BodyInserter<T, ReactiveHttpOutputMessage> fromObject(T body) {
+		Assert.notNull(body, "'body' must not be null");
+		return BodyInserter.of(
+				(response, context) -> writeWithMessageWriters(response, context,
+						Mono.just(body), ResolvableType.forInstance(body)),
+				() -> body);
+	}
+
+	/**
+	 * Return a {@code BodyInserter} 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 BodyInserter} that writes a {@code Publisher}
+	 */
+	public static <S extends Publisher<T>, T> BodyInserter<S, ReactiveHttpOutputMessage> fromPublisher(S publisher,
+			Class<T> elementClass) {
+
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+		return fromPublisher(publisher, ResolvableType.forClass(elementClass));
+	}
+
+	/**
+	 * Return a {@code BodyInserter} 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 BodyInserter} that writes a {@code Publisher}
+	 */
+	public static <S extends Publisher<T>, T> BodyInserter<S, ReactiveHttpOutputMessage> fromPublisher(S publisher,
+			ResolvableType elementType) {
+
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementType, "'elementType' must not be null");
+		return BodyInserter.of(
+				(response, context) -> writeWithMessageWriters(response, context,
+						publisher, elementType),
+				() -> publisher
+		);
+	}
+
+	/**
+	 * Return a {@code BodyInserter} 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 BodyInserter} that writes a {@code Publisher}
+	 */
+	public static <T extends Resource> BodyInserter<T, ReactiveHttpOutputMessage> fromResource(T resource) {
+		Assert.notNull(resource, "'resource' must not be null");
+		return BodyInserter.of(
+				(response, context) -> {
+					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 BodyInserter} 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 BodyInserter} 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>>> BodyInserter<S, ServerHttpResponse> fromServerSentEvents(
+			S eventsPublisher) {
+
+		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
+		return BodyInserter.of(
+				(response, context) -> {
+					ServerSentEventHttpMessageWriter messageWriter = sseMessageWriter();
+					MediaType contentType = response.getHeaders().getContentType();
+					return messageWriter.write(eventsPublisher, SERVER_SIDE_EVENT_TYPE,
+							contentType, response, Collections.emptyMap());
+				},
+				() -> eventsPublisher
+		);
+	}
+
+	/**
+	 * Return a {@code BodyInserter} 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 BodyInserter} 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>> BodyInserter<S, ServerHttpResponse> fromServerSentEvents(S eventsPublisher,
+			Class<T> eventClass) {
+
+		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
+		Assert.notNull(eventClass, "'eventClass' must not be null");
+		return fromServerSentEvents(eventsPublisher, ResolvableType.forClass(eventClass));
+	}
+
+	/**
+	 * Return a {@code BodyInserter} 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 BodyInserter} 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>> BodyInserter<S, ServerHttpResponse> fromServerSentEvents(S eventsPublisher,
+			ResolvableType eventType) {
+
+		Assert.notNull(eventsPublisher, "'eventsPublisher' must not be null");
+		Assert.notNull(eventType, "'eventType' must not be null");
+		return BodyInserter.of(
+				(response, context) -> {
+					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(ReactiveHttpOutputMessage outputMessage,
+			BodyInserter.Context context,
+			Publisher<T> body,
+			ResolvableType bodyType) {
+
+		MediaType contentType = outputMessage.getHeaders().getContentType();
+		Supplier<Stream<HttpMessageWriter<?>>> messageWriters = context.messageWriters();
+		return messageWriters.get()
+				.filter(messageWriter -> messageWriter.canWrite(bodyType, contentType))
+				.findFirst()
+				.map(BodyInserters::cast)
+				.map(messageWriter -> messageWriter
+						.write(body, bodyType, contentType, outputMessage, Collections
+								.emptyMap()))
+				.orElseGet(() -> {
+					List<MediaType> supportedMediaTypes = messageWriters.get()
+							.flatMap(reader -> reader.getWritableMediaTypes().stream())
+							.collect(Collectors.toList());
+					UnsupportedMediaTypeException error =
+							new UnsupportedMediaTypeException(contentType, supportedMediaTypes);
+					return Mono.error(error);
+				});
+	}
+
+	@SuppressWarnings("unchecked")
+	private static <T> HttpMessageWriter<T> cast(HttpMessageWriter<?> messageWriter) {
+		return (HttpMessageWriter<T>) messageWriter;
+	}
+
+	static class DefaultBodyInserter<T, M extends ReactiveHttpOutputMessage>
+			implements BodyInserter<T, M> {
+
+		private final BiFunction<M, Context, Mono<Void>> writer;
+
+		private final Supplier<T> supplier;
+
+		public DefaultBodyInserter(
+				BiFunction<M, Context, Mono<Void>> writer,
+				Supplier<T> supplier) {
+			this.writer = writer;
+			this.supplier = supplier;
+		}
+
+		@Override
+		public Mono<Void> insert(M outputMessage, Context context) {
+			return this.writer.apply(outputMessage, context);
+		}
+
+		@Override
+		public T t() {
+			return this.supplier.get();
+		}
+
+	}
+
+
+}

spring-web/src/main/java/org/springframework/http/codec/UnsupportedMediaTypeException.java

@@ -0,0 +1,74 @@
+/*
+ * 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.http.codec;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import org.springframework.core.NestedRuntimeException;
+import org.springframework.http.MediaType;
+
+/**
+ * Exception thrown to indicate that a {@code Content-Type} is not supported.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+@SuppressWarnings("serial")
+public class UnsupportedMediaTypeException extends NestedRuntimeException {
+
+	private final MediaType contentType;
+
+	private final List<MediaType> supportedMediaTypes;
+
+
+	/**
+	 * Constructor for when the specified Content-Type is invalid.
+	 */
+	public UnsupportedMediaTypeException(String reason) {
+		super(reason);
+		this.contentType = null;
+		this.supportedMediaTypes = Collections.emptyList();
+	}
+
+	/**
+	 * Constructor for when the Content-Type can be parsed but is not supported.
+	 */
+	public UnsupportedMediaTypeException(MediaType contentType, List<MediaType> supportedMediaTypes) {
+		super("Content type '" + contentType + "' not supported");
+		this.contentType = contentType;
+		this.supportedMediaTypes = Collections.unmodifiableList(supportedMediaTypes);
+	}
+
+
+	/**
+	 * Return the request Content-Type header if it was parsed successfully.
+	 */
+	public Optional<MediaType> getContentType() {
+		return Optional.ofNullable(this.contentType);
+	}
+
+	/**
+	 * Return the list of supported content types in cases when the Content-Type
+	 * header is parsed but not supported, or an empty list otherwise.
+	 */
+	public List<MediaType> getSupportedMediaTypes() {
+		return this.supportedMediaTypes;
+	}
+
+}