From 6f461647274eb58781c4434d59668e111d0fbf6c Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Fri, 27 May 2016 09:19:42 +0200 Subject: [PATCH] Changed reduceToSingleBuffer to splitOnNewline In order to be more "reactive", changed StringDecoder's default from merging all buffers in the stream to a single buffer into splitting the buffers along newline (\r, \n) characters. --- .../core/codec/support/StringDecoder.java | 34 +++++++------- .../core/codec/support/XmlEventDecoder.java | 8 +++- .../codec/support/StringDecoderTests.java | 46 +++++++------------ .../codec/support/StringEncoderTests.java | 8 ++-- 4 files changed, 45 insertions(+), 51 deletions(-) diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringDecoder.java index f27ff8005e..ca477cde85 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringDecoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringDecoder.java @@ -19,6 +19,7 @@ package org.springframework.core.codec.support; import java.nio.CharBuffer; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.List; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; @@ -32,11 +33,9 @@ import org.springframework.util.MimeTypeUtils; /** * Decode from a bytes stream to a String stream. * - *

By default, this decoder will buffer the received elements into a single - * {@code ByteBuffer} and will emit a single {@code String} once the stream of - * elements is complete. This behavior can be turned off using an constructor - * argument but the {@code Subcriber} should pay attention to split characters - * issues. + *

By default, this decoder will split the received {@link DataBuffer}s along newline + * characters ({@code \r\n}), but this can be changed by passing {@code false} as + * constructor argument. * * @author Sebastien Deleuze * @author Brian Clozel @@ -48,13 +47,12 @@ public class StringDecoder extends AbstractDecoder { public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8; - private final boolean reduceToSingleBuffer; + private final boolean splitOnNewline; /** * Create a {@code StringDecoder} that decodes a bytes stream to a String stream * - *

By default, this decoder will buffer bytes and - * emit a single String as a result. + *

By default, this decoder will split along new lines. */ public StringDecoder() { this(true); @@ -63,12 +61,12 @@ public class StringDecoder extends AbstractDecoder { /** * Create a {@code StringDecoder} that decodes a bytes stream to a String stream * - * @param reduceToSingleBuffer whether this decoder should buffer all received items - * and decode a single consolidated String or re-emit items as they are provided + * @param splitOnNewline whether this decoder should split the received data buffers + * along newline characters */ - public StringDecoder(boolean reduceToSingleBuffer) { + public StringDecoder(boolean splitOnNewline) { super(new MimeType("text", "*", DEFAULT_CHARSET), MimeTypeUtils.ALL); - this.reduceToSingleBuffer = reduceToSingleBuffer; + this.splitOnNewline = splitOnNewline; } @Override @@ -81,8 +79,12 @@ public class StringDecoder extends AbstractDecoder { public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { Flux inputFlux = Flux.from(inputStream); - if (this.reduceToSingleBuffer) { - inputFlux = Flux.from(inputFlux.reduce(DataBuffer::write)); + if (this.splitOnNewline) { + inputFlux = inputFlux.flatMap(dataBuffer -> { + List tokens = + DataBufferUtils.tokenize(dataBuffer, b -> b == '\n' || b == '\r'); + return Flux.fromIterable(tokens); + }); } Charset charset = getCharset(mimeType); return inputFlux.map(dataBuffer -> { @@ -93,8 +95,8 @@ public class StringDecoder extends AbstractDecoder { } private Charset getCharset(MimeType mimeType) { - if (mimeType != null && mimeType.getCharSet() != null) { - return mimeType.getCharSet(); + if (mimeType != null && mimeType.getCharset() != null) { + return mimeType.getCharset(); } else { return DEFAULT_CHARSET; diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/XmlEventDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/XmlEventDecoder.java index b0174ad4c7..3f27b97fb7 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/XmlEventDecoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/XmlEventDecoder.java @@ -16,6 +16,7 @@ package org.springframework.core.codec.support; +import java.io.InputStream; import java.util.ArrayList; import java.util.List; import java.util.function.Function; @@ -91,9 +92,9 @@ public class XmlEventDecoder extends AbstractDecoder { else { Mono singleBuffer = flux.reduce(DataBuffer::write); return singleBuffer. - map(DataBuffer::asInputStream). - flatMap(is -> { + flatMap(dataBuffer -> { try { + InputStream is = dataBuffer.asInputStream(); XMLEventReader eventReader = inputFactory.createXMLEventReader(is); return Flux @@ -102,6 +103,9 @@ public class XmlEventDecoder extends AbstractDecoder { catch (XMLStreamException ex) { return Mono.error(ex); } + finally { + DataBufferUtils.release(dataBuffer); + } }); } } diff --git a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java index c1d69680af..34a8b0ae24 100644 --- a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java @@ -18,18 +18,16 @@ package org.springframework.core.codec.support; import org.junit.Before; import org.junit.Test; -import reactor.core.converter.RxJava1SingleConverter; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import reactor.core.test.TestSubscriber; -import rx.Single; import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; -import static org.junit.Assert.*; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; /** * @author Sebastien Deleuze @@ -62,48 +60,36 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase { @Test public void decode() throws InterruptedException { + this.decoder = new StringDecoder(false); Flux source = Flux.just(stringBuffer("foo"), stringBuffer("bar"), stringBuffer("baz")); Flux output = this.decoder.decode(source, ResolvableType.forClass(String.class), null); TestSubscriber testSubscriber = new TestSubscriber<>(); - testSubscriber.bindTo(output).assertValues("foobarbaz"); + testSubscriber.bindTo(output). + assertNoError(). + assertComplete(). + assertValues("foo", "bar", "baz"); } @Test - public void decodeDoNotBuffer() throws InterruptedException { - StringDecoder decoder = new StringDecoder(false); - Flux source = Flux.just(stringBuffer("foo"), stringBuffer("bar")); + public void decodeNewLine() throws InterruptedException { + DataBuffer fooBar = stringBuffer("\nfoo\r\nbar\r"); + DataBuffer baz = stringBuffer("\nbaz"); + Flux source = Flux.just(fooBar, baz); Flux output = decoder.decode(source, ResolvableType.forClass(String.class), null); TestSubscriber testSubscriber = new TestSubscriber<>(); - testSubscriber.bindTo(output).assertValues("foo", "bar"); + testSubscriber.bindTo(output). + assertNoError(). + assertComplete(). + assertValues("foo", "bar", "baz"); } - @Test - public void decodeMono() throws InterruptedException { - Flux source = - Flux.just(stringBuffer("foo"), stringBuffer("bar"), stringBuffer("baz")); - Mono mono = Mono.from(this.decoder.decode(source, - ResolvableType.forClassWithGenerics(Mono.class, String.class), - MediaType.TEXT_PLAIN)); - TestSubscriber testSubscriber = new TestSubscriber<>(); - testSubscriber.bindTo(mono).assertValues("foobarbaz"); - } - - @Test - public void decodeSingle() throws InterruptedException { - Flux source = Flux.just(stringBuffer("foo"), stringBuffer("bar")); - Single single = RxJava1SingleConverter.from(this.decoder.decode(source, - ResolvableType.forClassWithGenerics(Single.class, String.class), - MediaType.TEXT_PLAIN)); - String result = single.toBlocking().value(); - assertEquals("foobar", result); - } @Test public void decodeEmpty() throws InterruptedException { - Mono source = Mono.just(stringBuffer("")); + Flux source = Flux.just(stringBuffer("")); Flux output = this.decoder.decode(source, ResolvableType.forClass(String.class), null); TestSubscriber testSubscriber = new TestSubscriber<>(); diff --git a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java index 3742a6589a..20e642d934 100644 --- a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java @@ -27,6 +27,7 @@ import reactor.core.test.TestSubscriber; import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase; +import org.springframework.core.io.buffer.support.DataBufferUtils; import org.springframework.http.MediaType; import static org.junit.Assert.assertFalse; @@ -60,9 +61,10 @@ public class StringEncoderTests extends AbstractDataBufferAllocatingTestCase { Flux output = Flux.from( this.encoder.encode(Flux.just("foo"), this.allocator, null, null)) .map(chunk -> { - byte[] b = new byte[chunk.readableByteCount()]; - chunk.read(b); - return new String(b, StandardCharsets.UTF_8); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); + DataBufferUtils.release(chunk); + return new String(b, StandardCharsets.UTF_8); }); TestSubscriber testSubscriber = new TestSubscriber<>(); testSubscriber.bindTo(output).assertValues("foo");