From 2aae81ef0cf415a2ae4501ac5bf16b24591e1bd6 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Wed, 10 Apr 2019 17:33:08 -0400 Subject: [PATCH 1/5] Join buffers in decodeToMono for Jackson and Jaxb2 Closes gh-22783 --- .../codec/json/AbstractJackson2Decoder.java | 88 +++++++++++-------- .../http/codec/xml/Jaxb2XmlDecoder.java | 25 +++++- .../http/codec/xml/XmlEventDecoder.java | 2 +- 3 files changed, 77 insertions(+), 38 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java index aa346306c3..f5f2690256 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java @@ -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,73 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple Flux 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 decodeToMono(Publisher input, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { - Flux tokens = Jackson2Tokenizer.tokenize( - Flux.from(input), this.jsonFactory, getObjectMapper(), false); - return decodeInternal(tokens, elementType, mimeType, hints).singleOrEmpty(); + return DataBufferUtils.join(input).map(dataBuffer -> { + try { + ObjectReader objectReader = getObjectReader(elementType, hints); + Object value = objectReader.readValue(dataBuffer.asInputStream()); + logValue(value, hints); + return value; + } + catch (IOException ex) { + throw processException(ex); + } + finally { + DataBufferUtils.release(dataBuffer); + } + }); } - private Flux decodeInternal(Flux tokens, ResolvableType elementType, - @Nullable MimeType mimeType, @Nullable Map hints) { - - Assert.notNull(tokens, "'tokens' must not be null"); + private ObjectReader getObjectReader(ResolvableType elementType, @Nullable Map 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 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); } diff --git a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java index ab63d30741..04227f73e2 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java @@ -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 { */ 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,24 @@ public class Jaxb2XmlDecoder extends AbstractDecoder { } @Override - public Mono decodeToMono(Publisher inputStream, ResolvableType elementType, + @SuppressWarnings({"rawtypes", "unchecked", "cast"}) // XMLEventReader is Iterator on JDK 9 + public Mono decodeToMono(Publisher input, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { - return decode(inputStream, elementType, mimeType, hints).singleOrEmpty(); + return DataBufferUtils.join(input).map(dataBuffer -> { + try { + Iterator eventReader = inputFactory.createXMLEventReader(dataBuffer.asInputStream()); + List events = new ArrayList<>(); + eventReader.forEachRemaining(event -> events.add((XMLEvent) event)); + return unmarshal(events, elementType.toClass()); + } + catch (XMLStreamException ex) { + throw Exceptions.propagate(ex); + } + finally { + DataBufferUtils.release(dataBuffer); + } + }); } private Object unmarshal(List events, Class outputClass) { diff --git a/spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java b/spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java index d44ff23a11..b417de7610 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java @@ -95,7 +95,7 @@ public class XmlEventDecoder extends AbstractDecoder { @Override - @SuppressWarnings({"rawtypes", "unchecked", "cast"}) // on JDK 9 where XMLEventReader is Iterator instead of simply Iterator + @SuppressWarnings({"rawtypes", "unchecked", "cast"}) // XMLEventReader is Iterator on JDK 9 public Flux decode(Publisher input, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { From a912d8de1ea26cd071342f919eb0c6bad4945062 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Thu, 11 Apr 2019 13:30:55 -0400 Subject: [PATCH 2/5] Add option to decode from a DataBuffer See gh-22782 --- .../core/codec/AbstractDataBufferDecoder.java | 11 ++++- .../core/codec/ByteArrayDecoder.java | 4 +- .../core/codec/ByteBufferDecoder.java | 4 +- .../core/codec/DataBufferDecoder.java | 4 +- .../springframework/core/codec/Decoder.java | 29 +++++++++++++ .../core/codec/ResourceDecoder.java | 2 +- .../core/codec/StringDecoder.java | 2 +- .../codec/json/AbstractJackson2Decoder.java | 34 ++++++++------- .../http/codec/protobuf/ProtobufDecoder.java | 42 +++++++++++-------- .../http/codec/xml/Jaxb2XmlDecoder.java | 35 +++++++++------- 10 files changed, 111 insertions(+), 56 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java b/spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java index b03d8079db..f6a29a747a 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java @@ -45,6 +45,7 @@ import org.springframework.util.MimeType; * @since 5.0 * @param the element type */ +@SuppressWarnings("deprecation") public abstract class AbstractDataBufferDecoder extends AbstractDecoder { @@ -70,8 +71,14 @@ public abstract class AbstractDataBufferDecoder extends AbstractDecoder { /** * 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 hints); + @Deprecated + protected T decodeDataBuffer(DataBuffer buffer, ResolvableType elementType, + @Nullable MimeType mimeType, @Nullable Map hints) { + + return decode(buffer, elementType, mimeType, hints); + } } diff --git a/spring-core/src/main/java/org/springframework/core/codec/ByteArrayDecoder.java b/spring-core/src/main/java/org/springframework/core/codec/ByteArrayDecoder.java index 24ac760817..65f8222d17 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/ByteArrayDecoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/ByteArrayDecoder.java @@ -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 { } @Override - protected byte[] decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType, + public byte[] decode(DataBuffer dataBuffer, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { byte[] result = new byte[dataBuffer.readableByteCount()]; diff --git a/spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java b/spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java index c60501d2ff..9c1133fb1a 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/ByteBufferDecoder.java @@ -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 { } @Override - protected ByteBuffer decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType, + public ByteBuffer decode(DataBuffer dataBuffer, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { int byteCount = dataBuffer.readableByteCount(); diff --git a/spring-core/src/main/java/org/springframework/core/codec/DataBufferDecoder.java b/spring-core/src/main/java/org/springframework/core/codec/DataBufferDecoder.java index 17d6a424ab..34b150378e 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/DataBufferDecoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/DataBufferDecoder.java @@ -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 { } @Override - protected DataBuffer decodeDataBuffer(DataBuffer buffer, ResolvableType elementType, + public DataBuffer decode(DataBuffer buffer, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { if (logger.isDebugEnabled()) { diff --git a/spring-core/src/main/java/org/springframework/core/codec/Decoder.java b/spring-core/src/main/java/org/springframework/core/codec/Decoder.java index 3b37ecfef5..c4e84383f6 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/Decoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/Decoder.java @@ -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 { Mono decodeToMono(Publisher inputStream, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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 hints) throws DecodingException { + + MonoProcessor 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. */ diff --git a/spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java b/spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java index d2b94d13ea..eeb307cfcd 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/ResourceDecoder.java @@ -64,7 +64,7 @@ public class ResourceDecoder extends AbstractDataBufferDecoder { } @Override - protected Resource decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType, + public Resource decode(DataBuffer dataBuffer, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { byte[] bytes = new byte[dataBuffer.readableByteCount()]; diff --git a/spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java b/spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java index 28cf7df55e..9001de29c7 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java @@ -202,7 +202,7 @@ public final class StringDecoder extends AbstractDataBufferDecoder { } @Override - protected String decodeDataBuffer(DataBuffer dataBuffer, ResolvableType elementType, + public String decode(DataBuffer dataBuffer, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { Charset charset = getCharset(mimeType); diff --git a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java index f5f2690256..389d56470f 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java @@ -110,20 +110,26 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple public Mono decodeToMono(Publisher input, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { - return DataBufferUtils.join(input).map(dataBuffer -> { - try { - ObjectReader objectReader = getObjectReader(elementType, hints); - Object value = objectReader.readValue(dataBuffer.asInputStream()); - logValue(value, hints); - return value; - } - catch (IOException ex) { - throw processException(ex); - } - finally { - DataBufferUtils.release(dataBuffer); - } - }); + return DataBufferUtils.join(input) + .map(dataBuffer -> decode(dataBuffer, elementType, mimeType, hints)); + } + + @Override + public Object decode(DataBuffer dataBuffer, ResolvableType targetType, + @Nullable MimeType mimeType, @Nullable Map hints) throws DecodingException { + + 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 hints) { diff --git a/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java b/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java index c1f0e17bf7..37d7ae4d90 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java @@ -127,26 +127,32 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder decodeToMono(Publisher inputStream, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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 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. *

This method uses a ConcurrentHashMap for caching method lookups. diff --git a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java index 04227f73e2..dd5517259c 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java @@ -143,20 +143,27 @@ public class Jaxb2XmlDecoder extends AbstractDecoder { public Mono decodeToMono(Publisher input, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { - return DataBufferUtils.join(input).map(dataBuffer -> { - try { - Iterator eventReader = inputFactory.createXMLEventReader(dataBuffer.asInputStream()); - List events = new ArrayList<>(); - eventReader.forEachRemaining(event -> events.add((XMLEvent) event)); - return unmarshal(events, elementType.toClass()); - } - catch (XMLStreamException ex) { - throw Exceptions.propagate(ex); - } - finally { - DataBufferUtils.release(dataBuffer); - } - }); + return DataBufferUtils.join(input) + .map(dataBuffer -> decode(dataBuffer, elementType, mimeType, hints)); + } + + @Override + @SuppressWarnings({"rawtypes", "unchecked", "cast"}) // XMLEventReader is Iterator on JDK 9 + public Object decode(DataBuffer dataBuffer, ResolvableType targetType, + @Nullable MimeType mimeType, @Nullable Map hints) throws DecodingException { + + try { + Iterator eventReader = inputFactory.createXMLEventReader(dataBuffer.asInputStream()); + List 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 events, Class outputClass) { From f89d2ac14891d21714462e6ea4f8dc32d325f1cf Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Wed, 10 Apr 2019 22:08:00 -0400 Subject: [PATCH 3/5] Use decode from a DataBuffer where feasible See gh-22782 --- .../PayloadMethodArgumentResolver.java | 10 ++++---- .../rsocket/DefaultRSocketRequester.java | 8 +++--- .../ServerSentEventHttpMessageReader.java | 25 +++++++++---------- 3 files changed, 21 insertions(+), 22 deletions(-) diff --git a/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolver.java index 88a1a91298..62fa3af05e 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolver.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolver.java @@ -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)); } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java index 63facb327a..c112cb16bd 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java @@ -244,8 +244,8 @@ final class DefaultRSocketRequester implements RSocketRequester { } Decoder decoder = strategies.decoder(elementType, dataMimeType); - return (Mono) decoder.decodeToMono( - payloadMono.map(this::retainDataAndReleasePayload), elementType, dataMimeType, EMPTY_HINTS); + return (Mono) payloadMono.map(this::retainDataAndReleasePayload) + .map(dataBuffer -> decoder.decode(dataBuffer, elementType, dataMimeType, EMPTY_HINTS)); } @SuppressWarnings("unchecked") @@ -261,8 +261,8 @@ final class DefaultRSocketRequester implements RSocketRequester { Decoder decoder = strategies.decoder(elementType, dataMimeType); - return payloadFlux.map(this::retainDataAndReleasePayload).concatMap(dataBuffer -> - (Mono) 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) { diff --git a/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageReader.java b/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageReader.java index d66f59c601..dba78e8ec2 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageReader.java +++ b/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageReader.java @@ -106,10 +106,11 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader line.equals("")) - .concatMap(lines -> buildEvent(lines, valueType, shouldWrap, hints)); + .concatMap(lines -> Mono.justOrEmpty(buildEvent(lines, valueType, shouldWrap, hints))); } - private Mono buildEvent(List lines, ResolvableType valueType, boolean shouldWrap, + @Nullable + private Object buildEvent(List lines, ResolvableType valueType, boolean shouldWrap, Map hints) { ServerSentEvent.Builder sseBuilder = shouldWrap ? ServerSentEvent.builder() : null; @@ -138,34 +139,32 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader 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 hints) { + private Object decodeData(String data, ResolvableType dataType, Map 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 From 181482fa15b02a1ae27f54aea39ea0a8acdc35ec Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Thu, 11 Apr 2019 13:50:56 -0400 Subject: [PATCH 4/5] Add option to encode with an Object value See gh-22782 --- .../core/codec/ByteArrayEncoder.java | 24 +++++--- .../core/codec/ByteBufferEncoder.java | 24 +++++--- .../core/codec/CharSequenceEncoder.java | 55 +++++++++--------- .../core/codec/DataBufferEncoder.java | 24 +++++--- .../springframework/core/codec/Encoder.java | 26 ++++++++- .../core/io/buffer/DataBufferUtils.java | 4 ++ .../codec/json/AbstractJackson2Encoder.java | 21 ++++--- .../http/codec/protobuf/ProtobufEncoder.java | 56 +++++++++++-------- .../http/codec/xml/Jaxb2XmlEncoder.java | 53 ++++++++++-------- 9 files changed, 181 insertions(+), 106 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/codec/ByteArrayEncoder.java b/spring-core/src/main/java/org/springframework/core/codec/ByteArrayEncoder.java index 6b538dbf7e..6eef1a1f77 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/ByteArrayEncoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/ByteArrayEncoder.java @@ -52,15 +52,21 @@ public class ByteArrayEncoder extends AbstractEncoder { DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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 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; } } diff --git a/spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java b/spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java index 1f394302b8..8d60066152 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/ByteBufferEncoder.java @@ -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 { DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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 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; } } diff --git a/spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java b/spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java index 10088ebb4c..267a68421f 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/CharSequenceEncoder.java @@ -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 { DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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 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) { diff --git a/spring-core/src/main/java/org/springframework/core/codec/DataBufferEncoder.java b/spring-core/src/main/java/org/springframework/core/codec/DataBufferEncoder.java index 3a7c853afa..88e8f1cc2a 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/DataBufferEncoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/DataBufferEncoder.java @@ -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 { @Nullable Map hints) { Flux 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 hints) { + + if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) { + logValue(buffer, hints); + } + return buffer; + } + + private void logValue(DataBuffer buffer, @Nullable Map hints) { + String logPrefix = Hints.getLogPrefix(hints); + logger.debug(logPrefix + "Writing " + buffer.readableByteCount() + " bytes"); + } + } diff --git a/spring-core/src/main/java/org/springframework/core/codec/Encoder.java b/spring-core/src/main/java/org/springframework/core/codec/Encoder.java index 7b1ee00190..00f42009ae 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/Encoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/Encoder.java @@ -60,13 +60,35 @@ public interface Encoder { * @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 encode(Publisher inputStream, DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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. + *

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 hints) { + + // It may not be possible to produce a single DataBuffer synchronously + throw new UnsupportedOperationException(); + } + /** * Return the list of mime types this encoder supports. */ diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java index b12f9434b2..96dd72abcd 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java @@ -441,6 +441,10 @@ public abstract class DataBufferUtils { public static Mono join(Publisher dataBuffers) { Assert.notNull(dataBuffers, "'dataBuffers' must not be null"); + if (dataBuffers instanceof Mono) { + return (Mono) dataBuffers; + } + return Flux.from(dataBuffers) .collectList() .filter(list -> !list.isEmpty()) diff --git a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java index 652ecb5d28..f9f4d73eda 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java @@ -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 hints, JsonEncoding encoding) { + @Override + public DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, + ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map hints) { + + return encodeValue(value, bufferFactory, valueType, mimeType, hints, getJsonEncoding(mimeType)); + } + + private DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, ResolvableType valueType, + @Nullable MimeType mimeType, @Nullable Map 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; diff --git a/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufEncoder.java b/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufEncoder.java index 3be1ac4774..3d7b3fe032 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufEncoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufEncoder.java @@ -73,29 +73,39 @@ public class ProtobufEncoder extends ProtobufCodecSupport implements HttpMessage public Flux encode(Publisher inputStream, DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map 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 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 diff --git a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java index 59a11970ad..9a8ee7890f 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java @@ -99,7 +99,15 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder { @Override protected Flux encode(Object value, DataBufferFactory bufferFactory, - ResolvableType type, @Nullable MimeType mimeType, @Nullable Map hints) { + ResolvableType valueType, @Nullable MimeType mimeType, @Nullable Map 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 hints) { if (!Hints.isLoggingSuppressed(hints)) { LogFormatUtils.traceDebug(logger, traceOn -> { @@ -108,30 +116,27 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder { }); } - 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 { From 5fc18064f26a2541cce0432a4cb0fc01104972e8 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Thu, 11 Apr 2019 18:56:54 -0400 Subject: [PATCH 5/5] Use encode with an Object value where feasible Closes gh-22782 --- .../core/io/buffer/DefaultDataBuffer.java | 6 +- ...stractEncoderMethodReturnValueHandler.java | 13 ++-- .../rsocket/DefaultRSocketRequester.java | 21 ++--- .../MessageMappingMessageHandlerTests.java | 2 +- .../PayloadMethodArgumentResolverTests.java | 7 +- .../ServerSentEventHttpMessageWriter.java | 50 ++++++------ ...ServerSentEventHttpMessageWriterTests.java | 76 +++++-------------- 7 files changed, 69 insertions(+), 106 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java index 0bf1d2a35c..75b83a6198 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java @@ -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; } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/AbstractEncoderMethodReturnValueHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/AbstractEncoderMethodReturnValueHandler.java index 8fc2e59879..983e1448f6 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/AbstractEncoderMethodReturnValueHandler.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/AbstractEncoderMethodReturnValueHandler.java @@ -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 Mono encodeValue( + private DataBuffer encodeValue( Object element, ResolvableType elementType, @Nullable Encoder encoder, DataBufferFactory bufferFactory, @Nullable MimeType mimeType, @Nullable Map 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 mono = Mono.just((T) element); - Flux dataBuffers = encoder.encode(mono, bufferFactory, elementType, mimeType, hints); - return DataBufferUtils.join(dataBuffers); + return encoder.encodeValue((T) element, bufferFactory, elementType, mimeType, hints); } /** diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java index c112cb16bd..97e0b2351e 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java @@ -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 payloadMono = encodeValue(input, ResolvableType.forInstance(input), null) + Mono 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 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 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 Mono encodeValue(T value, ResolvableType valueType, @Nullable Encoder encoder) { + private DataBuffer encodeValue(T value, ResolvableType valueType, @Nullable Encoder encoder) { if (encoder == null) { encoder = strategies.encoder(ResolvableType.forInstance(value), dataMimeType); } - return DataBufferUtils.join(((Encoder) encoder).encode( - Mono.just(value), strategies.dataBufferFactory(), valueType, dataMimeType, EMPTY_HINTS)); + return ((Encoder) encoder).encodeValue( + value, strategies.dataBufferFactory(), valueType, dataMimeType, EMPTY_HINTS); } private Payload firstPayload(DataBuffer data) { diff --git a/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandlerTests.java b/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandlerTests.java index 5496e87eeb..837311c403 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandlerTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/MessageMappingMessageHandlerTests.java @@ -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")); } diff --git a/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolverTests.java b/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolverTests.java index 83e1293cd8..da82e20dea 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolverTests.java +++ b/spring-messaging/src/test/java/org/springframework/messaging/handler/annotation/support/reactive/PayloadMethodArgumentResolverTests.java @@ -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 mono = resolveValue(param, Mono.just(toDataBuffer("12345")), new TestValidator()); + Mono 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 flux = resolveValue(param, Mono.just(toDataBuffer("12345678\n12345")), new TestValidator()); + Flux content = Flux.just(toDataBuffer("12345678"), toDataBuffer("12345")); + Flux flux = resolveValue(param, content, validator); StepVerifier.create(flux) .expectNext("12345678") diff --git a/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java index fec0ed09ec..3840bae308 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java +++ b/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java @@ -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> encode(Publisher input, ResolvableType elementType, - MediaType mediaType, DataBufferFactory factory, Map hints) { + MediaType mediaType, DataBufferFactory bufferFactory, Map 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 flux = Flux.concat( - encodeText(sb, mediaType, factory), - encodeData(data, valueType, mediaType, factory, hints), - encodeText("\n", mediaType, factory)); + Mono 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 Flux encodeData(@Nullable T dataValue, ResolvableType valueType, + private List encodeEvent(CharSequence markup, @Nullable T data, ResolvableType dataType, MediaType mediaType, DataBufferFactory factory, Map hints) { - if (dataValue == null) { - return Flux.empty(); + List 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) 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) 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 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 diff --git a/spring-web/src/test/java/org/springframework/http/codec/ServerSentEventHttpMessageWriterTests.java b/spring-web/src/test/java/org/springframework/http/codec/ServerSentEventHttpMessageWriterTests.java index 10eb6bffee..6949efb9e9 100644 --- a/spring-web/src/test/java/org/springframework/http/codec/ServerSentEventHttpMessageWriterTests.java +++ b/spring-web/src/test/java/org/springframework/http/codec/ServerSentEventHttpMessageWriterTests.java @@ -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();