Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move Body[Inserter|Extractor] to web.reactive.function

Browse Source
This commit is contained in:
Arjen Poutsma
2016-12-15 13:20:35 +01:00
parent aac20b3fd1
commit 97558ab4de
29 changed files with 54 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,56 +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.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 input message.
* @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();
}
}

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

@@ -1,141 +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.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.core.io.buffer.DataBuffer;
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 (inputMessage, context) -> readWithMessageReaders(inputMessage, context,
elementType,
reader -> reader.read(elementType, inputMessage, Collections.emptyMap()),
Flux::error);
}
/**
* Return a {@code BodyExtractor} that returns the body of the message as a {@link Flux} of
* {@link DataBuffer}s.
* <p><strong>Note</strong> that the returned buffers should be released after usage by calling
* {@link org.springframework.core.io.buffer.DataBufferUtils#release(DataBuffer)}
* @return a {@code BodyExtractor} that returns the body
* @see ReactiveHttpInputMessage#getBody()
*/
public static BodyExtractor<Flux<DataBuffer>, ReactiveHttpInputMessage> toDataBuffers() {
return (inputMessage, context) -> inputMessage.getBody();
}
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;
}
}

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

@@ -1,59 +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.http.codec;
import java.util.function.Supplier;
import java.util.stream.Stream;
import reactor.core.publisher.Mono;
import org.springframework.http.ReactiveHttpOutputMessage;
/**
* A combination of functions that can populate a {@link ReactiveHttpOutputMessage} body.
*
* @author Arjen Poutsma
* @since 5.0
* @see BodyInserters
*/
@FunctionalInterface
public interface BodyInserter<T, M extends ReactiveHttpOutputMessage> {
/**
* Insert into the given output message.
* @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);
/**
* 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();
}
}

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

@@ -1,246 +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.http.codec;
import java.util.Collections;
import java.util.List;
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.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpOutputMessage;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.util.Assert;
/**
* 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 BodyInserter<Void, ReactiveHttpOutputMessage> EMPTY =
(response, context) -> response.setComplete();
/**
* Return an empty {@code BodyInserter} that writes nothing.
* @return an empty {@code BodyInserter}
*/
@SuppressWarnings("unchecked")
public static <T> BodyInserter<T, ReactiveHttpOutputMessage> empty() {
return (BodyInserter<T, ReactiveHttpOutputMessage>)EMPTY;
}
/**
* 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 bodyInserterFor(Mono.just(body), ResolvableType.forInstance(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 <P> the type of the {@code Publisher}
* @return a {@code BodyInserter} that writes a {@code Publisher}
*/
public static <T, P extends Publisher<T>> BodyInserter<P, ReactiveHttpOutputMessage> fromPublisher(P publisher,
Class<T> elementClass) {
Assert.notNull(publisher, "'publisher' must not be null");
Assert.notNull(elementClass, "'elementClass' must not be null");
return bodyInserterFor(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 <P> the type of the {@code Publisher}
* @return a {@code BodyInserter} that writes a {@code Publisher}
*/
public static <T, P extends Publisher<T>> BodyInserter<P, ReactiveHttpOutputMessage> fromPublisher(P publisher,
ResolvableType elementType) {
Assert.notNull(publisher, "'publisher' must not be null");
Assert.notNull(elementType, "'elementType' must not be null");
return bodyInserterFor(publisher, elementType);
}
/**
* 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 (response, context) -> {
HttpMessageWriter<Resource> messageWriter = resourceHttpMessageWriter(context);
return messageWriter.write(Mono.just(resource), RESOURCE_TYPE, null,
response, Collections.emptyMap());
};
}
private static HttpMessageWriter<Resource> resourceHttpMessageWriter(BodyInserter.Context context) {
return context.messageWriters().get()
.filter(messageWriter -> messageWriter.canWrite(RESOURCE_TYPE, null))
.findFirst()
.map(BodyInserters::<Resource>cast)
.orElseThrow(() -> new IllegalStateException(
"Could not find HttpMessageWriter that supports Resources."));
}
/**
* 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 (response, context) -> {
HttpMessageWriter<ServerSentEvent<T>> messageWriter = sseMessageWriter(context);
return messageWriter.write(eventsPublisher, SERVER_SIDE_EVENT_TYPE,
MediaType.TEXT_EVENT_STREAM, response, Collections.emptyMap());
};
}
/**
* 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 (outputMessage, context) -> {
HttpMessageWriter<T> messageWriter = sseMessageWriter(context);
return messageWriter.write(eventsPublisher, eventType,
MediaType.TEXT_EVENT_STREAM, outputMessage, Collections.emptyMap());
};
}
/**
* Return a {@code BodyInserter} that writes the given {@code Publisher<DataBuffer>} to the
* body.
* @param publisher the data buffer publisher to write
* @param <T> the type of the publisher
* @return a {@code BodyInserter} that writes directly to the body
* @see ReactiveHttpOutputMessage#writeWith(Publisher)
*/
public static <T extends Publisher<DataBuffer>> BodyInserter<T, ReactiveHttpOutputMessage> fromDataBuffers(T publisher) {
Assert.notNull(publisher, "'publisher' must not be null");
return (outputMessage, context) -> outputMessage.writeWith(publisher);
}
private static <T> HttpMessageWriter<T> sseMessageWriter(BodyInserter.Context context) {
return context.messageWriters().get()
.filter(messageWriter -> messageWriter
.canWrite(SERVER_SIDE_EVENT_TYPE, MediaType.TEXT_EVENT_STREAM))
.findFirst()
.map(BodyInserters::<T>cast)
.orElseThrow(() -> new IllegalStateException(
"Could not find HttpMessageWriter that supports " +
MediaType.TEXT_EVENT_STREAM_VALUE));
}
private static <T, P extends Publisher<?>, M extends ReactiveHttpOutputMessage> BodyInserter<T, M> bodyInserterFor(P body, ResolvableType bodyType) {
return (m, context) -> {
MediaType contentType = m.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, m, 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;
}
}