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

Split HttpMessageConverter into ~Reader and ~Writer

Browse Source
This commit is contained in:
Rossen Stoyanchev
2016-07-18 21:54:51 -04:00
parent dca80788d4
commit 4b92bf2af1
39 changed files with 712 additions and 561 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-web/src/main/java/org/springframework/http/codec/SseEventEncoder.java
View File

@@ -90,8 +90,7 @@ public class SseEventEncoder extends AbstractEncoder<Object> {
Object data = event.getData();
Flux<DataBuffer> dataBuffer = Flux.empty();
MediaType mediaType = (event.getMediaType() == null ?
MediaType.ALL : event.getMediaType());
MediaType mediaType = (event.getMediaType() == null ? MediaType.ALL : event.getMediaType());
if (data != null) {
sb.append("data:");
if (data instanceof String) {

92
spring-web/src/main/java/org/springframework/http/converter/reactive/DecoderHttpMessageReader.java Normal file
View File

@@ -0,0 +1,92 @@
/*
* 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.converter.reactive;
import java.util.Collections;
import java.util.List;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpInputMessage;
/**
* Implementation of the {@link HttpMessageReader} interface that delegates to
* a {@link Decoder}.
*
* @author Arjen Poutsma
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @since 5.0
*/
public class DecoderHttpMessageReader<T> implements HttpMessageReader<T> {
private final Decoder<T> decoder;
private final List<MediaType> readableMediaTypes;
/**
* Create a {@code CodecHttpMessageConverter} with the given {@link Decoder}.
* @param decoder the decoder to use
*/
public DecoderHttpMessageReader(Decoder<T> decoder) {
this.decoder = decoder;
this.readableMediaTypes = decoder != null ?
MediaType.toMediaTypes(decoder.getDecodableMimeTypes()) :
Collections.emptyList();
}
@Override
public boolean canRead(ResolvableType type, MediaType mediaType) {
return this.decoder != null && this.decoder.canDecode(type, mediaType);
}
@Override
public List<MediaType> getReadableMediaTypes() {
return this.readableMediaTypes;
}
@Override
public Flux<T> read(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
if (this.decoder == null) {
return Flux.error(new IllegalStateException("No decoder set"));
}
MediaType contentType = getContentType(inputMessage);
return this.decoder.decode(inputMessage.getBody(), type, contentType);
}
@Override
public Mono<T> readMono(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
if (this.decoder == null) {
return Mono.error(new IllegalStateException("No decoder set"));
}
MediaType contentType = getContentType(inputMessage);
return this.decoder.decodeToMono(inputMessage.getBody(), type, contentType);
}
private MediaType getContentType(ReactiveHttpInputMessage inputMessage) {
MediaType contentType = inputMessage.getHeaders().getContentType();
return (contentType != null ? contentType : MediaType.APPLICATION_OCTET_STREAM);
}
}

77
spring-web/src/main/java/org/springframework/http/converter/reactive/CodecHttpMessageConverter.java → spring-web/src/main/java/org/springframework/http/converter/reactive/EncoderHttpMessageWriter.java
View File

@@ -24,119 +24,52 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpInputMessage;
import org.springframework.http.ReactiveHttpOutputMessage;
/**
* Implementation of the {@link HttpMessageConverter} interface that delegates to
* {@link Encoder} and {@link Decoder}.
* Implementation of the {@link HttpMessageWriter} interface that delegates to
* an {@link Encoder}.
*
* @author Arjen Poutsma
* @author Sebastien Deleuze
* @author Rossen Stoyanchev
* @since 5.0
*/
public class CodecHttpMessageConverter<T> implements HttpMessageConverter<T> {
public class EncoderHttpMessageWriter<T> implements HttpMessageWriter<T> {
private final Encoder<T> encoder;
private final Decoder<T> decoder;
private final List<MediaType> readableMediaTypes;
private final List<MediaType> writableMediaTypes;
/**
* Create a {@code CodecHttpMessageConverter} with the given {@link Encoder}. When
* using this constructor, all read-related methods will in {@code false} or an
* {@link IllegalStateException}.
* Create a {@code CodecHttpMessageConverter} with the given {@link Encoder}.
* @param encoder the encoder to use
*/
public CodecHttpMessageConverter(Encoder<T> encoder) {
this(encoder, null);
}
/**
* Create a {@code CodecHttpMessageConverter} with the given {@link Decoder}. When
* using this constructor, all write-related methods will in {@code false} or an
* {@link IllegalStateException}.
* @param decoder the decoder to use
*/
public CodecHttpMessageConverter(Decoder<T> decoder) {
this(null, decoder);
}
/**
* Create a {@code CodecHttpMessageConverter} with the given {@link Encoder} and
* {@link Decoder}.
* @param encoder the encoder to use, can be {@code null}
* @param decoder the decoder to use, can be {@code null}
*/
public CodecHttpMessageConverter(Encoder<T> encoder, Decoder<T> decoder) {
public EncoderHttpMessageWriter(Encoder<T> encoder) {
this.encoder = encoder;
this.decoder = decoder;
this.readableMediaTypes = decoder != null ?
MediaType.toMediaTypes(decoder.getDecodableMimeTypes()) :
Collections.emptyList();
this.writableMediaTypes = encoder != null ?
MediaType.toMediaTypes(encoder.getEncodableMimeTypes()) :
Collections.emptyList();
}
@Override
public boolean canRead(ResolvableType type, MediaType mediaType) {
return this.decoder != null && this.decoder.canDecode(type, mediaType);
}
@Override
public boolean canWrite(ResolvableType type, MediaType mediaType) {
return this.encoder != null && this.encoder.canEncode(type, mediaType);
}
@Override
public List<MediaType> getReadableMediaTypes() {
return this.readableMediaTypes;
}
@Override
public List<MediaType> getWritableMediaTypes() {
return this.writableMediaTypes;
}
@Override
public Flux<T> read(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
if (this.decoder == null) {
return Flux.error(new IllegalStateException("No decoder set"));
}
MediaType contentType = getContentType(inputMessage);
return this.decoder.decode(inputMessage.getBody(), type, contentType);
}
@Override
public Mono<T> readMono(ResolvableType type, ReactiveHttpInputMessage inputMessage) {
if (this.decoder == null) {
return Mono.error(new IllegalStateException("No decoder set"));
}
MediaType contentType = getContentType(inputMessage);
return this.decoder.decodeToMono(inputMessage.getBody(), type, contentType);
}
private MediaType getContentType(ReactiveHttpInputMessage inputMessage) {
MediaType contentType = inputMessage.getHeaders().getContentType();
return (contentType != null ? contentType : MediaType.APPLICATION_OCTET_STREAM);
}
@Override
public Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType type,
MediaType contentType, ReactiveHttpOutputMessage outputMessage) {

34
spring-web/src/main/java/org/springframework/http/converter/reactive/HttpMessageConverter.java → spring-web/src/main/java/org/springframework/http/converter/reactive/HttpMessageReader.java
View File

@@ -28,13 +28,14 @@ import org.springframework.http.ReactiveHttpInputMessage;
import org.springframework.http.ReactiveHttpOutputMessage;
/**
* Strategy interface that specifies a converter that can convert from and to HTTP
* requests and responses.
* Strategy interface that specifies a reader that can convert from the HTTP
* request body from a stream of bytes to Objects.
*
* @author Rossen Stoyanchev
* @author Arjen Poutsma
* @since 5.0
*/
public interface HttpMessageConverter<T> {
public interface HttpMessageReader<T> {
/**
* Indicates whether the given class can be read by this converter.
@@ -71,31 +72,4 @@ public interface HttpMessageConverter<T> {
*/
Mono<T> readMono(ResolvableType type, ReactiveHttpInputMessage inputMessage);
/**
* Indicates whether the given class can be written by this converter.
* @param type the class to test for writability
* @param mediaType the media type to write, can be {@code null} if not specified.
* Typically the value of an {@code Accept} header.
* @return {@code true} if writable; {@code false} otherwise
*/
boolean canWrite(ResolvableType type, MediaType mediaType);
/**
* Return the list of {@link MediaType} objects that can be written by this converter.
* @return the list of supported readable media types
*/
List<MediaType> getWritableMediaTypes();
/**
* Write an given object to the given output message.
* @param inputStream the input stream to write
* @param type the stream element type to process.
* @param contentType the content type to use when writing. May be {@code null} to
* indicate that the default content type of the converter must be used.
* @param outputMessage the message to write to
* @return the converted {@link Mono} of object
*/
Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType type,
MediaType contentType, ReactiveHttpOutputMessage outputMessage);
}

65
spring-web/src/main/java/org/springframework/http/converter/reactive/HttpMessageWriter.java Normal file
View File

@@ -0,0 +1,65 @@
/*
* 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.converter.reactive;
import java.util.List;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpOutputMessage;
/**
* Strategy interface that specifies a converter that can convert a stream of
* Objects to a stream of bytes to be written to the HTTP response body.
*
* @author Rossen Stoyanchev
* @author Arjen Poutsma
* @since 5.0
*/
public interface HttpMessageWriter<T> {
/**
* Indicates whether the given class can be written by this converter.
* @param type the class to test for writability
* @param mediaType the media type to write, can be {@code null} if not specified.
* Typically the value of an {@code Accept} header.
* @return {@code true} if writable; {@code false} otherwise
*/
boolean canWrite(ResolvableType type, MediaType mediaType);
/**
* Return the list of {@link MediaType} objects that can be written by this converter.
* @return the list of supported readable media types
*/
List<MediaType> getWritableMediaTypes();
/**
* Write an given object to the given output message.
* @param inputStream the input stream to write
* @param type the stream element type to process.
* @param contentType the content type to use when writing. May be {@code null} to
* indicate that the default content type of the converter must be used.
* @param outputMessage the message to write to
* @return the converted {@link Mono} of object
*/
Mono<Void> write(Publisher<? extends T> inputStream, ResolvableType type,
MediaType contentType, ReactiveHttpOutputMessage outputMessage);
}

31
spring-web/src/main/java/org/springframework/http/converter/reactive/ResourceHttpMessageConverter.java → spring-web/src/main/java/org/springframework/http/converter/reactive/ResourceHttpMessageWriter.java
View File

@@ -37,27 +37,31 @@ import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ResourceUtils;
/**
* Implementation of {@link HttpMessageConverter} that can read and write
* {@link Resource Resources} and supports byte range requests.
* Implementation of {@link HttpMessageWriter} that can write
* {@link Resource Resources}.
*
* <p>For a Resource reader simply use {@link ResourceDecoder} wrapped with
* {@link DecoderHttpMessageReader}.
*
* @author Arjen Poutsma
* @since 5.0
*/
public class ResourceHttpMessageConverter extends CodecHttpMessageConverter<Resource> {
public class ResourceHttpMessageWriter extends EncoderHttpMessageWriter<Resource> {
public ResourceHttpMessageConverter() {
super(new ResourceEncoder(), new ResourceDecoder());
public ResourceHttpMessageWriter() {
super(new ResourceEncoder());
}
public ResourceHttpMessageConverter(int bufferSize) {
super(new ResourceEncoder(bufferSize), new ResourceDecoder());
public ResourceHttpMessageWriter(int bufferSize) {
super(new ResourceEncoder(bufferSize));
}
@Override
public Mono<Void> write(Publisher<? extends Resource> inputStream,
ResolvableType type, MediaType contentType,
ReactiveHttpOutputMessage outputMessage) {
public Mono<Void> write(Publisher<? extends Resource> inputStream, ResolvableType type,
MediaType contentType, ReactiveHttpOutputMessage outputMessage) {
return Mono.from(Flux.from(inputStream).
take(1).
concatMap(resource -> {
@@ -68,11 +72,9 @@ public class ResourceHttpMessageConverter extends CodecHttpMessageConverter<Reso
}));
}
protected void addHeaders(HttpHeaders headers, Resource resource,
MediaType contentType) {
protected void addHeaders(HttpHeaders headers, Resource resource, MediaType contentType) {
if (headers.getContentType() == null) {
if (contentType == null ||
!contentType.isConcrete() ||
if (contentType == null || !contentType.isConcrete() ||
MediaType.APPLICATION_OCTET_STREAM.equals(contentType)) {
contentType = MimeTypeUtils.getMimeType(resource.getFilename()).
map(MediaType::toMediaType).
@@ -87,6 +89,7 @@ public class ResourceHttpMessageConverter extends CodecHttpMessageConverter<Reso
private Mono<Void> writeContent(Resource resource, ResolvableType type,
MediaType contentType, ReactiveHttpOutputMessage outputMessage) {
if (outputMessage instanceof ZeroCopyHttpOutputMessage) {
Optional<File> file = getFile(resource);
if (file.isPresent()) {

6
spring-web/src/main/java/org/springframework/web/client/reactive/BodyExtractor.java
View File

@@ -19,7 +19,7 @@ package org.springframework.web.client.reactive;
import java.util.List;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Contract to extract the content of a raw {@link ClientHttpResponse} decoding
@@ -36,9 +36,9 @@ public interface BodyExtractor<T> {
/**
* Extract content from the response body
* @param clientResponse the raw HTTP response
* @param messageConverters the message converters that decode the response body
* @param messageReaders the message readers that decode the response body
* @return the relevant content
*/
T extract(ClientHttpResponse clientResponse, List<HttpMessageConverter<?>> messageConverters);
T extract(ClientHttpResponse clientResponse, List<HttpMessageReader<?>> messageReaders);
}

8
spring-web/src/main/java/org/springframework/web/client/reactive/DefaultResponseErrorHandler.java
View File

@@ -20,7 +20,7 @@ import java.util.List;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Default implementation of the {@link ResponseErrorHandler} interface
@@ -33,13 +33,13 @@ import org.springframework.http.converter.reactive.HttpMessageConverter;
public class DefaultResponseErrorHandler implements ResponseErrorHandler {
@Override
public void handleError(ClientHttpResponse response, List<HttpMessageConverter<?>> messageConverters) {
public void handleError(ClientHttpResponse response, List<HttpMessageReader<?>> messageReaders) {
HttpStatus responseStatus = response.getStatusCode();
if (responseStatus.is4xxClientError()) {
throw new WebClientErrorException(response, messageConverters);
throw new WebClientErrorException(response, messageReaders);
}
if (responseStatus.is5xxServerError()) {
throw new WebServerErrorException(response, messageConverters);
throw new WebServerErrorException(response, messageReaders);
}
}
}

4
spring-web/src/main/java/org/springframework/web/client/reactive/ResponseErrorHandler.java
View File

@@ -19,7 +19,7 @@ package org.springframework.web.client.reactive;
import java.util.List;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Strategy interface used by the {@link WebClient} to handle errors in
@@ -37,6 +37,6 @@ public interface ResponseErrorHandler {
* {@link ClientHttpResponse#getStatusCode() HttpStatus} of the response and
* throw {@link WebClientException}s in case of errors.
*/
void handleError(ClientHttpResponse response, List<HttpMessageConverter<?>> messageConverters);
void handleError(ClientHttpResponse response, List<HttpMessageReader<?>> messageReaders);
}

32
spring-web/src/main/java/org/springframework/web/client/reactive/ResponseExtractors.java
View File

@@ -26,7 +26,7 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Static factory methods for {@link ResponseExtractor} and {@link BodyExtractor},
@@ -47,9 +47,9 @@ public class ResponseExtractors {
public static <T> ResponseExtractor<Mono<T>> body(ResolvableType bodyType) {
return (clientResponse, webClientConfig) -> (Mono<T>) clientResponse
.doOnNext(response -> webClientConfig.getResponseErrorHandler()
.handleError(response, webClientConfig.getMessageConverters()))
.handleError(response, webClientConfig.getMessageReaders()))
.flatMap(resp -> decodeResponseBodyAsMono(resp, bodyType,
webClientConfig.getMessageConverters()))
webClientConfig.getMessageReaders()))
.next();
}
@@ -86,8 +86,8 @@ public class ResponseExtractors {
public static <T> ResponseExtractor<Flux<T>> bodyStream(ResolvableType bodyType) {
return (clientResponse, webClientConfig) -> clientResponse
.doOnNext(response -> webClientConfig.getResponseErrorHandler()
.handleError(response, webClientConfig.getMessageConverters()))
.flatMap(resp -> decodeResponseBody(resp, bodyType, webClientConfig.getMessageConverters()));
.handleError(response, webClientConfig.getMessageReaders()))
.flatMap(resp -> decodeResponseBody(resp, bodyType, webClientConfig.getMessageReaders()));
}
/**
@@ -127,7 +127,7 @@ public class ResponseExtractors {
return (clientResponse, webClientConfig) -> clientResponse.then(response ->
Mono.when(
decodeResponseBodyAsMono(response, bodyType,
webClientConfig.getMessageConverters()).defaultIfEmpty(EMPTY_BODY),
webClientConfig.getMessageReaders()).defaultIfEmpty(EMPTY_BODY),
Mono.just(response.getHeaders()),
Mono.just(response.getStatusCode()))
).map(tuple -> {
@@ -153,7 +153,7 @@ public class ResponseExtractors {
public static <T> ResponseExtractor<Mono<ResponseEntity<Flux<T>>>> responseStream(ResolvableType type) {
return (clientResponse, webClientConfig) -> clientResponse
.map(response -> new ResponseEntity<>(
decodeResponseBody(response, type, webClientConfig.getMessageConverters()),
decodeResponseBody(response, type, webClientConfig.getMessageReaders()),
response.getHeaders(), response.getStatusCode()));
}
@@ -175,26 +175,26 @@ public class ResponseExtractors {
@SuppressWarnings("unchecked")
protected static <T> Flux<T> decodeResponseBody(ClientHttpResponse response,
ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
ResolvableType responseType, List<HttpMessageReader<?>> messageReaders) {
MediaType contentType = response.getHeaders().getContentType();
HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
return (Flux<T>) converter.read(responseType, response);
HttpMessageReader<?> reader = resolveMessageReader(messageReaders, responseType, contentType);
return (Flux<T>) reader.read(responseType, response);
}
@SuppressWarnings("unchecked")
protected static <T> Mono<T> decodeResponseBodyAsMono(ClientHttpResponse response,
ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
ResolvableType responseType, List<HttpMessageReader<?>> messageReaders) {
MediaType contentType = response.getHeaders().getContentType();
HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
return (Mono<T>) converter.readMono(responseType, response);
HttpMessageReader<?> reader = resolveMessageReader(messageReaders, responseType, contentType);
return (Mono<T>) reader.readMono(responseType, response);
}
protected static HttpMessageConverter<?> resolveConverter(
List<HttpMessageConverter<?>> messageConverters, ResolvableType responseType, MediaType contentType) {
protected static HttpMessageReader<?> resolveMessageReader(List<HttpMessageReader<?>> messageReaders,
ResolvableType responseType, MediaType contentType) {
return messageConverters.stream()
return messageReaders.stream()
.filter(e -> e.canRead(responseType, contentType))
.findFirst()
.orElseThrow(() ->

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

@@ -25,30 +25,30 @@ import java.util.function.Function;
import java.util.logging.Level;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.core.codec.ByteBufferDecoder;
import org.springframework.core.codec.ByteBufferEncoder;
import org.springframework.http.codec.json.JacksonJsonDecoder;
import org.springframework.http.codec.json.JacksonJsonEncoder;
import org.springframework.core.codec.ResourceDecoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.codec.StringEncoder;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ClientHttpRequest;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.codec.json.JacksonJsonDecoder;
import org.springframework.http.codec.json.JacksonJsonEncoder;
import org.springframework.http.codec.xml.Jaxb2Decoder;
import org.springframework.http.codec.xml.Jaxb2Encoder;
import org.springframework.http.converter.reactive.CodecHttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.ResourceHttpMessageConverter;
import org.springframework.http.converter.reactive.DecoderHttpMessageReader;
import org.springframework.http.converter.reactive.EncoderHttpMessageWriter;
import org.springframework.http.converter.reactive.HttpMessageReader;
import org.springframework.http.converter.reactive.HttpMessageWriter;
import org.springframework.http.converter.reactive.ResourceHttpMessageWriter;
import org.springframework.util.ClassUtils;
import reactor.core.publisher.Mono;
/**
* Reactive Web client supporting the HTTP/1.1 protocol
*
@@ -96,6 +96,7 @@ public final class WebClient {
private final DefaultWebClientConfig webClientConfig;
/**
* Create a {@code WebClient} instance, using the {@link ClientHttpConnector}
* implementation given as an argument to drive the underlying
@@ -114,39 +115,54 @@ public final class WebClient {
public WebClient(ClientHttpConnector clientHttpConnector) {
this.clientHttpConnector = clientHttpConnector;
this.webClientConfig = new DefaultWebClientConfig();
List<HttpMessageConverter<?>> converters = new ArrayList<>();
addDefaultHttpMessageConverters(converters);
this.webClientConfig.setMessageConverters(converters);
this.webClientConfig.setResponseErrorHandler(new DefaultResponseErrorHandler());
}
/**
* Adds default HTTP message converters
* Adds default HTTP message readers.
*/
protected final void addDefaultHttpMessageConverters(
List<HttpMessageConverter<?>> converters) {
converters.add(converter(new ByteBufferEncoder(), new ByteBufferDecoder()));
converters.add(converter(new StringEncoder(), new StringDecoder()));
converters.add(new ResourceHttpMessageConverter());
protected final void addDefaultHttpMessageReaders(List<HttpMessageReader<?>> messageReaders) {
messageReaders.add(new DecoderHttpMessageReader<>(new ByteBufferDecoder()));
messageReaders.add(new DecoderHttpMessageReader<>(new StringDecoder(false)));
messageReaders.add(new DecoderHttpMessageReader<>(new ResourceDecoder()));
if (jaxb2Present) {
converters.add(converter(new Jaxb2Encoder(), new Jaxb2Decoder()));
messageReaders.add(new DecoderHttpMessageReader<>(new Jaxb2Decoder()));
}
if (jackson2Present) {
converters.add(converter(new JacksonJsonEncoder(), new JacksonJsonDecoder()));
messageReaders.add(new DecoderHttpMessageReader<>(new JacksonJsonDecoder()));
}
}
private static <T> HttpMessageConverter<T> converter(Encoder<T> encoder,
Decoder<T> decoder) {
return new CodecHttpMessageConverter<>(encoder, decoder);
}
/**
* Set the list of {@link HttpMessageConverter}s to use for encoding and decoding HTTP
* messages
* Adds default HTTP message writers.
*/
public void setMessageConverters(List<HttpMessageConverter<?>> messageConverters) {
this.webClientConfig.setMessageConverters(messageConverters);
protected final void addDefaultHttpMessageWriters(List<HttpMessageWriter<?>> messageWriters) {
messageWriters.add(new EncoderHttpMessageWriter<>(new ByteBufferEncoder()));
messageWriters.add(new EncoderHttpMessageWriter<>(new StringEncoder()));
messageWriters.add(new ResourceHttpMessageWriter());
if (jaxb2Present) {
messageWriters.add(new EncoderHttpMessageWriter<>(new Jaxb2Encoder()));
}
if (jackson2Present) {
messageWriters.add(new EncoderHttpMessageWriter<>(new JacksonJsonEncoder()));
}
}
/**
* Set the list of {@link HttpMessageReader}s to use for decoding the HTTP
* response body.
*/
public void setMessageReaders(List<HttpMessageReader<?>> messageReaders) {
this.webClientConfig.setMessageReaders(messageReaders);
}
/**
* Set the list of {@link HttpMessageWriter}s to use for encoding the HTTP
* request body.
*/
public void setMessageWriters(List<HttpMessageWriter<?>> messageWrters) {
this.webClientConfig.setMessageWriters(messageWrters);
}
/**
@@ -163,7 +179,7 @@ public final class WebClient {
* Requesting from the exposed {@code Flux} will result in:
* <ul>
* <li>building the actual HTTP request using the provided {@code ClientWebRequestBuilder}</li>
* <li>encoding the HTTP request body with the configured {@code HttpMessageConverter}s</li>
* <li>encoding the HTTP request body with the configured {@code HttpMessageWriter}s</li>
* <li>returning the response with a publisher of the body</li>
* </ul>
*/
@@ -191,17 +207,36 @@ public final class WebClient {
protected class DefaultWebClientConfig implements WebClientConfig {
private List<HttpMessageConverter<?>> messageConverters;
private List<HttpMessageReader<?>> messageReaders;
private List<HttpMessageWriter<?>> messageWriters;
private ResponseErrorHandler responseErrorHandler;
@Override
public List<HttpMessageConverter<?>> getMessageConverters() {
return messageConverters;
public DefaultWebClientConfig() {
this.messageReaders = new ArrayList<>();
addDefaultHttpMessageReaders(this.messageReaders);
this.messageWriters = new ArrayList<>();
addDefaultHttpMessageWriters(this.messageWriters);
}
public void setMessageConverters(List<HttpMessageConverter<?>> messageConverters) {
this.messageConverters = messageConverters;
@Override
public List<HttpMessageReader<?>> getMessageReaders() {
return this.messageReaders;
}
public void setMessageReaders(List<HttpMessageReader<?>> messageReaders) {
this.messageReaders = messageReaders;
}
@Override
public List<HttpMessageWriter<?>> getMessageWriters() {
return this.messageWriters;
}
public void setMessageWriters(List<HttpMessageWriter<?>> messageWriters) {
this.messageWriters = messageWriters;
}
@Override
@@ -218,10 +253,12 @@ public final class WebClient {
private final ClientWebRequest clientWebRequest;
public DefaultRequestCallback(ClientWebRequest clientWebRequest) {
this.clientWebRequest = clientWebRequest;
}
@Override
public Mono<Void> apply(ClientHttpRequest clientHttpRequest) {
clientHttpRequest.getHeaders().putAll(this.clientWebRequest.getHttpHeaders());
@@ -229,13 +266,13 @@ public final class WebClient {
clientHttpRequest.getHeaders().setAccept(
Collections.singletonList(MediaType.ALL));
}
clientWebRequest.getCookies().values()
this.clientWebRequest.getCookies().values()
.stream().flatMap(cookies -> cookies.stream())
.forEach(cookie -> clientHttpRequest.getCookies().add(cookie.getName(), cookie));
if (this.clientWebRequest.getBody() != null) {
return writeRequestBody(this.clientWebRequest.getBody(),
this.clientWebRequest.getElementType(),
clientHttpRequest, webClientConfig.getMessageConverters());
clientHttpRequest, WebClient.this.webClientConfig.getMessageWriters());
}
else {
return clientHttpRequest.setComplete();
@@ -245,22 +282,22 @@ public final class WebClient {
@SuppressWarnings({ "unchecked", "rawtypes" })
protected Mono<Void> writeRequestBody(Publisher<?> content,
ResolvableType requestType, ClientHttpRequest request,
List<HttpMessageConverter<?>> messageConverters) {
List<HttpMessageWriter<?>> messageWriters) {
MediaType contentType = request.getHeaders().getContentType();
Optional<HttpMessageConverter<?>> converter = resolveConverter(messageConverters, requestType, contentType);
if (!converter.isPresent()) {
Optional<HttpMessageWriter<?>> messageWriter = resolveWriter(messageWriters, requestType, contentType);
if (!messageWriter.isPresent()) {
return Mono.error(new IllegalStateException(
"Could not encode request body of type '" + contentType
+ "' with target type '" + requestType.toString() + "'"));
}
return converter.get().write((Publisher) content, requestType, contentType, request);
return messageWriter.get().write((Publisher) content, requestType, contentType, request);
}
protected Optional<HttpMessageConverter<?>> resolveConverter(
List<HttpMessageConverter<?>> messageConverters, ResolvableType type,
MediaType mediaType) {
return messageConverters.stream().filter(e -> e.canWrite(type, mediaType)).findFirst();
protected Optional<HttpMessageWriter<?>> resolveWriter(List<HttpMessageWriter<?>> messageWriters,
ResolvableType type, MediaType mediaType) {
return messageWriters.stream().filter(e -> e.canWrite(type, mediaType)).findFirst();
}
}

12
spring-web/src/main/java/org/springframework/web/client/reactive/WebClientConfig.java
View File

@@ -18,7 +18,8 @@ package org.springframework.web.client.reactive;
import java.util.List;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
import org.springframework.http.converter.reactive.HttpMessageWriter;
/**
* Interface that makes the {@link WebClient} configuration information
@@ -30,9 +31,14 @@ import org.springframework.http.converter.reactive.HttpMessageConverter;
public interface WebClientConfig {
/**
* Return the message converters that can help encoding/decoding the HTTP message body
* Return the message readers that can help decoding the HTTP response body
*/
List<HttpMessageConverter<?>> getMessageConverters();
List<HttpMessageReader<?>> getMessageReaders();
/**
* Return the message writers that can help encode the HTTP request body
*/
List<HttpMessageWriter<?>> getMessageWriters();
/**
* Return the configured {@link ResponseErrorHandler}

10
spring-web/src/main/java/org/springframework/web/client/reactive/WebClientErrorException.java
View File

@@ -19,7 +19,7 @@ package org.springframework.web.client.reactive;
import java.util.List;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Exception thrown when an HTTP 4xx is received.
@@ -33,14 +33,14 @@ public class WebClientErrorException extends WebClientResponseException {
/**
* Construct a new instance of {@code HttpClientErrorException} based on a
* {@link ClientHttpResponse} and {@link HttpMessageConverter}s to optionally
* {@link ClientHttpResponse} and {@link HttpMessageReader}s to optionally
* help decoding the response body
*
* @param response the HTTP response
* @param converters the message converters that may decode the HTTP response body
* @param messageReaders the message converters that may decode the HTTP response body
*/
public WebClientErrorException(ClientHttpResponse response, List<HttpMessageConverter<?>> converters) {
super(initMessage(response), response, converters);
public WebClientErrorException(ClientHttpResponse response, List<HttpMessageReader<?>> messageReaders) {
super(initMessage(response), response, messageReaders);
}
private static String initMessage(ClientHttpResponse response) {

12
spring-web/src/main/java/org/springframework/web/client/reactive/WebClientResponseException.java
View File

@@ -21,7 +21,7 @@ import java.util.List;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Base class for exceptions associated with specific HTTP client response
@@ -35,20 +35,20 @@ public class WebClientResponseException extends WebClientException {
private final ClientHttpResponse clientResponse;
private final List<HttpMessageConverter<?>> messageConverters;
private final List<HttpMessageReader<?>> messageReaders;
/**
* Construct a new instance of {@code WebClientResponseException} with the given response data
* @param message the given error message
* @param clientResponse the HTTP response
* @param messageConverters the message converters that maay decode the HTTP response body
* @param messageReaders the message converters that maay decode the HTTP response body
*/
public WebClientResponseException(String message, ClientHttpResponse clientResponse,
List<HttpMessageConverter<?>> messageConverters) {
List<HttpMessageReader<?>> messageReaders) {
super(message);
this.clientResponse = clientResponse;
this.messageConverters = messageConverters;
this.messageReaders = messageReaders;
}
@@ -76,6 +76,6 @@ public class WebClientResponseException extends WebClientException {
* </pre>
*/
public <T> T getResponseBody(BodyExtractor<T> extractor) {
return extractor.extract(this.clientResponse, this.messageConverters);
return extractor.extract(this.clientResponse, this.messageReaders);
}
}

11
spring-web/src/main/java/org/springframework/web/client/reactive/WebServerErrorException.java
View File

@@ -18,9 +18,8 @@ package org.springframework.web.client.reactive;
import java.util.List;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
/**
* Exception thrown when an HTTP 5xx is received.
@@ -33,13 +32,13 @@ public class WebServerErrorException extends WebClientResponseException {
/**
* Construct a new instance of {@code HttpServerErrorException} based on a
* {@link ClientHttpResponse} and {@link HttpMessageConverter}s to optionally
* {@link ClientHttpResponse} and {@link HttpMessageReader}s to optionally
* help decoding the response body
* @param response the HTTP response
* @param converters the message converters that may decode the HTTP response body
* @param messageReaders the message converters that may decode the HTTP response body
*/
public WebServerErrorException(ClientHttpResponse response, List<HttpMessageConverter<?>> converters) {
super(initMessage(response), response, converters);
public WebServerErrorException(ClientHttpResponse response, List<HttpMessageReader<?>> messageReaders) {
super(initMessage(response), response, messageReaders);
}
private static String initMessage(ClientHttpResponse response) {

28
spring-web/src/main/java/org/springframework/web/client/reactive/support/RxJava1ResponseExtractors.java
View File

@@ -29,7 +29,7 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.converter.reactive.HttpMessageConverter;
import org.springframework.http.converter.reactive.HttpMessageReader;
import org.springframework.web.client.reactive.BodyExtractor;
import org.springframework.web.client.reactive.ResponseExtractor;
import org.springframework.web.client.reactive.WebClientException;
@@ -53,8 +53,8 @@ public class RxJava1ResponseExtractors {
return (clientResponse, webClientConfig) -> (Single<T>) RxJava1Adapter
.publisherToSingle(clientResponse
.doOnNext(response -> webClientConfig.getResponseErrorHandler()
.handleError(response, webClientConfig.getMessageConverters()))
.flatMap(resp -> decodeResponseBodyAsMono(resp, bodyType, webClientConfig.getMessageConverters())));
.handleError(response, webClientConfig.getMessageReaders()))
.flatMap(resp -> decodeResponseBodyAsMono(resp, bodyType, webClientConfig.getMessageReaders())));
}
/**
@@ -95,8 +95,8 @@ public class RxJava1ResponseExtractors {
return (clientResponse, webClientConfig) -> RxJava1Adapter
.publisherToObservable(clientResponse
.doOnNext(response -> webClientConfig.getResponseErrorHandler()
.handleError(response, webClientConfig.getMessageConverters()))
.flatMap(resp -> decodeResponseBody(resp, bodyType, webClientConfig.getMessageConverters())));
.handleError(response, webClientConfig.getMessageReaders()))
.flatMap(resp -> decodeResponseBody(resp, bodyType, webClientConfig.getMessageReaders())));
}
/**
@@ -139,7 +139,7 @@ public class RxJava1ResponseExtractors {
RxJava1Adapter.publisherToSingle(clientResponse
.then(response ->
Mono.when(
decodeResponseBody(response, bodyType, webClientConfig.getMessageConverters()).next(),
decodeResponseBody(response, bodyType, webClientConfig.getMessageReaders()).next(),
Mono.just(response.getHeaders()),
Mono.just(response.getStatusCode())))
.map(tuple ->
@@ -174,7 +174,7 @@ public class RxJava1ResponseExtractors {
return (clientResponse, webClientConfig) -> RxJava1Adapter.publisherToSingle(clientResponse
.map(response -> new ResponseEntity<>(
RxJava1Adapter
.publisherToObservable(decodeResponseBody(response, bodyType, webClientConfig.getMessageConverters())),
.publisherToObservable(decodeResponseBody(response, bodyType, webClientConfig.getMessageReaders())),
response.getHeaders(),
response.getStatusCode())));
}
@@ -189,26 +189,26 @@ public class RxJava1ResponseExtractors {
@SuppressWarnings("unchecked")
protected static <T> Flux<T> decodeResponseBody(ClientHttpResponse response,
ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
ResolvableType responseType, List<HttpMessageReader<?>> messageReaders) {
MediaType contentType = response.getHeaders().getContentType();
HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
HttpMessageReader<?> converter = resolveMessageReader(messageReaders, responseType, contentType);
return (Flux<T>) converter.read(responseType, response);
}
@SuppressWarnings("unchecked")
protected static <T> Mono<T> decodeResponseBodyAsMono(ClientHttpResponse response,
ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
ResolvableType responseType, List<HttpMessageReader<?>> messageReaders) {
MediaType contentType = response.getHeaders().getContentType();
HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
HttpMessageReader<?> converter = resolveMessageReader(messageReaders, responseType, contentType);
return (Mono<T>) converter.readMono(responseType, response);
}
protected static HttpMessageConverter<?> resolveConverter(
List<HttpMessageConverter<?>> messageConverters, ResolvableType responseType, MediaType contentType) {
protected static HttpMessageReader<?> resolveMessageReader(List<HttpMessageReader<?>> messageReaders,
ResolvableType responseType, MediaType contentType) {
return messageConverters.stream()
return messageReaders.stream()
.filter(e -> e.canRead(responseType, contentType))
.findFirst()
.orElseThrow(() ->