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

Merge messaging related codec updates

Browse Source
This commit is contained in:
Rossen Stoyanchev
2019-04-11 19:02:48 -04:00
parent 3d0ec509ab 5fc18064f2
commit e27e54317f
29 changed files with 430 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java
View File

@@ -45,6 +45,7 @@ import org.springframework.util.MimeType;
* @since 5.0
* @param <T> the element type
*/
@SuppressWarnings("deprecation")
public abstract class AbstractDataBufferDecoder<T> extends AbstractDecoder<T> {
@@ -70,8 +71,14 @@ public abstract class AbstractDataBufferDecoder<T> extends AbstractDecoder<T> {
/**
* How to decode a {@code DataBuffer} to the target element type.
* @deprecated as of 5.2, please implement
* {@link #decode(DataBuffer, ResolvableType, MimeType, Map)} instead
*/
protected abstract T decodeDataBuffer(DataBuffer buffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints);
@Deprecated
protected T decodeDataBuffer(DataBuffer buffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return decode(buffer, elementType, mimeType, hints);
}
}

4
spring-core/src/main/java/org/springframework/core/codec/ByteArrayDecoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -45,7 +45,7 @@ public class ByteArrayDecoder extends AbstractDataBufferDecoder<byte[]> {
}
@Override
protected byte[] decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType,
public byte[] decode(DataBuffer dataBuffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
byte[] result = new byte[dataBuffer.readableByteCount()];

24
spring-core/src/main/java/org/springframework/core/codec/ByteArrayEncoder.java
View File

@@ -52,15 +52,21 @@ public class ByteArrayEncoder extends AbstractEncoder<byte[]> {
DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType,
@Nullable Map<String, Object> hints) {
// The following (byte[] bytes) lambda signature declaration is necessary for Eclipse.
return Flux.from(inputStream).map((byte[] bytes) -> {
DataBuffer dataBuffer = bufferFactory.wrap(bytes);
if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
String logPrefix = Hints.getLogPrefix(hints);
logger.debug(logPrefix + "Writing " + dataBuffer.readableByteCount() + " bytes");
}
return dataBuffer;
});
// Use (byte[] bytes) for Eclipse
return Flux.from(inputStream).map((byte[] bytes) ->
encodeValue(bytes, bufferFactory, elementType, mimeType, hints));
}
@Override
public DataBuffer encodeValue(byte[] bytes, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
DataBuffer dataBuffer = bufferFactory.wrap(bytes);
if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
String logPrefix = Hints.getLogPrefix(hints);
logger.debug(logPrefix + "Writing " + dataBuffer.readableByteCount() + " bytes");
}
return dataBuffer;
}
}

4
spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -48,7 +48,7 @@ public class ByteBufferDecoder extends AbstractDataBufferDecoder<ByteBuffer> {
}
@Override
protected ByteBuffer decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType,
public ByteBuffer decode(DataBuffer dataBuffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
int byteCount = dataBuffer.readableByteCount();

24
spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -53,14 +53,20 @@ public class ByteBufferEncoder extends AbstractEncoder<ByteBuffer> {
DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType,
@Nullable Map<String, Object> hints) {
return Flux.from(inputStream).map(byteBuffer -> {
DataBuffer dataBuffer = bufferFactory.wrap(byteBuffer);
if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
String logPrefix = Hints.getLogPrefix(hints);
logger.debug(logPrefix + "Writing " + dataBuffer.readableByteCount() + " bytes");
}
return dataBuffer;
});
return Flux.from(inputStream).map(byteBuffer ->
encodeValue(byteBuffer, bufferFactory, elementType, mimeType, hints));
}
@Override
public DataBuffer encodeValue(ByteBuffer byteBuffer, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
DataBuffer dataBuffer = bufferFactory.wrap(byteBuffer);
if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
String logPrefix = Hints.getLogPrefix(hints);
logger.debug(logPrefix + "Writing " + dataBuffer.readableByteCount() + " bytes");
}
return dataBuffer;
}
}

55
spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -71,32 +71,37 @@ public final class CharSequenceEncoder extends AbstractEncoder<CharSequence> {
DataBufferFactory bufferFactory, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
Charset charset = getCharset(mimeType);
return Flux.from(inputStream).map(charSequence ->
encodeValue(charSequence, bufferFactory, elementType, mimeType, hints));
}
return Flux.from(inputStream).map(charSequence -> {
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
String formatted = LogFormatUtils.formatValue(charSequence, !traceOn);
return Hints.getLogPrefix(hints) + "Writing " + formatted;
});
@Override
public DataBuffer encodeValue(CharSequence charSequence, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
String formatted = LogFormatUtils.formatValue(charSequence, !traceOn);
return Hints.getLogPrefix(hints) + "Writing " + formatted;
});
}
boolean release = true;
Charset charset = getCharset(mimeType);
int capacity = calculateCapacity(charSequence, charset);
DataBuffer dataBuffer = bufferFactory.allocateBuffer(capacity);
try {
dataBuffer.write(charSequence, charset);
release = false;
}
catch (CoderMalfunctionError ex) {
throw new EncodingException("String encoding error: " + ex.getMessage(), ex);
}
finally {
if (release) {
DataBufferUtils.release(dataBuffer);
}
boolean release = true;
int capacity = calculateCapacity(charSequence, charset);
DataBuffer dataBuffer = bufferFactory.allocateBuffer(capacity);
try {
dataBuffer.write(charSequence, charset);
release = false;
}
catch (CoderMalfunctionError ex) {
throw new EncodingException("String encoding error: " + ex.getMessage(), ex);
}
finally {
if (release) {
DataBufferUtils.release(dataBuffer);
}
}
return dataBuffer;
});
}
return dataBuffer;
}
int calculateCapacity(CharSequence sequence, Charset charset) {

4
spring-core/src/main/java/org/springframework/core/codec/DataBufferDecoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -64,7 +64,7 @@ public class DataBufferDecoder extends AbstractDataBufferDecoder<DataBuffer> {
}
@Override
protected DataBuffer decodeDataBuffer(DataBuffer buffer, ResolvableType elementType,
public DataBuffer decode(DataBuffer buffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
if (logger.isDebugEnabled()) {

24
spring-core/src/main/java/org/springframework/core/codec/DataBufferEncoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -53,15 +53,25 @@ public class DataBufferEncoder extends AbstractEncoder<DataBuffer> {
@Nullable Map<String, Object> hints) {
Flux<DataBuffer> flux = Flux.from(inputStream);
if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
flux = flux.doOnNext(buffer -> {
String logPrefix = Hints.getLogPrefix(hints);
logger.debug(logPrefix + "Writing " + buffer.readableByteCount() + " bytes");
});
flux = flux.doOnNext(buffer -> logValue(buffer, hints));
}
return flux;
}
@Override
public DataBuffer encodeValue(DataBuffer buffer, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
logValue(buffer, hints);
}
return buffer;
}
private void logValue(DataBuffer buffer, @Nullable Map<String, Object> hints) {
String logPrefix = Hints.getLogPrefix(hints);
logger.debug(logPrefix + "Writing " + buffer.readableByteCount() + " bytes");
}
}

29
spring-core/src/main/java/org/springframework/core/codec/Decoder.java
View File

@@ -22,10 +22,12 @@ import java.util.Map;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
@@ -75,6 +77,33 @@ public interface Decoder<T> {
Mono<T> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints);
/**
* Decode a data buffer to an Object of type T. This is useful when the input
* stream consists of discrete messages (or events) and the content for each
* can be decoded on its own.
* @param buffer the {@code DataBuffer} to decode
* @param targetType the expected output type
* @param mimeType the MIME type associated with the data
* @param hints additional information about how to do encode
* @return the decoded value, possibly {@code null}
* @since 5.2
*/
@SuppressWarnings("ConstantConditions")
default T decode(DataBuffer buffer, ResolvableType targetType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
MonoProcessor<T> processor = MonoProcessor.create();
decodeToMono(Mono.just(buffer), targetType, mimeType, hints).subscribeWith(processor);
Assert.state(processor.isTerminated(), "DataBuffer decoding should have completed.");
Throwable ex = processor.getError();
if (ex != null) {
throw (ex instanceof CodecException ? (CodecException) ex :
new DecodingException("Failed to decode: " + ex.getMessage(), ex));
}
return processor.peek();
}
/**
* Return the list of MIME types this decoder supports.
*/

26
spring-core/src/main/java/org/springframework/core/codec/Encoder.java
View File

@@ -60,13 +60,35 @@ public interface Encoder<T> {
* @param elementType the expected type of elements in the input stream;
* this type must have been previously passed to the {@link #canEncode}
* method and it must have returned {@code true}.
* @param mimeType the MIME type for the output stream (optional)
* @param hints additional information about how to do encode
* @param mimeType the MIME type for the output content (optional)
* @param hints additional information about how to encode
* @return the output stream
*/
Flux<DataBuffer> encode(Publisher<? extends T> inputStream, DataBufferFactory bufferFactory,
ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints);
/**
* Encode an Object of type T to a data buffer. This is useful for scenarios
* that produce a stream of discrete messages (or events) and the
* content for each is encoded individually.
* <p>By default this method raises {@link UnsupportedOperationException}
* and it is expected that some encoders cannot produce a single buffer or
* cannot do so synchronously (e.g. encoding a {@code Resource}).
* @param value the value to be encoded
* @param bufferFactory for creating the output {@code DataBuffer}
* @param valueType the type for the value being encoded
* @param mimeType the MIME type for the output content (optional)
* @param hints additional information about how to encode
* @return the encoded content
* @since 5.2
*/
default DataBuffer encodeValue(T value, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
// It may not be possible to produce a single DataBuffer synchronously
throw new UnsupportedOperationException();
}
/**
* Return the list of mime types this encoder supports.
*/

2
spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java
View File

@@ -64,7 +64,7 @@ public class ResourceDecoder extends AbstractDataBufferDecoder<Resource> {
}
@Override
protected Resource decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType,
public Resource decode(DataBuffer dataBuffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
byte[] bytes = new byte[dataBuffer.readableByteCount()];

2
spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java
View File

@@ -202,7 +202,7 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
}
@Override
protected String decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType,
public String decode(DataBuffer dataBuffer, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
Charset charset = getCharset(mimeType);

4
spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
View File

@@ -441,6 +441,10 @@ public abstract class DataBufferUtils {
public static Mono<DataBuffer> join(Publisher<DataBuffer> dataBuffers) {
Assert.notNull(dataBuffers, "'dataBuffers' must not be null");
if (dataBuffers instanceof Mono) {
return (Mono<DataBuffer>) dataBuffers;
}
return Flux.from(dataBuffers)
.collectList()
.filter(list -> !list.isEmpty())

6
spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java
View File

@@ -80,10 +80,14 @@ public class DefaultDataBuffer implements DataBuffer {
/**
* Directly exposes the native {@code ByteBuffer} that this buffer is based on.
* Directly exposes the native {@code ByteBuffer} that this buffer is based
* on also updating the {@code ByteBuffer's} position and limit to match
* the current {@link #readPosition()} and {@link #readableByteCount()}.
* @return the wrapped byte buffer
*/
public ByteBuffer getNativeBuffer() {
this.byteBuffer.position(this.readPosition);
this.byteBuffer.limit(readableByteCount());
return this.byteBuffer;
}

10
spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolver.java
View File

@@ -232,26 +232,26 @@ public class PayloadMethodArgumentResolver implements HandlerMethodArgumentResol
if (decoder.canDecode(elementType, mimeType)) {
if (adapter != null && adapter.isMultiValue()) {
Flux<?> flux = content
.concatMap(buffer -> decoder.decode(Mono.just(buffer), elementType, mimeType, hints))
.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
.onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex)));
if (isContentRequired) {
flux = flux.switchIfEmpty(Flux.error(() -> handleMissingBody(parameter, message)));
}
if (validator != null) {
flux = flux.doOnNext(validator::accept);
flux = flux.doOnNext(validator);
}
return Mono.just(adapter.fromPublisher(flux));
}
else {
// Single-value (with or without reactive type wrapper)
Mono<?> mono = decoder
.decodeToMono(content.next(), targetType, mimeType, hints)
Mono<?> mono = content.next()
.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
.onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex)));
if (isContentRequired) {
mono = mono.switchIfEmpty(Mono.error(() -> handleMissingBody(parameter, message)));
}
if (validator != null) {
mono = mono.doOnNext(validator::accept);
mono = mono.doOnNext(validator);
}
return (adapter != null ? Mono.just(adapter.fromPublisher(mono)) : Mono.from(mono));
}

13
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/AbstractEncoderMethodReturnValueHandler.java
View File

@@ -33,7 +33,6 @@ import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Encoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -148,7 +147,7 @@ public abstract class AbstractEncoderMethodReturnValueHandler implements Handler
Encoder<?> encoder = getEncoder(elementType, mimeType);
return Flux.from((Publisher) publisher).concatMap(value ->
return Flux.from((Publisher) publisher).map(value ->
encodeValue(value, elementType, encoder, bufferFactory, mimeType, hints));
}
@@ -176,7 +175,7 @@ public abstract class AbstractEncoderMethodReturnValueHandler implements Handler
}
@SuppressWarnings("unchecked")
private <T> Mono<DataBuffer> encodeValue(
private <T> DataBuffer encodeValue(
Object element, ResolvableType elementType, @Nullable Encoder<T> encoder,
DataBufferFactory bufferFactory, @Nullable MimeType mimeType,
@Nullable Map<String, Object> hints) {
@@ -184,13 +183,11 @@ public abstract class AbstractEncoderMethodReturnValueHandler implements Handler
if (encoder == null) {
encoder = getEncoder(ResolvableType.forInstance(element), mimeType);
if (encoder == null) {
return Mono.error(new MessagingException(
"No encoder for " + elementType + ", current value type is " + element.getClass()));
throw new MessagingException(
"No encoder for " + elementType + ", current value type is " + element.getClass());
}
}
Mono<T> mono = Mono.just((T) element);
Flux<DataBuffer> dataBuffers = encoder.encode(mono, bufferFactory, elementType, mimeType, hints);
return DataBufferUtils.join(dataBuffers);
return encoder.encodeValue((T) element, bufferFactory, elementType, mimeType, hints);
}
/**

29
spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java
View File

@@ -32,7 +32,6 @@ 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.DataBufferUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
@@ -124,8 +123,10 @@ final class DefaultRSocketRequester implements RSocketRequester {
publisher = adapter.toPublisher(input);
}
else {
Mono<Payload> payloadMono = encodeValue(input, ResolvableType.forInstance(input), null)
Mono<Payload> payloadMono = Mono
.fromCallable(() -> encodeValue(input, ResolvableType.forInstance(input), null))
.map(this::firstPayload)
.doOnDiscard(Payload.class, Payload::release)
.switchIfEmpty(emptyPayload());
return new DefaultResponseSpec(payloadMono);
}
@@ -140,36 +141,36 @@ final class DefaultRSocketRequester implements RSocketRequester {
if (adapter != null && !adapter.isMultiValue()) {
Mono<Payload> payloadMono = Mono.from(publisher)
.flatMap(value -> encodeValue(value, dataType, encoder))
.map(value -> encodeValue(value, dataType, encoder))
.map(this::firstPayload)
.switchIfEmpty(emptyPayload());
return new DefaultResponseSpec(payloadMono);
}
Flux<Payload> payloadFlux = Flux.from(publisher)
.concatMap(value -> encodeValue(value, dataType, encoder))
.map(value -> encodeValue(value, dataType, encoder))
.switchOnFirst((signal, inner) -> {
DataBuffer data = signal.get();
if (data != null) {
return Flux.concat(
Mono.just(firstPayload(data)),
inner.skip(1).map(PayloadUtils::createPayload));
return Mono.fromCallable(() -> firstPayload(data))
.concatWith(inner.skip(1).map(PayloadUtils::createPayload));
}
else {
return inner.map(PayloadUtils::createPayload);
}
})
.doOnDiscard(Payload.class, Payload::release)
.switchIfEmpty(emptyPayload());
return new DefaultResponseSpec(payloadFlux);
}
@SuppressWarnings("unchecked")
private <T> Mono<DataBuffer> encodeValue(T value, ResolvableType valueType, @Nullable Encoder<?> encoder) {
private <T> DataBuffer encodeValue(T value, ResolvableType valueType, @Nullable Encoder<?> encoder) {
if (encoder == null) {
encoder = strategies.encoder(ResolvableType.forInstance(value), dataMimeType);
}
return DataBufferUtils.join(((Encoder<T>) encoder).encode(
Mono.just(value), strategies.dataBufferFactory(), valueType, dataMimeType, EMPTY_HINTS));
return ((Encoder<T>) encoder).encodeValue(
value, strategies.dataBufferFactory(), valueType, dataMimeType, EMPTY_HINTS);
}
private Payload firstPayload(DataBuffer data) {
@@ -244,8 +245,8 @@ final class DefaultRSocketRequester implements RSocketRequester {
}
Decoder<?> decoder = strategies.decoder(elementType, dataMimeType);
return (Mono<T>) decoder.decodeToMono(
payloadMono.map(this::retainDataAndReleasePayload), elementType, dataMimeType, EMPTY_HINTS);
return (Mono<T>) payloadMono.map(this::retainDataAndReleasePayload)
.map(dataBuffer -> decoder.decode(dataBuffer, elementType, dataMimeType, EMPTY_HINTS));
}
@SuppressWarnings("unchecked")
@@ -261,8 +262,8 @@ final class DefaultRSocketRequester implements RSocketRequester {
Decoder<?> decoder = strategies.decoder(elementType, dataMimeType);
return payloadFlux.map(this::retainDataAndReleasePayload).concatMap(dataBuffer ->
(Mono<T>) decoder.decodeToMono(Mono.just(dataBuffer), elementType, dataMimeType, EMPTY_HINTS));
return payloadFlux.map(this::retainDataAndReleasePayload).map(dataBuffer ->
(T) decoder.decode(dataBuffer, elementType, dataMimeType, EMPTY_HINTS));
}
private DataBuffer retainDataAndReleasePayload(Payload payload) {

2
spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandlerTests.java
View File

@@ -81,7 +81,7 @@ public class MessageMappingMessageHandlerTests {
@Test
public void handleFluxString() {
MessageMappingMessageHandler messsageHandler = initMesssageHandler();
messsageHandler.handleMessage(message("fluxString", "abc\ndef\nghi")).block(Duration.ofSeconds(5));
messsageHandler.handleMessage(message("fluxString", "abc", "def", "ghi")).block(Duration.ofSeconds(5));
verifyOutputContent(Arrays.asList("abc::response", "def::response", "ghi::response"));
}

7
spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolverTests.java
View File

@@ -129,9 +129,10 @@ public class PayloadMethodArgumentResolverTests {
@Test
public void validateStringMono() {
TestValidator validator = new TestValidator();
ResolvableType type = ResolvableType.forClassWithGenerics(Mono.class, String.class);
MethodParameter param = this.testMethod.arg(type);
Mono<Object> mono = resolveValue(param, Mono.just(toDataBuffer("12345")), new TestValidator());
Mono<Object> mono = resolveValue(param, Mono.just(toDataBuffer("12345")), validator);
StepVerifier.create(mono).expectNextCount(0)
.expectError(MethodArgumentNotValidException.class).verify();
@@ -139,9 +140,11 @@ public class PayloadMethodArgumentResolverTests {
@Test
public void validateStringFlux() {
TestValidator validator = new TestValidator();
ResolvableType type = ResolvableType.forClassWithGenerics(Flux.class, String.class);
MethodParameter param = this.testMethod.arg(type);
Flux<Object> flux = resolveValue(param, Mono.just(toDataBuffer("12345678\n12345")), new TestValidator());
Flux<DataBuffer> content = Flux.just(toDataBuffer("12345678"), toDataBuffer("12345"));
Flux<Object> flux = resolveValue(param, content, validator);
StepVerifier.create(flux)
.expectNext("12345678")

25
spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageReader.java
View File

@@ -106,10 +106,11 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
return stringDecoder.decode(message.getBody(), STRING_TYPE, null, hints)
.bufferUntil(line -> line.equals(""))
.concatMap(lines -> buildEvent(lines, valueType, shouldWrap, hints));
.concatMap(lines -> Mono.justOrEmpty(buildEvent(lines, valueType, shouldWrap, hints)));
}
private Mono<?> buildEvent(List<String> lines, ResolvableType valueType, boolean shouldWrap,
@Nullable
private Object buildEvent(List<String> lines, ResolvableType valueType, boolean shouldWrap,
Map<String, Object> hints) {
ServerSentEvent.Builder<Object> sseBuilder = shouldWrap ? ServerSentEvent.builder() : null;
@@ -138,34 +139,32 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
}
}
Mono<?> decodedData = (data != null ? decodeData(data.toString(), valueType, hints) : Mono.empty());
Object decodedData = data != null ? decodeData(data.toString(), valueType, hints) : null;
if (shouldWrap) {
if (comment != null) {
sseBuilder.comment(comment.toString().substring(0, comment.length() - 1));
}
return decodedData.map(o -> {
sseBuilder.data(o);
return sseBuilder.build();
});
if (decodedData != null) {
sseBuilder.data(decodedData);
}
return sseBuilder.build();
}
else {
return decodedData;
}
}
private Mono<?> decodeData(String data, ResolvableType dataType, Map<String, Object> hints) {
private Object decodeData(String data, ResolvableType dataType, Map<String, Object> hints) {
if (String.class == dataType.resolve()) {
return Mono.just(data.substring(0, data.length() - 1));
return data.substring(0, data.length() - 1);
}
if (this.decoder == null) {
return Mono.error(new CodecException("No SSE decoder configured and the data is not String."));
throw new CodecException("No SSE decoder configured and the data is not String.");
}
byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
DataBuffer buffer = bufferFactory.wrap(bytes); // wrapping only, no allocation
return this.decoder.decodeToMono(Mono.just(buffer), dataType, MediaType.TEXT_EVENT_STREAM, hints);
return this.decoder.decode(buffer, dataType, MediaType.TEXT_EVENT_STREAM, hints);
}
@Override

50
spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java
View File

@@ -18,6 +18,7 @@ package org.springframework.http.codec;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -111,9 +112,9 @@ public class ServerSentEventHttpMessageWriter implements HttpMessageWriter<Objec
}
private Flux<Publisher<DataBuffer>> encode(Publisher<?> input, ResolvableType elementType,
MediaType mediaType, DataBufferFactory factory, Map<String, Object> hints) {
MediaType mediaType, DataBufferFactory bufferFactory, Map<String, Object> hints) {
ResolvableType valueType = (ServerSentEvent.class.isAssignableFrom(elementType.toClass()) ?
ResolvableType dataType = (ServerSentEvent.class.isAssignableFrom(elementType.toClass()) ?
elementType.getGeneric() : elementType);
return Flux.from(input).map(element -> {
@@ -143,12 +144,10 @@ public class ServerSentEventHttpMessageWriter implements HttpMessageWriter<Objec
sb.append("data:");
}
Flux<DataBuffer> flux = Flux.concat(
encodeText(sb, mediaType, factory),
encodeData(data, valueType, mediaType, factory, hints),
encodeText("\n", mediaType, factory));
Mono<DataBuffer> bufferMono = Mono.fromCallable(() ->
bufferFactory.join(encodeEvent(sb, data, dataType, mediaType, bufferFactory, hints)));
return flux.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
return bufferMono.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
});
}
@@ -160,31 +159,32 @@ public class ServerSentEventHttpMessageWriter implements HttpMessageWriter<Objec
}
@SuppressWarnings("unchecked")
private <T> Flux<DataBuffer> encodeData(@Nullable T dataValue, ResolvableType valueType,
private <T> List<DataBuffer> encodeEvent(CharSequence markup, @Nullable T data, ResolvableType dataType,
MediaType mediaType, DataBufferFactory factory, Map<String, Object> hints) {
if (dataValue == null) {
return Flux.empty();
List<DataBuffer> result = new ArrayList<>(4);
result.add(encodeText(markup, mediaType, factory));
if (data != null) {
if (data instanceof String) {
String dataLine = StringUtils.replace((String) data, "\n", "\ndata:") + "\n";
result.add(encodeText(dataLine, mediaType, factory));
}
else if (this.encoder == null) {
throw new CodecException("No SSE encoder configured and the data is not String.");
}
else {
result.add(((Encoder<T>) this.encoder).encodeValue(data, factory, dataType, mediaType, hints));
result.add(encodeText("\n", mediaType, factory));
}
}
if (dataValue instanceof String) {
String text = (String) dataValue;
return Flux.from(encodeText(StringUtils.replace(text, "\n", "\ndata:") + "\n", mediaType, factory));
}
if (this.encoder == null) {
return Flux.error(new CodecException("No SSE encoder configured and the data is not String."));
}
return ((Encoder<T>) this.encoder)
.encode(Mono.just(dataValue), factory, valueType, mediaType, hints)
.concatWith(encodeText("\n", mediaType, factory));
result.add(encodeText("\n", mediaType, factory));
return result;
}
private Mono<DataBuffer> encodeText(CharSequence text, MediaType mediaType, DataBufferFactory bufferFactory) {
private DataBuffer encodeText(CharSequence text, MediaType mediaType, DataBufferFactory bufferFactory) {
Assert.notNull(mediaType.getCharset(), "Expected MediaType with charset");
byte[] bytes = text.toString().getBytes(mediaType.getCharset());
return Mono.just(bufferFactory.wrap(bytes)); // wrapping, not allocating
return bufferFactory.wrap(bytes); // wrapping, not allocating
}
@Override

92
spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java
View File

@@ -38,6 +38,7 @@ import org.springframework.core.codec.CodecException;
import org.springframework.core.codec.DecodingException;
import org.springframework.core.codec.Hints;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.log.LogFormatUtils;
import org.springframework.http.codec.HttpMessageDecoder;
import org.springframework.http.server.reactive.ServerHttpRequest;
@@ -88,56 +89,79 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple
Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
Flux.from(input), this.jsonFactory, getObjectMapper(), true);
return decodeInternal(tokens, elementType, mimeType, hints);
ObjectReader reader = getObjectReader(elementType, hints);
return tokens.handle((tokenBuffer, sink) -> {
try {
Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
logValue(value, hints);
if (value != null) {
sink.next(value);
}
}
catch (IOException ex) {
sink.error(processException(ex));
}
});
}
@Override
public Mono<Object> decodeToMono(Publisher<DataBuffer> input, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
Flux.from(input), this.jsonFactory, getObjectMapper(), false);
return decodeInternal(tokens, elementType, mimeType, hints).singleOrEmpty();
return DataBufferUtils.join(input)
.map(dataBuffer -> decode(dataBuffer, elementType, mimeType, hints));
}
private Flux<Object> decodeInternal(Flux<TokenBuffer> tokens, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
@Override
public Object decode(DataBuffer dataBuffer, ResolvableType targetType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
Assert.notNull(tokens, "'tokens' must not be null");
try {
ObjectReader objectReader = getObjectReader(targetType, hints);
Object value = objectReader.readValue(dataBuffer.asInputStream());
logValue(value, hints);
return value;
}
catch (IOException ex) {
throw processException(ex);
}
finally {
DataBufferUtils.release(dataBuffer);
}
}
private ObjectReader getObjectReader(ResolvableType elementType, @Nullable Map<String, Object> hints) {
Assert.notNull(elementType, "'elementType' must not be null");
MethodParameter param = getParameter(elementType);
Class<?> contextClass = (param != null ? param.getContainingClass() : null);
JavaType javaType = getJavaType(elementType.getType(), contextClass);
Class<?> jsonView = (hints != null ? (Class<?>) hints.get(Jackson2CodecSupport.JSON_VIEW_HINT) : null);
ObjectReader reader = (jsonView != null ?
return jsonView != null ?
getObjectMapper().readerWithView(jsonView).forType(javaType) :
getObjectMapper().readerFor(javaType));
getObjectMapper().readerFor(javaType);
}
return tokens.handle((tokenBuffer, sink) -> {
try {
Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
String formatted = LogFormatUtils.formatValue(value, !traceOn);
return Hints.getLogPrefix(hints) + "Decoded [" + formatted + "]";
});
}
if (value != null) {
sink.next(value);
}
}
catch (InvalidDefinitionException ex) {
sink.error(new CodecException("Type definition error: " + ex.getType(), ex));
}
catch (JsonProcessingException ex) {
sink.error(new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex));
}
catch (IOException ex) {
sink.error(new DecodingException("I/O error while parsing input stream", ex));
}
});
private void logValue(@Nullable Object value, @Nullable Map<String, Object> hints) {
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
String formatted = LogFormatUtils.formatValue(value, !traceOn);
return Hints.getLogPrefix(hints) + "Decoded [" + formatted + "]";
});
}
}
private CodecException processException(IOException ex) {
if (ex instanceof InvalidDefinitionException) {
JavaType type = ((InvalidDefinitionException) ex).getType();
return new CodecException("Type definition error: " + type, ex);
}
if (ex instanceof JsonProcessingException) {
String originalMessage = ((JsonProcessingException) ex).getOriginalMessage();
return new DecodingException("JSON decoding error: " + originalMessage, ex);
}
return new DecodingException("I/O error while parsing input stream", ex);
}

21
spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java
View File

@@ -119,7 +119,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
if (inputStream instanceof Mono) {
return Mono.from(inputStream).map(value ->
encodeValue(value, mimeType, bufferFactory, elementType, hints, encoding)).flux();
encodeValue(value, bufferFactory, elementType, mimeType, hints, encoding)).flux();
}
else {
return this.streamingMediaTypes.stream()
@@ -129,7 +129,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
byte[] separator = STREAM_SEPARATORS.getOrDefault(mediaType, NEWLINE_SEPARATOR);
return Flux.from(inputStream).map(value -> {
DataBuffer buffer = encodeValue(
value, mimeType, bufferFactory, elementType, hints, encoding);
value, bufferFactory, elementType, mimeType, hints, encoding);
if (separator != null) {
buffer.write(separator);
}
@@ -139,13 +139,20 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
.orElseGet(() -> {
ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
return Flux.from(inputStream).collectList().map(list ->
encodeValue(list, mimeType, bufferFactory, listType, hints, encoding)).flux();
encodeValue(list, bufferFactory, listType, mimeType, hints, encoding)).flux();
});
}
}
private DataBuffer encodeValue(Object value, @Nullable MimeType mimeType, DataBufferFactory bufferFactory,
ResolvableType elementType, @Nullable Map<String, Object> hints, JsonEncoding encoding) {
@Override
public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return encodeValue(value, bufferFactory, valueType, mimeType, hints, getJsonEncoding(mimeType));
}
private DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, ResolvableType valueType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints, JsonEncoding encoding) {
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
@@ -154,7 +161,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
});
}
JavaType javaType = getJavaType(elementType.getType(), null);
JavaType javaType = getJavaType(valueType.getType(), null);
Class<?> jsonView = (hints != null ? (Class<?>) hints.get(Jackson2CodecSupport.JSON_VIEW_HINT) : null);
ObjectWriter writer = (jsonView != null ?
getObjectMapper().writerWithView(jsonView) : getObjectMapper().writer());
@@ -163,7 +170,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple
writer = writer.forType(javaType);
}
writer = customizeWriter(writer, mimeType, elementType, hints);
writer = customizeWriter(writer, mimeType, valueType, hints);
DataBuffer buffer = bufferFactory.allocateBuffer();
boolean release = true;

42
spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java
View File

@@ -127,26 +127,32 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
public Mono<Message> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return DataBufferUtils.join(inputStream).map(dataBuffer -> {
try {
Message.Builder builder = getMessageBuilder(elementType.toClass());
ByteBuffer buffer = dataBuffer.asByteBuffer();
builder.mergeFrom(CodedInputStream.newInstance(buffer), this.extensionRegistry);
return builder.build();
}
catch (IOException ex) {
throw new DecodingException("I/O error while parsing input stream", ex);
}
catch (Exception ex) {
throw new DecodingException("Could not read Protobuf message: " + ex.getMessage(), ex);
}
finally {
DataBufferUtils.release(dataBuffer);
}
}
);
return DataBufferUtils.join(inputStream)
.map(dataBuffer -> decode(dataBuffer, elementType, mimeType, hints));
}
@Override
public Message decode(DataBuffer dataBuffer, ResolvableType targetType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
try {
Message.Builder builder = getMessageBuilder(targetType.toClass());
ByteBuffer buffer = dataBuffer.asByteBuffer();
builder.mergeFrom(CodedInputStream.newInstance(buffer), this.extensionRegistry);
return builder.build();
}
catch (IOException ex) {
throw new DecodingException("I/O error while parsing input stream", ex);
}
catch (Exception ex) {
throw new DecodingException("Could not read Protobuf message: " + ex.getMessage(), ex);
}
finally {
DataBufferUtils.release(dataBuffer);
}
}
/**
* Create a new {@code Message.Builder} instance for the given class.
* <p>This method uses a ConcurrentHashMap for caching method lookups.

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

@@ -73,29 +73,39 @@ public class ProtobufEncoder extends ProtobufCodecSupport implements HttpMessage
public Flux<DataBuffer> encode(Publisher<? extends Message> inputStream, DataBufferFactory bufferFactory,
ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return Flux.from(inputStream)
.map(message -> {
DataBuffer buffer = bufferFactory.allocateBuffer();
boolean release = true;
try {
if (!(inputStream instanceof Mono)) {
message.writeDelimitedTo(buffer.asOutputStream());
}
else {
message.writeTo(buffer.asOutputStream());
}
release = false;
return buffer;
}
catch (IOException ex) {
throw new IllegalStateException("Unexpected I/O error while writing to data buffer", ex);
}
finally {
if (release) {
DataBufferUtils.release(buffer);
}
}
});
return Flux.from(inputStream).map(message ->
encodeValue(message, bufferFactory, !(inputStream instanceof Mono)));
}
@Override
public DataBuffer encodeValue(Message message, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return encodeValue(message, bufferFactory, false);
}
private DataBuffer encodeValue(Message message, DataBufferFactory bufferFactory, boolean delimited) {
DataBuffer buffer = bufferFactory.allocateBuffer();
boolean release = true;
try {
if (delimited) {
message.writeDelimitedTo(buffer.asOutputStream());
}
else {
message.writeTo(buffer.asOutputStream());
}
release = false;
return buffer;
}
catch (IOException ex) {
throw new IllegalStateException("Unexpected I/O error while writing to data buffer", ex);
}
finally {
if (release) {
DataBufferUtils.release(buffer);
}
}
}
@Override

32
spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java
View File

@@ -17,6 +17,7 @@
package org.springframework.http.codec.xml;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
@@ -31,9 +32,12 @@ import javax.xml.bind.annotation.XmlSchema;
import javax.xml.bind.annotation.XmlType;
import javax.xml.namespace.QName;
import javax.xml.stream.XMLEventReader;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.events.XMLEvent;
import org.reactivestreams.Publisher;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SynchronousSink;
@@ -44,6 +48,7 @@ import org.springframework.core.codec.CodecException;
import org.springframework.core.codec.DecodingException;
import org.springframework.core.codec.Hints;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.log.LogFormatUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -72,6 +77,8 @@ public class Jaxb2XmlDecoder extends AbstractDecoder<Object> {
*/
private static final String JAXB_DEFAULT_ANNOTATION_VALUE = "##default";
private static final XMLInputFactory inputFactory = StaxUtils.createDefensiveInputFactory();
private final XmlEventDecoder xmlEventDecoder = new XmlEventDecoder();
@@ -132,10 +139,31 @@ public class Jaxb2XmlDecoder extends AbstractDecoder<Object> {
}
@Override
public Mono<Object> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
@SuppressWarnings({"rawtypes", "unchecked", "cast"}) // XMLEventReader is Iterator<Object> on JDK 9
public Mono<Object> decodeToMono(Publisher<DataBuffer> input, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return decode(inputStream, elementType, mimeType, hints).singleOrEmpty();
return DataBufferUtils.join(input)
.map(dataBuffer -> decode(dataBuffer, elementType, mimeType, hints));
}
@Override
@SuppressWarnings({"rawtypes", "unchecked", "cast"}) // XMLEventReader is Iterator<Object> on JDK 9
public Object decode(DataBuffer dataBuffer, ResolvableType targetType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
try {
Iterator eventReader = inputFactory.createXMLEventReader(dataBuffer.asInputStream());
List<XMLEvent> events = new ArrayList<>();
eventReader.forEachRemaining(event -> events.add((XMLEvent) event));
return unmarshal(events, targetType.toClass());
}
catch (XMLStreamException ex) {
throw Exceptions.propagate(ex);
}
finally {
DataBufferUtils.release(dataBuffer);
}
}
private Object unmarshal(List<XMLEvent> events, Class<?> outputClass) {

53
spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java
View File

@@ -99,7 +99,15 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder<Object> {
@Override
protected Flux<DataBuffer> encode(Object value, DataBufferFactory bufferFactory,
ResolvableType type, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
// we're relying on doOnDiscard in base class
return Mono.fromCallable(() -> encodeValue(value, bufferFactory, valueType, mimeType, hints)).flux();
}
@Override
public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory,
ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
if (!Hints.isLoggingSuppressed(hints)) {
LogFormatUtils.traceDebug(logger, traceOn -> {
@@ -108,30 +116,27 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder<Object> {
});
}
return Flux.defer(() -> {
boolean release = true;
DataBuffer buffer = bufferFactory.allocateBuffer(1024);
try {
OutputStream outputStream = buffer.asOutputStream();
Class<?> clazz = ClassUtils.getUserClass(value);
Marshaller marshaller = initMarshaller(clazz);
marshaller.marshal(value, outputStream);
release = false;
return Mono.fromCallable(() -> buffer); // relying on doOnDiscard in base class
boolean release = true;
DataBuffer buffer = bufferFactory.allocateBuffer(1024);
try {
OutputStream outputStream = buffer.asOutputStream();
Class<?> clazz = ClassUtils.getUserClass(value);
Marshaller marshaller = initMarshaller(clazz);
marshaller.marshal(value, outputStream);
release = false;
return buffer;
}
catch (MarshalException ex) {
throw new EncodingException("Could not marshal " + value.getClass() + " to XML", ex);
}
catch (JAXBException ex) {
throw new CodecException("Invalid JAXB configuration", ex);
}
finally {
if (release) {
DataBufferUtils.release(buffer);
}
catch (MarshalException ex) {
return Flux.error(new EncodingException(
"Could not marshal " + value.getClass() + " to XML", ex));
}
catch (JAXBException ex) {
return Flux.error(new CodecException("Invalid JAXB configuration", ex));
}
finally {
if (release) {
DataBufferUtils.release(buffer);
}
}
});
}
}
private Marshaller initMarshaller(Class<?> clazz) throws JAXBException {

2
spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java
View File

@@ -95,7 +95,7 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
@Override
@SuppressWarnings({"rawtypes", "unchecked", "cast"}) // on JDK 9 where XMLEventReader is Iterator<Object> instead of simply Iterator
@SuppressWarnings({"rawtypes", "unchecked", "cast"}) // XMLEventReader is Iterator<Object> on JDK 9
public Flux<XMLEvent> decode(Publisher<DataBuffer> input, ResolvableType elementType,
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

76
spring-web/src/test/java/org/springframework/http/codec/ServerSentEventHttpMessageWriterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -40,7 +40,7 @@ import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
import static org.junit.Assert.*;
import static org.springframework.core.ResolvableType.forClass;
import static org.springframework.core.ResolvableType.*;
/**
* Unit tests for {@link ServerSentEventHttpMessageWriter}.
@@ -88,9 +88,8 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
testWrite(source, outputMessage, ServerSentEvent.class);
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(stringConsumer("id:c42\nevent:foo\nretry:123\n:bla\n:bla bla\n:bla bla bla\ndata:"))
.consumeNextWith(stringConsumer("bar\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer(
"id:c42\nevent:foo\nretry:123\n:bla\n:bla bla\n:bla bla bla\ndata:bar\n\n"))
.expectComplete()
.verify();
}
@@ -101,12 +100,8 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
testWrite(source, outputMessage, String.class);
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("foo\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("bar\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:foo\n\n"))
.consumeNextWith(stringConsumer("data:bar\n\n"))
.expectComplete()
.verify();
}
@@ -117,12 +112,8 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
testWrite(source, outputMessage, String.class);
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("foo\ndata:bar\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("foo\ndata:baz\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:foo\ndata:bar\n\n"))
.consumeNextWith(stringConsumer("data:foo\ndata:baz\n\n"))
.expectComplete()
.verify();
}
@@ -136,14 +127,11 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
assertEquals(mediaType, outputMessage.getHeaders().getContentType());
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(dataBuffer -> {
String value =
DataBufferTestUtils.dumpString(dataBuffer, charset);
String value = DataBufferTestUtils.dumpString(dataBuffer, charset);
DataBufferUtils.release(dataBuffer);
assertEquals("\u00A3\n", value);
assertEquals("data:\u00A3\n\n", value);
})
.consumeNextWith(stringConsumer("\n"))
.expectComplete()
.verify();
}
@@ -154,14 +142,8 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
testWrite(source, outputMessage, Pojo.class);
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n\n"))
.consumeNextWith(stringConsumer("data:{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n\n"))
.expectComplete()
.verify();
}
@@ -175,18 +157,12 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
testWrite(source, outputMessage, Pojo.class);
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("{\n" +
.consumeNextWith(stringConsumer("data:{\n" +
"data: \"foo\" : \"foofoo\",\n" +
"data: \"bar\" : \"barbar\"\n" + "data:}"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("data:"))
.consumeNextWith(stringConsumer("{\n" +
"data: \"bar\" : \"barbar\"\n" + "data:}\n\n"))
.consumeNextWith(stringConsumer("data:{\n" +
"data: \"foo\" : \"foofoofoo\",\n" +
"data: \"bar\" : \"barbarbar\"\n" + "data:}"))
.consumeNextWith(stringConsumer("\n"))
.consumeNextWith(stringConsumer("\n"))
"data: \"bar\" : \"barbarbar\"\n" + "data:}\n\n"))
.expectComplete()
.verify();
}
@@ -200,28 +176,10 @@ public class ServerSentEventHttpMessageWriterTests extends AbstractDataBufferAll
assertEquals(mediaType, outputMessage.getHeaders().getContentType());
StepVerifier.create(outputMessage.getBody())
.consumeNextWith(dataBuffer1 -> {
String value1 =
DataBufferTestUtils.dumpString(dataBuffer1, charset);
DataBufferUtils.release(dataBuffer1);
assertEquals("data:", value1);
})
.consumeNextWith(dataBuffer -> {
String value = DataBufferTestUtils.dumpString(dataBuffer, charset);
DataBufferUtils.release(dataBuffer);
assertEquals("{\"foo\":\"foo\uD834\uDD1E\",\"bar\":\"bar\uD834\uDD1E\"}", value);
})
.consumeNextWith(dataBuffer2 -> {
String value2 =
DataBufferTestUtils.dumpString(dataBuffer2, charset);
DataBufferUtils.release(dataBuffer2);
assertEquals("\n", value2);
})
.consumeNextWith(dataBuffer3 -> {
String value3 =
DataBufferTestUtils.dumpString(dataBuffer3, charset);
DataBufferUtils.release(dataBuffer3);
assertEquals("\n", value3);
assertEquals("data:{\"foo\":\"foo\uD834\uDD1E\",\"bar\":\"bar\uD834\uDD1E\"}\n\n", value);
})
.expectComplete()
.verify();