From 2981b5e6e88b25a4f9b743e34e815e2bac6cfc37 Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Tue, 26 Jan 2016 14:45:08 +0100 Subject: [PATCH] Updated Encoder and Decoder to use DataBuffer --- .../springframework/core/codec/Decoder.java | 8 +- .../springframework/core/codec/Encoder.java | 5 +- .../support/AbstractAllocatingEncoder.java | 42 ++++++++++ .../support/AbstractRawByteStreamDecoder.java | 78 +++++++++++-------- .../core/codec/support/ByteBufferDecoder.java | 8 +- .../core/codec/support/ByteBufferEncoder.java | 20 +++-- .../codec/support/JacksonJsonDecoder.java | 18 ++--- .../codec/support/JacksonJsonEncoder.java | 41 +++++----- .../core/codec/support/Jaxb2Decoder.java | 11 +-- .../core/codec/support/Jaxb2Encoder.java | 25 +++--- .../core/codec/support/JsonObjectDecoder.java | 45 ++++++----- .../core/codec/support/JsonObjectEncoder.java | 54 ++++++------- .../core/codec/support/StringDecoder.java | 30 ++++--- .../core/codec/support/StringEncoder.java | 21 +++-- .../support/AbstractAllocatingTestCase.java | 59 ++++++++++++++ .../support}/ByteBufferDecoderTests.java | 29 ++++--- .../support}/ByteBufferEncoderTests.java | 51 ++++++++---- .../support}/JacksonJsonDecoderTests.java | 13 ++-- .../support}/JacksonJsonEncoderTests.java | 20 +++-- .../codec/support}/Jaxb2DecoderTests.java | 13 ++-- .../codec/support}/Jaxb2EncoderTests.java | 20 +++-- .../support}/JsonObjectDecoderTests.java | 67 ++++++++-------- .../support}/JsonObjectEncoderTests.java | 60 +++++++------- .../codec => core/codec/support}/Pojo.java | 20 ++++- .../codec/support}/StringDecoderTests.java | 26 ++++--- .../codec/support}/StringEncoderTests.java | 22 ++++-- 26 files changed, 509 insertions(+), 297 deletions(-) create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractAllocatingEncoder.java create mode 100644 spring-web-reactive/src/test/java/org/springframework/core/codec/support/AbstractAllocatingTestCase.java rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/decoder => core/codec/support}/ByteBufferDecoderTests.java (68%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/encoder => core/codec/support}/ByteBufferEncoderTests.java (51%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/decoder => core/codec/support}/JacksonJsonDecoderTests.java (79%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/encoder => core/codec/support}/JacksonJsonEncoderTests.java (80%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/decoder => core/codec/support}/Jaxb2DecoderTests.java (78%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/encoder => core/codec/support}/Jaxb2EncoderTests.java (83%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/decoder => core/codec/support}/JsonObjectDecoderTests.java (54%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/encoder => core/codec/support}/JsonObjectEncoderTests.java (62%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec => core/codec/support}/Pojo.java (71%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/decoder => core/codec/support}/StringDecoderTests.java (80%) rename spring-web-reactive/src/test/java/org/springframework/{reactive/codec/encoder => core/codec/support}/StringEncoderTests.java (77%) diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/Decoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/Decoder.java index 7b3325110a..1b3d0410bb 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/Decoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/Decoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -16,13 +16,13 @@ package org.springframework.core.codec; -import java.nio.ByteBuffer; import java.util.List; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.util.MimeType; /** @@ -43,14 +43,14 @@ public interface Decoder { boolean canDecode(ResolvableType type, MimeType mimeType, Object... hints); /** - * Decode an input {@link ByteBuffer} stream to an output stream of {@code T}. + * Decode an input {@link DataBuffer} stream to an output stream of {@code T}. * @param inputStream the input stream to process. * @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 decode(Publisher inputStream, ResolvableType type, + Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints); /** 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 761ceb9b4b..0f82c73081 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -23,6 +23,7 @@ import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.util.MimeType; /** @@ -50,7 +51,7 @@ public interface Encoder { * @param hints Additional information about how to do decode, optional. * @return the output stream */ - Flux encode(Publisher inputStream, ResolvableType type, + Flux encode(Publisher inputStream, 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 new file mode 100644 index 0000000000..0275931fc8 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractAllocatingEncoder.java @@ -0,0 +1,42 @@ +/* + * 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/AbstractRawByteStreamDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractRawByteStreamDecoder.java index a77141afca..7fc77d8137 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractRawByteStreamDecoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractRawByteStreamDecoder.java @@ -16,7 +16,6 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLongFieldUpdater; @@ -28,6 +27,9 @@ import reactor.core.util.BackpressureUtils; import org.springframework.core.ResolvableType; import org.springframework.core.codec.Decoder; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; +import org.springframework.util.Assert; import org.springframework.util.MimeType; /** @@ -38,12 +40,19 @@ import org.springframework.util.MimeType; */ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder { - public AbstractRawByteStreamDecoder(MimeType... supportedMimeTypes) { + private final DataBufferAllocator allocator; + + public AbstractRawByteStreamDecoder(DataBufferAllocator allocator, + MimeType... supportedMimeTypes) { super(supportedMimeTypes); + Assert.notNull(allocator, "'allocator' must not be null"); + + this.allocator = allocator; } @Override - public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { + public Flux decode(Publisher inputStream, ResolvableType type, + MimeType mimeType, Object... hints) { return decodeInternal(Flux.from(inputStream).lift(bbs -> subscriberBarrier(bbs)), type, mimeType, hints); @@ -55,17 +64,20 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder *

Implementations should provide their own {@link SubscriberBarrier} or use one of the * provided implementations by this class */ - public abstract SubscriberBarrier subscriberBarrier(Subscriber subscriber); + public abstract SubscriberBarrier subscriberBarrier( + Subscriber subscriber); - public abstract Flux decodeInternal(Publisher inputStream, ResolvableType type + public abstract Flux decodeInternal(Publisher inputStream, + ResolvableType type , MimeType mimeType, Object... hints); /** * {@code SubscriberBarrier} implementation that buffers all received elements and emits a single - * {@code ByteBuffer} once the incoming stream has been completed + * {@code DataBuffer} once the incoming stream has been completed */ - public static class ReduceSingleByteStreamBarrier extends SubscriberBarrier { + public static class ReduceSingleByteStreamBarrier + extends SubscriberBarrier { @SuppressWarnings("rawtypes") static final AtomicLongFieldUpdater REQUESTED = @@ -74,16 +86,16 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder static final AtomicIntegerFieldUpdater TERMINATED = AtomicIntegerFieldUpdater.newUpdater(ReduceSingleByteStreamBarrier.class, "terminated"); - private volatile long requested; private volatile int terminated; - private ByteBuffer buffer; + private DataBuffer buffer; - public ReduceSingleByteStreamBarrier(Subscriber subscriber) { + public ReduceSingleByteStreamBarrier(Subscriber subscriber, + DataBufferAllocator allocator) { super(subscriber); - this.buffer = ByteBuffer.allocate(0); + this.buffer = allocator.allocateBuffer(); } @Override @@ -108,15 +120,12 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder * TODO: when available, wrap buffers with a single buffer and avoid copying data for every method call. */ @Override - protected void doNext(ByteBuffer byteBuffer) { - this.buffer = ByteBuffer.allocate(this.buffer.capacity() + byteBuffer.capacity()) - .put(this.buffer).put(byteBuffer); - this.buffer.flip(); + protected void doNext(DataBuffer dataBuffer) { + this.buffer.write(dataBuffer); } protected void drainLast() { if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) { - this.buffer.flip(); subscriber.onNext(this.buffer); super.doComplete(); } @@ -127,7 +136,8 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder * {@code SubscriberBarrier} implementation that splits incoming elements * using line return delimiters: {@code "\n"} and {@code "\r\n"} */ - public static class SplitLinesByteStreamBarrier extends SubscriberBarrier { + public static class SplitLinesByteStreamBarrier + extends SubscriberBarrier { @SuppressWarnings("rawtypes") static final AtomicLongFieldUpdater REQUESTED = @@ -136,16 +146,20 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder static final AtomicIntegerFieldUpdater TERMINATED = AtomicIntegerFieldUpdater.newUpdater(SplitLinesByteStreamBarrier.class, "terminated"); + private final DataBufferAllocator allocator; + private volatile long requested; private volatile int terminated; - private ByteBuffer buffer; + private DataBuffer buffer; - public SplitLinesByteStreamBarrier(Subscriber subscriber) { + public SplitLinesByteStreamBarrier(Subscriber subscriber, + DataBufferAllocator allocator) { super(subscriber); - this.buffer = ByteBuffer.allocate(0); + this.allocator = allocator; + this.buffer = allocator.allocateBuffer(); } @Override @@ -170,19 +184,20 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder * TODO: when available, wrap buffers with a single buffer and avoid copying data for every method call. */ @Override - protected void doNext(ByteBuffer byteBuffer) { - this.buffer = ByteBuffer.allocate(this.buffer.capacity() + byteBuffer.capacity()) - .put(this.buffer).put(byteBuffer); + protected void doNext(DataBuffer dataBuffer) { + this.buffer.write(dataBuffer); while (REQUESTED.get(this) > 0) { int separatorIndex = findEndOfLine(this.buffer); if (separatorIndex != -1) { if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) { byte[] message = new byte[separatorIndex]; - this.buffer.get(message); + this.buffer.read(message); consumeSeparator(this.buffer); - this.buffer = this.buffer.slice(); - super.doNext(ByteBuffer.wrap(message)); +// this.buffer = this.buffer.slice(); + DataBuffer buffer2 = allocator.allocateBuffer(message.length); + buffer2.write(message); + super.doNext(buffer2); } } else { @@ -191,9 +206,9 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder } } - protected int findEndOfLine(ByteBuffer buffer) { + protected int findEndOfLine(DataBuffer buffer) { - final int n = buffer.limit(); + final int n = buffer.readableByteCount(); for (int i = 0; i < n; i++) { final byte b = buffer.get(i); if (b == '\n') { @@ -207,16 +222,15 @@ public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder return -1; } - protected void consumeSeparator(ByteBuffer buffer) { - byte sep = buffer.get(); + protected void consumeSeparator(DataBuffer buffer) { + byte sep = buffer.read(); if (sep == '\r') { - buffer.get(); + buffer.read(); } } protected void drainLast() { if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) { - this.buffer.flip(); subscriber.onNext(this.buffer); super.doComplete(); } diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferDecoder.java index 42595a351c..bea4ad9a1f 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferDecoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/ByteBufferDecoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -22,6 +22,7 @@ import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -43,10 +44,9 @@ public class ByteBufferDecoder extends AbstractDecoder { } @Override - public Flux decode(Publisher inputStream, ResolvableType type, + public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { - - return Flux.from(inputStream); + return Flux.from(inputStream).map(DataBuffer::asByteBuffer); } } \ No newline at end of file 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 7a76eb9232..3b72c7e355 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -22,17 +22,18 @@ import org.reactivestreams.Publisher; 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; import org.springframework.util.MimeTypeUtils; /** * @author Sebastien Deleuze */ -public class ByteBufferEncoder extends AbstractEncoder { +public class ByteBufferEncoder extends AbstractAllocatingEncoder { - - public ByteBufferEncoder() { - super(MimeTypeUtils.ALL); + public ByteBufferEncoder(DataBufferAllocator allocator) { + super(allocator, MimeTypeUtils.ALL); } @@ -43,11 +44,16 @@ public class ByteBufferEncoder extends AbstractEncoder { } @Override - public Flux encode(Publisher inputStream, ResolvableType type, + public Flux encode(Publisher inputStream, + ResolvableType type, MimeType mimeType, Object... hints) { //noinspection unchecked - return Flux.from(inputStream); + return Flux.from(inputStream).map(byteBuffer -> { + DataBuffer dataBuffer = allocator().allocateBuffer(byteBuffer.remaining()); + dataBuffer.write(byteBuffer); + return dataBuffer; + }); } } \ No newline at end of file diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonDecoder.java index e60a59a6cf..05f6dbf2c5 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonDecoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonDecoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -17,7 +17,6 @@ package org.springframework.core.codec.support; import java.io.IOException; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import com.fasterxml.jackson.databind.ObjectMapper; @@ -28,7 +27,7 @@ import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; import org.springframework.core.codec.CodecException; import org.springframework.core.codec.Decoder; -import org.springframework.util.ByteBufferInputStream; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.util.MimeType; @@ -42,39 +41,38 @@ public class JacksonJsonDecoder extends AbstractDecoder { private final ObjectMapper mapper; - private Decoder preProcessor; + private Decoder preProcessor; public JacksonJsonDecoder() { this(new ObjectMapper(), null); } - public JacksonJsonDecoder(Decoder preProcessor) { + public JacksonJsonDecoder(Decoder preProcessor) { this(new ObjectMapper(), preProcessor); } - public JacksonJsonDecoder(ObjectMapper mapper, Decoder preProcessor) { + public JacksonJsonDecoder(ObjectMapper mapper, Decoder preProcessor) { super(new MimeType("application", "json", StandardCharsets.UTF_8), new MimeType("application", "*+json", StandardCharsets.UTF_8)); this.mapper = mapper; this.preProcessor = preProcessor; } - @Override - public Flux decode(Publisher inputStream, ResolvableType type, + public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { ObjectReader reader = this.mapper.readerFor(type.getRawClass()); - Flux stream = Flux.from(inputStream); + Flux stream = Flux.from(inputStream); if (this.preProcessor != null) { stream = this.preProcessor.decode(inputStream, type, mimeType, hints); } return stream.map(content -> { try { - return reader.readValue(new ByteBufferInputStream(content)); + return reader.readValue(content.asInputStream()); } catch (IOException e) { throw new CodecException("Error while reading the data", e); 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 a42d655e95..6e86446f8a 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -17,19 +17,19 @@ package org.springframework.core.codec.support; import java.io.IOException; -import java.nio.ByteBuffer; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; import com.fasterxml.jackson.databind.ObjectMapper; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; import org.springframework.core.codec.CodecException; import org.springframework.core.codec.Encoder; -import org.springframework.util.BufferOutputStream; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.util.MimeType; /** @@ -38,50 +38,49 @@ import org.springframework.util.MimeType; * @author Sebastien Deleuze * @see JacksonJsonDecoder */ -public class JacksonJsonEncoder extends AbstractEncoder { +public class JacksonJsonEncoder extends AbstractAllocatingEncoder { private final ObjectMapper mapper; - private Encoder postProcessor; + private Encoder postProcessor; - - public JacksonJsonEncoder() { - this(new ObjectMapper(), null); + public JacksonJsonEncoder(DataBufferAllocator allocator) { + this(allocator, new ObjectMapper(), null); } - public JacksonJsonEncoder(Encoder postProcessor) { - this(new ObjectMapper(), postProcessor); + public JacksonJsonEncoder(DataBufferAllocator allocator, + Encoder postProcessor) { + this(allocator, new ObjectMapper(), postProcessor); } - - public JacksonJsonEncoder(ObjectMapper mapper, Encoder postProcessor) { - super(new MimeType("application", "json", StandardCharsets.UTF_8), + public JacksonJsonEncoder(DataBufferAllocator allocator, ObjectMapper mapper, + Encoder postProcessor) { + super(allocator, 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, + public Flux encode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { - Publisher stream = (inputStream instanceof Mono ? + 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)); } - private ByteBuffer serialize(Object value) { - Buffer buffer = new Buffer(); - BufferOutputStream outputStream = new BufferOutputStream(buffer); + private DataBuffer serialize(Object value) { + DataBuffer buffer = allocator().allocateBuffer(); + OutputStream outputStream = buffer.asOutputStream(); try { this.mapper.writeValue(outputStream, value); } catch (IOException e) { throw new CodecException("Error while writing the data", e); } - buffer.flip(); - return buffer.byteBuffer(); + return buffer; } } diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Decoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Decoder.java index 94fda4c8ca..ec1e260fe9 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Decoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/Jaxb2Decoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -16,7 +16,6 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import javax.xml.bind.JAXBContext; @@ -38,8 +37,9 @@ import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; import org.springframework.core.codec.CodecException; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.support.DataBufferUtils; import org.springframework.util.Assert; -import org.springframework.util.ByteBufferPublisherInputStream; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -60,12 +60,13 @@ public class Jaxb2Decoder extends AbstractDecoder { @Override - public Flux decode(Publisher inputStream, ResolvableType type, + public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { Class outputClass = type.getRawClass(); try { - Source source = processSource(new StreamSource(new ByteBufferPublisherInputStream(inputStream))); + Source source = processSource( + new StreamSource(DataBufferUtils.toInputStream(inputStream))); Unmarshaller unmarshaller = createUnmarshaller(outputClass); if (outputClass.isAnnotationPresent(XmlRootElement.class)) { return Flux.just(unmarshaller.unmarshal(source)); 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 2482dd0fd0..652c7e3cdb 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -16,7 +16,7 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -27,12 +27,12 @@ import javax.xml.bind.Marshaller; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; -import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; import org.springframework.core.codec.CodecException; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.util.Assert; -import org.springframework.util.BufferOutputStream; import org.springframework.util.ClassUtils; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -43,30 +43,29 @@ import org.springframework.util.MimeTypeUtils; * @author Sebastien Deleuze * @see Jaxb2Decoder */ -public class Jaxb2Encoder extends AbstractEncoder { +public class Jaxb2Encoder extends AbstractAllocatingEncoder { private final ConcurrentMap, JAXBContext> jaxbContexts = new ConcurrentHashMap<>(64); - - public Jaxb2Encoder() { - super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML); + public Jaxb2Encoder(DataBufferAllocator allocator) { + super(allocator, MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML); } @Override - public Flux encode(Publisher messageStream, ResolvableType type, + public Flux encode(Publisher messageStream, + ResolvableType type, MimeType mimeType, Object... hints) { return Flux.from(messageStream).map(value -> { try { - Buffer buffer = new Buffer(); - BufferOutputStream outputStream = new BufferOutputStream(buffer); + DataBuffer buffer = allocator().allocateBuffer(1024); + OutputStream outputStream = buffer.asOutputStream(); Class clazz = ClassUtils.getUserClass(value); Marshaller marshaller = createMarshaller(clazz); marshaller.setProperty(Marshaller.JAXB_ENCODING, StandardCharsets.UTF_8.name()); marshaller.marshal(value, outputStream); - buffer.flip(); - return buffer.byteBuffer(); + return buffer; } catch (MarshalException ex) { throw new CodecException("Could not marshal [" + value + "]: " + ex.getMessage(), ex); diff --git a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectDecoder.java index 127a4ba74b..2bdc000320 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectDecoder.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectDecoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -16,7 +16,6 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; @@ -29,6 +28,8 @@ import reactor.core.publisher.Flux; import reactor.fn.Function; 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,7 +45,7 @@ import org.springframework.util.MimeType; * @author Sebastien Deleuze * @see JsonObjectEncoder */ -public class JsonObjectDecoder extends AbstractDecoder { +public class JsonObjectDecoder extends AbstractDecoder { private static final int ST_CORRUPTED = -1; @@ -54,38 +55,40 @@ public class JsonObjectDecoder extends AbstractDecoder { private static final int ST_DECODING_ARRAY_STREAM = 2; + private final DataBufferAllocator allocator; private final int maxObjectLength; private final boolean streamArrayElements; - - public JsonObjectDecoder() { + public JsonObjectDecoder(DataBufferAllocator allocator) { // 1 MB - this(1024 * 1024); + this(allocator, 1024 * 1024); } - public JsonObjectDecoder(int maxObjectLength) { - this(maxObjectLength, true); + public JsonObjectDecoder(DataBufferAllocator allocator, int maxObjectLength) { + this(allocator, maxObjectLength, true); } - public JsonObjectDecoder(boolean streamArrayElements) { - this(1024 * 1024, streamArrayElements); + public JsonObjectDecoder(DataBufferAllocator allocator, boolean streamArrayElements) { + this(allocator, 1024 * 1024, streamArrayElements); } /** + * @param allocator * @param maxObjectLength maximum number of bytes a JSON object/array may * use (including braces and all). Objects exceeding this length are dropped * and an {@link IllegalStateException} is thrown. * @param streamArrayElements if set to true and the "top level" JSON object * is an array, each of its entries is passed through the pipeline individually * and immediately after it was fully received, allowing for arrays with - * "infinitely" many elements. */ - public JsonObjectDecoder(int maxObjectLength, boolean streamArrayElements) { + public JsonObjectDecoder(DataBufferAllocator allocator, int maxObjectLength, + boolean streamArrayElements) { super(new MimeType("application", "json", StandardCharsets.UTF_8), new MimeType("application", "*+json", StandardCharsets.UTF_8)); + this.allocator = allocator; if (maxObjectLength < 1) { throw new IllegalArgumentException("maxObjectLength must be a positive int"); } @@ -94,10 +97,11 @@ public class JsonObjectDecoder extends AbstractDecoder { } @Override - public Flux decode(Publisher inputStream, ResolvableType type, + public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { - return Flux.from(inputStream).flatMap(new Function>() { + return Flux.from(inputStream) + .flatMap(new Function>() { int openBraces; int index; @@ -107,14 +111,15 @@ public class JsonObjectDecoder extends AbstractDecoder { Integer writerIndex; @Override - public Publisher apply(ByteBuffer b) { - List chunks = new ArrayList<>(); + public Publisher apply(DataBuffer b) { + List chunks = new ArrayList<>(); if (this.input == null) { - this.input = Unpooled.copiedBuffer(b); + this.input = Unpooled.copiedBuffer(b.asByteBuffer()); this.writerIndex = this.input.writerIndex(); } else { - this.input = Unpooled.copiedBuffer(this.input, Unpooled.copiedBuffer(b)); + this.input = Unpooled.copiedBuffer(this.input, + Unpooled.copiedBuffer(b.asByteBuffer())); this.writerIndex = this.input.writerIndex(); } if (this.state == ST_CORRUPTED) { @@ -139,7 +144,7 @@ public class JsonObjectDecoder extends AbstractDecoder { ByteBuf json = extractObject(this.input, this.input.readerIndex(), this.index + 1 - this.input.readerIndex()); if (json != null) { - chunks.add(json.nioBuffer()); + chunks.add(allocator.wrap(json.nioBuffer())); } // The JSON object/array was extracted => discard the bytes from @@ -173,7 +178,7 @@ public class JsonObjectDecoder extends AbstractDecoder { idxNoSpaces + 1 - this.input.readerIndex()); if (json != null) { - chunks.add(json.nioBuffer()); + chunks.add(allocator.wrap(json.nioBuffer())); } this.input.readerIndex(this.index + 1); 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 a2c5c5d34c..29e58fb4ff 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -16,7 +16,6 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLongFieldUpdater; @@ -27,9 +26,10 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.subscriber.SubscriberBarrier; import reactor.core.util.BackpressureUtils; -import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.util.MimeType; /** @@ -42,25 +42,25 @@ import org.springframework.util.MimeType; * * @see JsonObjectDecoder */ -public class JsonObjectEncoder extends AbstractEncoder { +public class JsonObjectEncoder extends AbstractAllocatingEncoder { - public JsonObjectEncoder() { - super(new MimeType("application", "json", StandardCharsets.UTF_8), + public JsonObjectEncoder(DataBufferAllocator allocator) { + super(allocator, new MimeType("application", "json", StandardCharsets.UTF_8), new MimeType("application", "*+json", StandardCharsets.UTF_8)); } @Override - public Flux encode(Publisher inputStream, + public Flux encode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { - if (inputStream instanceof Mono) { return Flux.from(inputStream); } - return Flux.from(inputStream).lift(s -> new JsonArrayEncoderBarrier(s)); + return Flux.from(inputStream) + .lift(s -> new JsonArrayEncoderBarrier(s, allocator())); } - - private static class JsonArrayEncoderBarrier extends SubscriberBarrier { + private static class JsonArrayEncoderBarrier + extends SubscriberBarrier { @SuppressWarnings("rawtypes") static final AtomicLongFieldUpdater REQUESTED = @@ -69,8 +69,9 @@ public class JsonObjectEncoder extends AbstractEncoder { static final AtomicIntegerFieldUpdater TERMINATED = AtomicIntegerFieldUpdater.newUpdater(JsonArrayEncoderBarrier.class, "terminated"); + private final DataBufferAllocator allocator; - private ByteBuffer prev = null; + private DataBuffer prev = null; private long count = 0; @@ -78,9 +79,10 @@ public class JsonObjectEncoder extends AbstractEncoder { private volatile int terminated; - - public JsonArrayEncoderBarrier(Subscriber subscriber) { + public JsonArrayEncoderBarrier(Subscriber subscriber, + DataBufferAllocator allocator) { super(subscriber); + this.allocator = allocator; } @@ -96,34 +98,32 @@ public class JsonObjectEncoder extends AbstractEncoder { } @Override - protected void doNext(ByteBuffer next) { + protected void doNext(DataBuffer next) { this.count++; - ByteBuffer tmp = this.prev; + DataBuffer tmp = this.prev; this.prev = next; - Buffer buffer = new Buffer(); + DataBuffer buffer = allocator.allocateBuffer(); if (this.count == 1) { - buffer.append("["); + buffer.write((byte) '['); } if (tmp != null) { - buffer.append(tmp); + buffer.write(tmp); } if (this.count > 1) { - buffer.append(","); + buffer.write((byte) ','); } - buffer.flip(); BackpressureUtils.getAndSub(REQUESTED, this, 1L); - subscriber.onNext(buffer.byteBuffer()); + subscriber.onNext(buffer); } protected void drainLast(){ if(BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) { - Buffer buffer = new Buffer(); - buffer.append(this.prev); - buffer.append("]"); - buffer.flip(); - subscriber.onNext(buffer.byteBuffer()); + DataBuffer buffer = allocator.allocateBuffer(); + buffer.write(this.prev); + buffer.write((byte) ']'); + subscriber.onNext(buffer); super.doComplete(); } } 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 9785ed5cbb..7889e1593e 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 @@ -16,7 +16,6 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; @@ -26,6 +25,8 @@ import reactor.core.publisher.Flux; import reactor.core.subscriber.SubscriberBarrier; import org.springframework.core.ResolvableType; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; import org.springframework.util.MimeType; /** @@ -47,14 +48,16 @@ public class StringDecoder extends AbstractRawByteStreamDecoder { public final boolean reduceToSingleBuffer; + private final DataBufferAllocator allocator; + /** * 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. */ - public StringDecoder() { - this(true); + public StringDecoder(DataBufferAllocator allocator) { + this(allocator, true); } /** @@ -63,9 +66,10 @@ public class StringDecoder extends AbstractRawByteStreamDecoder { * @param reduceToSingleBuffer whether this decoder should buffer all received items * and decode a single consolidated String or re-emit items as they are provided */ - public StringDecoder(boolean reduceToSingleBuffer) { - super(new MimeType("text", "plain", DEFAULT_CHARSET)); + public StringDecoder(DataBufferAllocator allocator, boolean reduceToSingleBuffer) { + super(allocator, new MimeType("text", "plain", DEFAULT_CHARSET)); this.reduceToSingleBuffer = reduceToSingleBuffer; + this.allocator = allocator; } @Override @@ -75,18 +79,20 @@ public class StringDecoder extends AbstractRawByteStreamDecoder { } @Override - public SubscriberBarrier subscriberBarrier(Subscriber subscriber) { + public SubscriberBarrier subscriberBarrier( + Subscriber subscriber) { if (reduceToSingleBuffer) { - return new ReduceSingleByteStreamBarrier(subscriber); + return new ReduceSingleByteStreamBarrier(subscriber, allocator); } else { - return new SubscriberBarrier(subscriber); + return new SubscriberBarrier(subscriber); } } @Override - public Flux decodeInternal(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { + public Flux decodeInternal(Publisher inputStream, + ResolvableType type, MimeType mimeType, Object... hints) { Charset charset; if (mimeType != null && mimeType.getCharSet() != null) { charset = mimeType.getCharSet(); @@ -94,7 +100,11 @@ public class StringDecoder extends AbstractRawByteStreamDecoder { else { charset = DEFAULT_CHARSET; } - return Flux.from(inputStream).map(content -> new String(content.duplicate().array(), charset)); + return Flux.from(inputStream).map(content -> { + byte[] bytes = new byte[content.readableByteCount()]; + content.read(bytes); + return new String(bytes, charset); + }); } } 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 50ea51a26d..e7a80b693e 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -16,7 +16,6 @@ package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; @@ -24,6 +23,8 @@ import org.reactivestreams.Publisher; 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; /** @@ -32,13 +33,12 @@ import org.springframework.util.MimeType; * @author Sebastien Deleuze * @see StringDecoder */ -public class StringEncoder extends AbstractEncoder { +public class StringEncoder extends AbstractAllocatingEncoder { public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8; - - public StringEncoder() { - super(new MimeType("text", "plain", DEFAULT_CHARSET)); + public StringEncoder(DataBufferAllocator allocator) { + super(allocator, new MimeType("text", "plain", DEFAULT_CHARSET)); } @@ -49,7 +49,7 @@ public class StringEncoder extends AbstractEncoder { } @Override - public Flux encode(Publisher elementStream, + public Flux encode(Publisher elementStream, ResolvableType type, MimeType mimeType, Object... hints) { Charset charset; @@ -59,7 +59,12 @@ public class StringEncoder extends AbstractEncoder { else { charset = DEFAULT_CHARSET; } - return Flux.from(elementStream).map(s -> ByteBuffer.wrap(s.getBytes(charset))); + return Flux.from(elementStream).map(s -> { + byte[] bytes = s.getBytes(charset); + 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/AbstractAllocatingTestCase.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/AbstractAllocatingTestCase.java new file mode 100644 index 0000000000..75710ac4d6 --- /dev/null +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/AbstractAllocatingTestCase.java @@ -0,0 +1,59 @@ +/* + * 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 java.nio.charset.StandardCharsets; + +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.UnpooledByteBufAllocator; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferAllocator; +import org.springframework.core.io.buffer.DefaultDataBufferAllocator; +import org.springframework.core.io.buffer.NettyDataBufferAllocator; + +/** + * @author Arjen Poutsma + */ +@RunWith(Parameterized.class) +public abstract class AbstractAllocatingTestCase { + + @Parameterized.Parameter + public DataBufferAllocator allocator; + + @Parameterized.Parameters(name = "{0}") + public static Object[][] allocators() { + return new Object[][]{ + {new NettyDataBufferAllocator(new UnpooledByteBufAllocator(true))}, + {new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false))}, + {new NettyDataBufferAllocator(new PooledByteBufAllocator(true))}, + {new NettyDataBufferAllocator(new PooledByteBufAllocator(false))}, + {new DefaultDataBufferAllocator(true)}, + {new DefaultDataBufferAllocator(false)} + + }; + } + + protected DataBuffer stringBuffer(String value) { + byte[] bytes = value.getBytes(StandardCharsets.UTF_8); + DataBuffer buffer = allocator.allocateBuffer(bytes.length); + buffer.write(bytes); + return buffer; + } +} diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/ByteBufferDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferDecoderTests.java similarity index 68% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/ByteBufferDecoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferDecoderTests.java index 85106d1130..6c59f5fb79 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/ByteBufferDecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferDecoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.reactive.codec.decoder; +package org.springframework.core.codec.support; import java.nio.ByteBuffer; import java.util.List; @@ -26,7 +26,7 @@ import reactor.core.publisher.Flux; import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.support.ByteBufferDecoder; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; import static java.util.stream.Collectors.toList; @@ -35,7 +35,7 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class ByteBufferDecoderTests { +public class ByteBufferDecoderTests extends AbstractAllocatingTestCase { private final ByteBufferDecoder decoder = new ByteBufferDecoder(); @@ -48,14 +48,25 @@ public class ByteBufferDecoderTests { @Test public void decode() throws InterruptedException { - ByteBuffer fooBuffer = Buffer.wrap("foo").byteBuffer(); - ByteBuffer barBuffer = Buffer.wrap("bar").byteBuffer(); - Flux source = Flux.just(fooBuffer, barBuffer); + DataBuffer fooBuffer = stringBuffer("foo"); + DataBuffer barBuffer = stringBuffer("bar"); + Flux source = Flux.just(fooBuffer, barBuffer); Flux output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(2, results.size()); - assertEquals(fooBuffer, results.get(0)); - assertEquals(barBuffer, results.get(1)); + + assertBufferEquals(fooBuffer, results.get(0)); + assertBufferEquals(barBuffer, results.get(1)); + } + + public void assertBufferEquals(DataBuffer expected, ByteBuffer actual) { + byte[] byteBufferBytes = new byte[actual.remaining()]; + actual.get(byteBufferBytes); + + byte[] dataBufferBytes = new byte[expected.readableByteCount()]; + expected.read(dataBufferBytes); + + assertArrayEquals(dataBufferBytes, byteBufferBytes); } } diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/ByteBufferEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java similarity index 51% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/ByteBufferEncoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java index 16eb7a5644..ae4e817d64 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/ByteBufferEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/ByteBufferEncoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,19 +14,20 @@ * limitations under the License. */ -package org.springframework.reactive.codec.encoder; +package org.springframework.core.codec.support; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.stream.StreamSupport; +import org.junit.Before; import org.junit.Test; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; -import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.support.ByteBufferEncoder; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; import static java.util.stream.Collectors.toList; @@ -35,27 +36,47 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class ByteBufferEncoderTests { +public class ByteBufferEncoderTests extends AbstractAllocatingTestCase { - private final ByteBufferEncoder encoder = new ByteBufferEncoder(); + private ByteBufferEncoder encoder; + + @Before + public void createEncoder() { + encoder = new ByteBufferEncoder(allocator); + } @Test - public void canDecode() { + public void canEncode() { assertTrue(encoder.canEncode(ResolvableType.forClass(ByteBuffer.class), MediaType.TEXT_PLAIN)); assertFalse(encoder.canEncode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN)); assertTrue(encoder.canEncode(ResolvableType.forClass(ByteBuffer.class), MediaType.APPLICATION_JSON)); } @Test - public void decode() throws InterruptedException { - ByteBuffer fooBuffer = Buffer.wrap("foo").byteBuffer(); - ByteBuffer barBuffer = Buffer.wrap("bar").byteBuffer(); - Flux source = Flux.just(fooBuffer, barBuffer); - Flux output = encoder.encode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null); - List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); + public void encode() throws Exception { + byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8); + byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8); + Flux source = + Flux.just(ByteBuffer.wrap(fooBytes), ByteBuffer.wrap(barBytes)); + + Flux output = encoder.encode(source, + ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), + null); + List results = + StreamSupport.stream(output.toIterable().spliterator(), false) + .collect(toList()); + assertEquals(2, results.size()); - assertEquals(fooBuffer, results.get(0)); - assertEquals(barBuffer, results.get(1)); + assertEquals(3, results.get(0).readableByteCount()); + assertEquals(3, results.get(1).readableByteCount()); + + byte[] buf = new byte[3]; + results.get(0).read(buf); + assertArrayEquals(fooBytes, buf); + + results.get(1).read(buf); + assertArrayEquals(barBytes, buf); + } } diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonDecoderTests.java similarity index 79% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonDecoderTests.java index 8a0e476c43..83d7ac9e9a 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonDecoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,9 +14,8 @@ * limitations under the License. */ -package org.springframework.reactive.codec.decoder; +package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.util.List; import java.util.stream.StreamSupport; @@ -25,9 +24,8 @@ import reactor.core.publisher.Flux; import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.support.JacksonJsonDecoder; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; -import org.springframework.reactive.codec.Pojo; import static java.util.stream.Collectors.toList; import static org.junit.Assert.*; @@ -35,7 +33,7 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class JacksonJsonDecoderTests { +public class JacksonJsonDecoderTests extends AbstractAllocatingTestCase { private final JacksonJsonDecoder decoder = new JacksonJsonDecoder(); @@ -47,7 +45,8 @@ public class JacksonJsonDecoderTests { @Test public void decode() throws InterruptedException { - Flux source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer()); + Flux source = + Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")); Flux output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(1, results.size()); diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java similarity index 80% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java index 18492f1d88..d31ad368c2 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,18 +14,17 @@ * limitations under the License. */ -package org.springframework.reactive.codec.encoder; +package org.springframework.core.codec.support; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.stream.StreamSupport; +import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; -import org.springframework.core.codec.support.JacksonJsonEncoder; import org.springframework.http.MediaType; -import org.springframework.reactive.codec.Pojo; import static java.util.stream.Collectors.toList; import static org.junit.Assert.*; @@ -33,9 +32,14 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class JacksonJsonEncoderTests { +public class JacksonJsonEncoderTests extends AbstractAllocatingTestCase { - private final JacksonJsonEncoder encoder = new JacksonJsonEncoder(); + private JacksonJsonEncoder encoder; + + @Before + public void createEncoder() { + encoder = new JacksonJsonEncoder(allocator); + } @Test public void canWrite() { @@ -47,8 +51,8 @@ public class JacksonJsonEncoderTests { public void write() throws InterruptedException { Flux source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar")); Flux output = encoder.encode(source, null, null).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2DecoderTests.java similarity index 78% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2DecoderTests.java index a50b061afc..eaf983e56e 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2DecoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,9 +14,8 @@ * limitations under the License. */ -package org.springframework.reactive.codec.decoder; +package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.util.List; import java.util.stream.StreamSupport; @@ -25,9 +24,8 @@ import reactor.core.publisher.Flux; import reactor.io.buffer.Buffer; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.support.Jaxb2Decoder; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; -import org.springframework.reactive.codec.Pojo; import static java.util.stream.Collectors.toList; import static org.junit.Assert.*; @@ -35,7 +33,7 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class Jaxb2DecoderTests { +public class Jaxb2DecoderTests extends AbstractAllocatingTestCase { private final Jaxb2Decoder decoder = new Jaxb2Decoder(); @@ -48,7 +46,8 @@ public class Jaxb2DecoderTests { @Test public void decode() throws InterruptedException { - Flux source = Flux.just(Buffer.wrap("barbarfoofoo").byteBuffer()); + Flux source = Flux.just(stringBuffer( + "barbarfoofoo")); Flux output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(1, results.size()); diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java similarity index 83% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java index dbbb849d6c..1d9dec12a4 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Jaxb2EncoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,18 +14,17 @@ * limitations under the License. */ -package org.springframework.reactive.codec.encoder; +package org.springframework.core.codec.support; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.stream.StreamSupport; +import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; -import org.springframework.core.codec.support.Jaxb2Encoder; import org.springframework.http.MediaType; -import org.springframework.reactive.codec.Pojo; import static java.util.stream.Collectors.toList; import static org.junit.Assert.*; @@ -33,9 +32,14 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class Jaxb2EncoderTests { +public class Jaxb2EncoderTests extends AbstractAllocatingTestCase { - private final Jaxb2Encoder encoder = new Jaxb2Encoder(); + private Jaxb2Encoder encoder; + + @Before + public void createEncoder() { + encoder = new Jaxb2Encoder(allocator); + } @Test public void canEncode() { @@ -48,8 +52,8 @@ public class Jaxb2EncoderTests { public void encode() throws InterruptedException { Flux source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar")); Flux output = encoder.encode(source, null, null).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectDecoderTests.java similarity index 54% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectDecoderTests.java index c0b27df9f8..3bb4f66f63 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectDecoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,18 +14,16 @@ * limitations under the License. */ -package org.springframework.reactive.codec.decoder; +package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.stream.StreamSupport; import org.junit.Test; import reactor.core.publisher.Flux; -import reactor.io.buffer.Buffer; -import org.springframework.core.codec.support.JsonObjectDecoder; +import org.springframework.core.io.buffer.DataBuffer; import static java.util.stream.Collectors.toList; import static org.junit.Assert.assertEquals; @@ -33,17 +31,16 @@ import static org.junit.Assert.assertEquals; /** * @author Sebastien Deleuze */ -public class JsonObjectDecoderTests { +public class JsonObjectDecoderTests extends AbstractAllocatingTestCase { + @Test public void decodeSingleChunkToJsonObject() throws InterruptedException { - JsonObjectDecoder decoder = new JsonObjectDecoder(); - Flux source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer()); - Flux output = decoder.decode(source, null, null).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); - return new String(b, StandardCharsets.UTF_8); - }); + JsonObjectDecoder decoder = new JsonObjectDecoder(allocator); + Flux source = + Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")); + Flux output = + decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(1, results.size()); assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0)); @@ -51,13 +48,11 @@ public class JsonObjectDecoderTests { @Test public void decodeMultipleChunksToJsonObject() throws InterruptedException { - JsonObjectDecoder decoder = new JsonObjectDecoder(); - Flux source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\"").byteBuffer(), Buffer.wrap(", \"bar\": \"barbar\"}").byteBuffer()); - Flux output = decoder.decode(source, null, null).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); - return new String(b, StandardCharsets.UTF_8); - }); + JsonObjectDecoder decoder = new JsonObjectDecoder(allocator); + Flux source = Flux.just(stringBuffer("{\"foo\": \"foofoo\""), + stringBuffer(", \"bar\": \"barbar\"}")); + Flux output = + decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(1, results.size()); assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0)); @@ -65,13 +60,12 @@ public class JsonObjectDecoderTests { @Test public void decodeSingleChunkToArray() throws InterruptedException { - JsonObjectDecoder decoder = new JsonObjectDecoder(); - Flux source = Flux.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer()); - Flux output = decoder.decode(source, null, null).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); - return new String(b, StandardCharsets.UTF_8); - }); + JsonObjectDecoder decoder = new JsonObjectDecoder(allocator); + Flux source = Flux.just(stringBuffer( + "[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]")); + Flux output = + decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString); + List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(2, results.size()); assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0)); @@ -80,17 +74,22 @@ public class JsonObjectDecoderTests { @Test public void decodeMultipleChunksToArray() throws InterruptedException { - JsonObjectDecoder decoder = new JsonObjectDecoder(); - Flux source = Flux.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\"").byteBuffer(), Buffer.wrap(": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer()); - Flux output = decoder.decode(source, null, null).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); - return new String(b, StandardCharsets.UTF_8); - }); + JsonObjectDecoder decoder = new JsonObjectDecoder(allocator); + Flux source = + Flux.just(stringBuffer("[{\"foo\": \"foofoo\", \"bar\""), stringBuffer( + ": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]")); + Flux output = + decoder.decode(source, null, null).map(JsonObjectDecoderTests::toString); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(2, results.size()); assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0)); assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1)); } + private static String toString(DataBuffer buffer) { + byte[] b = new byte[buffer.readableByteCount()]; + buffer.read(b); + return new String(b, StandardCharsets.UTF_8); + } + } diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java similarity index 62% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java index ee6ac9f07f..04034ea0b1 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/JsonObjectEncoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,45 +14,52 @@ * limitations under the License. */ -package org.springframework.reactive.codec.encoder; +package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.io.buffer.Buffer; -import org.springframework.core.codec.support.JsonObjectEncoder; +import org.springframework.core.io.buffer.DataBuffer; import static org.junit.Assert.assertEquals; /** * @author Sebastien Deleuze */ -public class JsonObjectEncoderTests { +public class JsonObjectEncoderTests extends AbstractAllocatingTestCase { + + private JsonObjectEncoder encoder; + + @Before + public void createEncoder() { + encoder = new JsonObjectEncoder(allocator); + } @Test public void encodeSingleElementFlux() throws InterruptedException { - JsonObjectEncoder encoder = new JsonObjectEncoder(); - Flux source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer()); + Flux source = + Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")); Iterable results = Flux.from(encoder.encode(source, null, null)).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }).toIterable(); String result = String.join("", results); assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"}]", result); } + @Test public void encodeSingleElementMono() throws InterruptedException { - JsonObjectEncoder encoder = new JsonObjectEncoder(); - Mono source = Mono.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer()); + Mono source = + Mono.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}")); Iterable results = Flux.from(encoder.encode(source, null, null)).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }).toIterable(); String result = String.join("", results); @@ -61,13 +68,12 @@ public class JsonObjectEncoderTests { @Test public void encodeTwoElementsFlux() throws InterruptedException { - JsonObjectEncoder encoder = new JsonObjectEncoder(); - Flux source = Flux.just( - Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(), - Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer()); + Flux source = + Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"), + stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}")); Iterable results = Flux.from(encoder.encode(source, null, null)).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }).toIterable(); String result = String.join("", results); @@ -76,15 +82,15 @@ public class JsonObjectEncoderTests { @Test public void encodeThreeElementsFlux() throws InterruptedException { - JsonObjectEncoder encoder = new JsonObjectEncoder(); - Flux source = Flux.just( - Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(), - Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer(), - Buffer.wrap("{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}").byteBuffer() + Flux source = + Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"), + stringBuffer("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}"), + stringBuffer( + "{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}") ); Iterable results = Flux.from(encoder.encode(source, null, null)).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }).toIterable(); String result = String.join("", results); diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/Pojo.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Pojo.java similarity index 71% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/Pojo.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/Pojo.java index ee55b9ec50..bcf0b24265 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/Pojo.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/Pojo.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.reactive.codec; +package org.springframework.core.codec.support; import javax.xml.bind.annotation.XmlRootElement; @@ -52,4 +52,20 @@ public class Pojo { this.bar = bar; } + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o instanceof Pojo) { + Pojo other = (Pojo) o; + return this.foo.equals(other.foo) && this.bar.equals(other.bar); + } + return false; + } + + @Override + public int hashCode() { + return 31 * foo.hashCode() + bar.hashCode(); + } } diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java similarity index 80% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java index e3538a757b..7b45773de8 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringDecoderTests.java @@ -14,12 +14,12 @@ * limitations under the License. */ -package org.springframework.reactive.codec.decoder; +package org.springframework.core.codec.support; -import java.nio.ByteBuffer; import java.util.List; import java.util.stream.StreamSupport; +import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -28,7 +28,7 @@ import reactor.io.buffer.Buffer; import rx.Single; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.support.StringDecoder; +import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.MediaType; import static java.util.stream.Collectors.toList; @@ -38,9 +38,15 @@ import static org.junit.Assert.*; * @author Sebastien Deleuze * @author Brian Clozel */ -public class StringDecoderTests { +public class StringDecoderTests extends AbstractAllocatingTestCase { + + private StringDecoder decoder; + + @Before + public void createEncoder() { + decoder = new StringDecoder(allocator); + } - private final StringDecoder decoder = new StringDecoder(); @Test public void canDecode() { @@ -51,7 +57,7 @@ public class StringDecoderTests { @Test public void decode() throws InterruptedException { - Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + Flux source = Flux.just(stringBuffer("foo"), stringBuffer("bar")); Flux output = this.decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(1, results.size()); @@ -60,8 +66,8 @@ public class StringDecoderTests { @Test public void decodeDoNotBuffer() throws InterruptedException { - StringDecoder decoder = new StringDecoder(false); - Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + StringDecoder decoder = new StringDecoder(allocator, false); + Flux source = Flux.just(stringBuffer("foo"), stringBuffer("bar")); Flux output = decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList()); assertEquals(2, results.size()); @@ -71,7 +77,7 @@ public class StringDecoderTests { @Test public void decodeMono() throws InterruptedException { - Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + Flux source = Flux.just(stringBuffer("foo"), stringBuffer("bar")); Mono mono = Mono.from(this.decoder.decode(source, ResolvableType.forClassWithGenerics(Mono.class, String.class), MediaType.TEXT_PLAIN)); @@ -81,7 +87,7 @@ public class StringDecoderTests { @Test public void decodeSingle() throws InterruptedException { - Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + 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)); diff --git a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java similarity index 77% rename from spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java rename to spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java index 18a1c8993b..e1a509743e 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/core/codec/support/StringEncoderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * 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. @@ -14,17 +14,19 @@ * limitations under the License. */ -package org.springframework.reactive.codec.encoder; +package org.springframework.core.codec.support; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.stream.StreamSupport; +import org.junit.Before; import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; import org.springframework.core.ResolvableType; -import org.springframework.core.codec.support.StringEncoder; import org.springframework.http.MediaType; import static java.util.stream.Collectors.toList; @@ -33,9 +35,15 @@ import static org.junit.Assert.*; /** * @author Sebastien Deleuze */ -public class StringEncoderTests { +@RunWith(Parameterized.class) +public class StringEncoderTests extends AbstractAllocatingTestCase { - private final StringEncoder encoder = new StringEncoder(); + private StringEncoder encoder; + + @Before + public void createEncoder() { + encoder = new StringEncoder(allocator); + } @Test public void canWrite() { @@ -47,8 +55,8 @@ public class StringEncoderTests { @Test public void write() throws InterruptedException { Flux output = Flux.from(encoder.encode(Flux.just("foo"), null, null)).map(chunk -> { - byte[] b = new byte[chunk.remaining()]; - chunk.get(b); + byte[] b = new byte[chunk.readableByteCount()]; + chunk.read(b); return new String(b, StandardCharsets.UTF_8); }); List results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());