From 1836b2825b3c844097105322aa9e659409906e86 Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Fri, 18 Mar 2016 16:05:30 +0100 Subject: [PATCH] Added DataBufferAllocator parameter to Encoder.encode, and use that instead of AbstractAllocatingEncoder base class. --- .../springframework/core/codec/Encoder.java | 8 ++-- .../support/AbstractAllocatingEncoder.java | 42 ------------------- .../core/codec/support/ByteBufferEncoder.java | 13 +++--- .../codec/support/JacksonJsonEncoder.java | 36 +++++++++------- .../core/codec/support/Jaxb2Encoder.java | 17 ++++---- .../core/codec/support/JsonObjectEncoder.java | 9 ++-- .../core/codec/support/StringEncoder.java | 16 +++---- .../codec/support/ByteBufferEncoderTests.java | 6 +-- .../support/JacksonJsonEncoderTests.java | 9 ++-- .../core/codec/support/Jaxb2EncoderTests.java | 9 ++-- .../codec/support/JsonObjectEncoderTests.java | 17 ++++---- .../codec/support/StringEncoderTests.java | 11 +++-- 12 files changed, 83 insertions(+), 110 deletions(-) delete mode 100644 spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractAllocatingEncoder.java diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/Encoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/Encoder.java index 0f82c73081..0a53c9552b 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/Encoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/Encoder.java @@ -16,7 +16,6 @@ package org.springframework.core.codec; -import java.nio.ByteBuffer; import java.util.List; import org.reactivestreams.Publisher; @@ -24,6 +23,7 @@ import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.util.MimeType; /** @@ -44,14 +44,16 @@ public interface Encoder { boolean canEncode(ResolvableType type, MimeType mimeType, Object... hints); /** - * Encode an input stream of {@code T} to an output {@link ByteBuffer} stream. + * Encode an input stream of {@code T} to an output {@link DataBuffer} stream. * @param inputStream the input stream to process. + * @param allocator a buffer allocator used to create the output * @param type the stream element type to process. * @param mimeType the mime type to process. * @param hints Additional information about how to do decode, optional. * @return the output stream */ - Flux encode(Publisher inputStream, ResolvableType type, + Flux encode(Publisher inputStream, + DataBufferAllocator allocator, ResolvableType type, MimeType mimeType, Object... hints); /** diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractAllocatingEncoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractAllocatingEncoder.java deleted file mode 100644 index 0275931fc8..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractAllocatingEncoder.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2002-2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.core.codec.support; - -import org.springframework.core.io.buffer.DataBufferAllocator; -import org.springframework.util.Assert; -import org.springframework.util.MimeType; - -/** - * @author Arjen Poutsma - */ -public abstract class AbstractAllocatingEncoder extends AbstractEncoder { - - private final DataBufferAllocator allocator; - - public AbstractAllocatingEncoder(DataBufferAllocator allocator, - MimeType... supportedMimeTypes) { - super(supportedMimeTypes); - Assert.notNull(allocator, "'allocator' must not be null"); - - this.allocator = allocator; - } - - public DataBufferAllocator allocator() { - return allocator; - } - -} diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferEncoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferEncoder.java index 3b72c7e355..4e05e1d62b 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferEncoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferEncoder.java @@ -30,10 +30,10 @@ import org.springframework.util.MimeTypeUtils; /** * @author Sebastien Deleuze */ -public class ByteBufferEncoder extends AbstractAllocatingEncoder { +public class ByteBufferEncoder extends AbstractEncoder { - public ByteBufferEncoder(DataBufferAllocator allocator) { - super(allocator, MimeTypeUtils.ALL); + public ByteBufferEncoder() { + super(MimeTypeUtils.ALL); } @@ -45,12 +45,11 @@ public class ByteBufferEncoder extends AbstractAllocatingEncoder { @Override public Flux encode(Publisher inputStream, - ResolvableType type, - MimeType mimeType, Object... hints) { - + DataBufferAllocator allocator, ResolvableType type, MimeType mimeType, + Object... hints) { //noinspection unchecked return Flux.from(inputStream).map(byteBuffer -> { - DataBuffer dataBuffer = allocator().allocateBuffer(byteBuffer.remaining()); + DataBuffer dataBuffer = allocator.allocateBuffer(byteBuffer.remaining()); dataBuffer.write(byteBuffer); return dataBuffer; }); diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonEncoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonEncoder.java index 6e86446f8a..fe86fe85c6 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonEncoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonEncoder.java @@ -30,6 +30,7 @@ import org.springframework.core.codec.CodecException; import org.springframework.core.codec.Encoder; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferAllocator; +import org.springframework.core.io.buffer.DefaultDataBufferAllocator; import org.springframework.util.MimeType; /** @@ -38,41 +39,46 @@ import org.springframework.util.MimeType; * @author Sebastien Deleuze * @see JacksonJsonDecoder */ -public class JacksonJsonEncoder extends AbstractAllocatingEncoder { +public class JacksonJsonEncoder extends AbstractEncoder { private final ObjectMapper mapper; private Encoder postProcessor; - public JacksonJsonEncoder(DataBufferAllocator allocator) { - this(allocator, new ObjectMapper(), null); + public JacksonJsonEncoder() { + this(new ObjectMapper(), null); } - public JacksonJsonEncoder(DataBufferAllocator allocator, - Encoder postProcessor) { - this(allocator, new ObjectMapper(), postProcessor); + public JacksonJsonEncoder(Encoder postProcessor) { + this(new ObjectMapper(), postProcessor); } - public JacksonJsonEncoder(DataBufferAllocator allocator, ObjectMapper mapper, + public JacksonJsonEncoder(ObjectMapper mapper, Encoder postProcessor) { - super(allocator, new MimeType("application", "json", StandardCharsets.UTF_8), + super(new MimeType("application", "json", StandardCharsets.UTF_8), new MimeType("application", "*+json", StandardCharsets.UTF_8)); this.mapper = mapper; this.postProcessor = postProcessor; } @Override - public Flux encode(Publisher inputStream, - ResolvableType type, MimeType mimeType, Object... hints) { + public Flux encode(Publisher inputStream, + DataBufferAllocator allocator, ResolvableType type, MimeType mimeType, + Object... hints) { Publisher stream = (inputStream instanceof Mono ? - ((Mono)inputStream).map(this::serialize) : - Flux.from(inputStream).map(this::serialize)); - return (this.postProcessor == null ? Flux.from(stream) : this.postProcessor.encode(stream, type, mimeType, hints)); + ((Mono) inputStream).map(value -> serialize(value, allocator)) : + Flux.from(inputStream).map(value -> serialize(value, allocator))); + // TODO: figure out why using the parameter allocator for the postprocessor + // commits the response too early + DefaultDataBufferAllocator tempAllocator = new DefaultDataBufferAllocator(); + + return (this.postProcessor == null ? Flux.from(stream) : + this.postProcessor.encode(stream, tempAllocator, type, mimeType, hints)); } - private DataBuffer serialize(Object value) { - DataBuffer buffer = allocator().allocateBuffer(); + private DataBuffer serialize(Object value, DataBufferAllocator allocator) { + DataBuffer buffer = allocator.allocateBuffer(); OutputStream outputStream = buffer.asOutputStream(); try { this.mapper.writeValue(outputStream, value); diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Encoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Encoder.java index 652c7e3cdb..5c95b9a00f 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Encoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Encoder.java @@ -43,23 +43,22 @@ import org.springframework.util.MimeTypeUtils; * @author Sebastien Deleuze * @see Jaxb2Decoder */ -public class Jaxb2Encoder extends AbstractAllocatingEncoder { +public class Jaxb2Encoder extends AbstractEncoder { private final ConcurrentMap, JAXBContext> jaxbContexts = new ConcurrentHashMap<>(64); - public Jaxb2Encoder(DataBufferAllocator allocator) { - super(allocator, MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML); + public Jaxb2Encoder() { + super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML); } - @Override - public Flux encode(Publisher messageStream, - ResolvableType type, - MimeType mimeType, Object... hints) { + public Flux encode(Publisher inputStream, + DataBufferAllocator allocator, ResolvableType type, MimeType mimeType, + Object... hints) { - return Flux.from(messageStream).map(value -> { + return Flux.from(inputStream).map(value -> { try { - DataBuffer buffer = allocator().allocateBuffer(1024); + DataBuffer buffer = allocator.allocateBuffer(1024); OutputStream outputStream = buffer.asOutputStream(); Class clazz = ClassUtils.getUserClass(value); Marshaller marshaller = createMarshaller(clazz); diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectEncoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectEncoder.java index 29e58fb4ff..804856bb1e 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectEncoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectEncoder.java @@ -42,21 +42,22 @@ import org.springframework.util.MimeType; * * @see JsonObjectDecoder */ -public class JsonObjectEncoder extends AbstractAllocatingEncoder { +public class JsonObjectEncoder extends AbstractEncoder { - public JsonObjectEncoder(DataBufferAllocator allocator) { - super(allocator, new MimeType("application", "json", StandardCharsets.UTF_8), + public JsonObjectEncoder() { + super(new MimeType("application", "json", StandardCharsets.UTF_8), new MimeType("application", "*+json", StandardCharsets.UTF_8)); } @Override public Flux encode(Publisher inputStream, + DataBufferAllocator allocator, ResolvableType type, MimeType mimeType, Object... hints) { if (inputStream instanceof Mono) { return Flux.from(inputStream); } return Flux.from(inputStream) - .lift(s -> new JsonArrayEncoderBarrier(s, allocator())); + .lift(s -> new JsonArrayEncoderBarrier(s, allocator)); } private static class JsonArrayEncoderBarrier diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringEncoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringEncoder.java index e7a80b693e..810d6de311 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringEncoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringEncoder.java @@ -33,12 +33,12 @@ import org.springframework.util.MimeType; * @author Sebastien Deleuze * @see StringDecoder */ -public class StringEncoder extends AbstractAllocatingEncoder { +public class StringEncoder extends AbstractEncoder { public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8; - public StringEncoder(DataBufferAllocator allocator) { - super(allocator, new MimeType("text", "plain", DEFAULT_CHARSET)); + public StringEncoder() { + super(new MimeType("text", "plain", DEFAULT_CHARSET)); } @@ -49,9 +49,9 @@ public class StringEncoder extends AbstractAllocatingEncoder { } @Override - public Flux encode(Publisher elementStream, - ResolvableType type, MimeType mimeType, Object... hints) { - + public Flux encode(Publisher inputStream, + DataBufferAllocator allocator, ResolvableType type, MimeType mimeType, + Object... hints) { Charset charset; if (mimeType != null && mimeType.getCharSet() != null) { charset = mimeType.getCharSet(); @@ -59,9 +59,9 @@ public class StringEncoder extends AbstractAllocatingEncoder { else { charset = DEFAULT_CHARSET; } - return Flux.from(elementStream).map(s -> { + return Flux.from(inputStream).map(s -> { byte[] bytes = s.getBytes(charset); - DataBuffer dataBuffer = allocator().allocateBuffer(bytes.length); + DataBuffer dataBuffer = allocator.allocateBuffer(bytes.length); dataBuffer.write(bytes); return dataBuffer; }); diff --git a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java index 3e95dcef1f..9e238c50ad 100644 --- a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java @@ -23,13 +23,13 @@ import org.junit.Before; import org.junit.Test; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; +import reactor.core.test.TestSubscriber; import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; import static org.junit.Assert.*; -import reactor.core.test.TestSubscriber; /** * @author Sebastien Deleuze @@ -40,7 +40,7 @@ public class ByteBufferEncoderTests extends AbstractAllocatingTestCase { @Before public void createEncoder() { - encoder = new ByteBufferEncoder(allocator); + encoder = new ByteBufferEncoder(); } @Test @@ -57,7 +57,7 @@ public class ByteBufferEncoderTests extends AbstractAllocatingTestCase { Flux source = Flux.just(ByteBuffer.wrap(fooBytes), ByteBuffer.wrap(barBytes)); - Flux output = encoder.encode(source, + Flux output = encoder.encode(source, allocator, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null); TestSubscriber testSubscriber = new TestSubscriber<>(); diff --git a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java index e61dc616dc..53383547f4 100644 --- a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java @@ -21,11 +21,12 @@ import java.nio.charset.StandardCharsets; import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; +import reactor.core.test.TestSubscriber; import org.springframework.http.MediaType; -import static org.junit.Assert.*; -import reactor.core.test.TestSubscriber; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; /** * @author Sebastien Deleuze @@ -36,7 +37,7 @@ public class JacksonJsonEncoderTests extends AbstractAllocatingTestCase { @Before public void createEncoder() { - encoder = new JacksonJsonEncoder(allocator); + encoder = new JacksonJsonEncoder(); } @Test @@ -48,7 +49,7 @@ public class JacksonJsonEncoderTests extends AbstractAllocatingTestCase { @Test public void write() { Flux source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar")); - Flux output = encoder.encode(source, null, null).map(chunk -> { + Flux output = encoder.encode(source, allocator, null, null).map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8); diff --git a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java index a9ef2ea615..0f7a802962 100644 --- a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java @@ -21,11 +21,12 @@ import java.nio.charset.StandardCharsets; import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; +import reactor.core.test.TestSubscriber; import org.springframework.http.MediaType; -import static org.junit.Assert.*; -import reactor.core.test.TestSubscriber; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; /** * @author Sebastien Deleuze @@ -36,7 +37,7 @@ public class Jaxb2EncoderTests extends AbstractAllocatingTestCase { @Before public void createEncoder() { - encoder = new Jaxb2Encoder(allocator); + encoder = new Jaxb2Encoder(); } @Test @@ -49,7 +50,7 @@ public class Jaxb2EncoderTests extends AbstractAllocatingTestCase { @Test public void encode() { Flux source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar")); - Flux output = encoder.encode(source, null, null).map(chunk -> { + Flux output = encoder.encode(source, allocator, null, null).map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8); diff --git a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java index 64116126e9..0d4c9f1add 100644 --- a/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java @@ -22,11 +22,10 @@ import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.test.TestSubscriber; import org.springframework.core.io.buffer.DataBuffer; -import reactor.core.test.TestSubscriber; - /** * @author Sebastien Deleuze */ @@ -36,14 +35,15 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase { @Before public void createEncoder() { - encoder = new JsonObjectEncoder(allocator); + encoder = new JsonObjectEncoder(); } @Test public void encodeSingleElementFlux() throws InterruptedException { Flux source = Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")); - Flux output = Flux.from(encoder.encode(source, null, null)).map(chunk -> { + Flux output = + Flux.from(encoder.encode(source, allocator, null, null)).map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8); @@ -57,7 +57,8 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase { public void encodeSingleElementMono() throws InterruptedException { Mono source = Mono.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")); - Flux output = Flux.from(encoder.encode(source, null, null)).map(chunk -> { + Flux output = + Flux.from(encoder.encode(source, allocator, null, null)).map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8); @@ -72,7 +73,8 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase { Flux source = Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"), stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}")); - Flux output = Flux.from(encoder.encode(source, null, null)).map(chunk -> { + Flux output = + Flux.from(encoder.encode(source, allocator, null, null)).map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8); @@ -91,7 +93,8 @@ public class JsonObjectEncoderTests extends AbstractAllocatingTestCase { stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}"), stringBuffer("{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}") ); - Flux output = Flux.from(encoder.encode(source, null, null)).map(chunk -> { + Flux output = + Flux.from(encoder.encode(source, allocator, null, null)).map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8); 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 415f2883d2..d69c284638 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 @@ -23,12 +23,13 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; +import reactor.core.test.TestSubscriber; import org.springframework.core.ResolvableType; import org.springframework.http.MediaType; -import static org.junit.Assert.*; -import reactor.core.test.TestSubscriber; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; /** * @author Sebastien Deleuze @@ -40,7 +41,7 @@ public class StringEncoderTests extends AbstractAllocatingTestCase { @Before public void createEncoder() { - encoder = new StringEncoder(allocator); + encoder = new StringEncoder(); } @Test @@ -52,7 +53,9 @@ public class StringEncoderTests extends AbstractAllocatingTestCase { @Test public void write() throws InterruptedException { - Flux output = Flux.from(encoder.encode(Flux.just("foo"), null, null)).map(chunk -> { + Flux output = + Flux.from(encoder.encode(Flux.just("foo"), allocator, null, null)) + .map(chunk -> { byte[] b = new byte[chunk.readableByteCount()]; chunk.read(b); return new String(b, StandardCharsets.UTF_8);