From cf6ceeed621531288579051495312f069ddc133d Mon Sep 17 00:00:00 2001 From: Brian Clozel Date: Tue, 15 Dec 2015 14:14:00 +0100 Subject: [PATCH] Allow Splitting/aggregation operations in Decoders When decoding buffers as plain strings, the StringDecoder returns a Publisher that may produce one or more `onNext` events. This is perfectly valid, but leads to errors when trying to convert the resulting Publisher into a `reactor.Mono` or `rx.Single`. If the original Publisher emits 2 or more `onNext` signals, converting to: * `rx.Single` will throw an error saying that the underlying Observable "emitted too many elements" * `reactor.Mono` may contain only the first emitted element This commit adds a `AbstractRawByteStreamDecoder` that takes a `SubscriberBarrier` to apply splitting/aggregation operations on the received elements. The `StringDecoder` class now inherits from this abstract class and uses one of the provided `SubscriberBarrier` implementations to buffer all received elements in a single buffer. --- .../support/AbstractRawByteStreamDecoder.java | 226 ++++++++++++++++++ .../core/codec/support/StringDecoder.java | 50 +++- .../codec/decoder/StringDecoderTests.java | 46 +++- 3 files changed, 308 insertions(+), 14 deletions(-) create mode 100644 spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractRawByteStreamDecoder.java 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 new file mode 100644 index 0000000000..fa20ce0c19 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/AbstractRawByteStreamDecoder.java @@ -0,0 +1,226 @@ +/* + * 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.ByteBuffer; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; + +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import reactor.Flux; +import reactor.core.subscriber.SubscriberBarrier; +import reactor.core.support.BackpressureUtils; + +import org.springframework.core.ResolvableType; +import org.springframework.core.codec.Decoder; +import org.springframework.util.MimeType; + +/** + * Abstract {@link Decoder} that plugs a {@link SubscriberBarrier} into the {@code Flux} + * pipeline in order to apply splitting/aggregation operations on the stream of data. + * + * @author Brian Clozel + */ +public abstract class AbstractRawByteStreamDecoder extends AbstractDecoder { + + public AbstractRawByteStreamDecoder(MimeType... supportedMimeTypes) { + super(supportedMimeTypes); + } + + @Override + public Flux decode(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { + + return decodeInternal(Flux.from(inputStream).lift(bbs -> subscriberBarrier(bbs)), + type, mimeType, hints); + } + + /** + * Create a {@link SubscriberBarrier} instance that will be plugged into the Publisher pipeline + * + *

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 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 + */ + public static class ReduceSingleByteStreamBarrier extends SubscriberBarrier { + + @SuppressWarnings("rawtypes") + static final AtomicLongFieldUpdater REQUESTED = + AtomicLongFieldUpdater.newUpdater(ReduceSingleByteStreamBarrier.class, "requested"); + + static final AtomicIntegerFieldUpdater TERMINATED = + AtomicIntegerFieldUpdater.newUpdater(ReduceSingleByteStreamBarrier.class, "terminated"); + + + private volatile long requested; + + private volatile int terminated; + + private ByteBuffer buffer; + + public ReduceSingleByteStreamBarrier(Subscriber subscriber) { + super(subscriber); + this.buffer = ByteBuffer.allocate(0); + } + + @Override + protected void doRequest(long n) { + BackpressureUtils.getAndAdd(REQUESTED, this, n); + if (TERMINATED.compareAndSet(this, 1, 2)) { + drainLast(); + } + else { + super.doRequest(Long.MAX_VALUE); + } + } + + @Override + protected void doComplete() { + if (TERMINATED.compareAndSet(this, 0, 1)) { + drainLast(); + } + } + + /* + * 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 drainLast() { + if (BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) { + this.buffer.flip(); + subscriber.onNext(this.buffer); + super.doComplete(); + } + } + } + + /** + * {@code SubscriberBarrier} implementation that splits incoming elements + * using line return delimiters: {@code "\n"} and {@code "\r\n"} + */ + public static class SplitLinesByteStreamBarrier extends SubscriberBarrier { + + @SuppressWarnings("rawtypes") + static final AtomicLongFieldUpdater REQUESTED = + AtomicLongFieldUpdater.newUpdater(SplitLinesByteStreamBarrier.class, "requested"); + + static final AtomicIntegerFieldUpdater TERMINATED = + AtomicIntegerFieldUpdater.newUpdater(SplitLinesByteStreamBarrier.class, "terminated"); + + + private volatile long requested; + + private volatile int terminated; + + private ByteBuffer buffer; + + public SplitLinesByteStreamBarrier(Subscriber subscriber) { + super(subscriber); + this.buffer = ByteBuffer.allocate(0); + } + + @Override + protected void doRequest(long n) { + BackpressureUtils.getAndAdd(REQUESTED, this, n); + if (TERMINATED.compareAndSet(this, 1, 2)) { + drainLast(); + } + else { + super.doRequest(n); + } + } + + @Override + protected void doComplete() { + if (TERMINATED.compareAndSet(this, 0, 1)) { + drainLast(); + } + } + + /* + * 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); + + 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); + consumeSeparator(this.buffer); + this.buffer = this.buffer.slice(); + super.doNext(ByteBuffer.wrap(message)); + } + } + else { + super.doRequest(1); + } + } + } + + protected int findEndOfLine(ByteBuffer buffer) { + + final int n = buffer.limit(); + for (int i = 0; i < n; i++) { + final byte b = buffer.get(i); + if (b == '\n') { + return i; + } + else if (b == '\r' && i < n - 1 && buffer.get(i + 1) == '\n') { + return i; + } + } + + return -1; + } + + protected void consumeSeparator(ByteBuffer buffer) { + byte sep = buffer.get(); + if (sep == '\r') { + buffer.get(); + } + } + + 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/StringDecoder.java b/spring-web-reactive/src/main/java/org/springframework/core/codec/support/StringDecoder.java index 5205a1e0d0..332af997ea 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 @@ -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. @@ -21,8 +21,9 @@ import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; import reactor.Flux; -import reactor.io.buffer.Buffer; +import reactor.core.subscriber.SubscriberBarrier; import org.springframework.core.ResolvableType; import org.springframework.util.MimeType; @@ -30,15 +31,41 @@ import org.springframework.util.MimeType; /** * Decode from a bytes stream to a String stream. * + *

By default, this decoder will buffer the received elements into a single + * {@code ByteBuffer} and will emit a single {@code String} once the stream of + * elements is complete. This behavior can be turned off using an constructor + * argument but the {@code Subcriber} should pay attention to split characters + * issues. + * * @author Sebastien Deleuze + * @author Brian Clozel * @see StringEncoder */ -public class StringDecoder extends AbstractDecoder { +public class StringDecoder extends AbstractRawByteStreamDecoder { public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8; + public final boolean reduceToSingleBuffer; + + /** + * 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); + } + + /** + * Create a {@code StringDecoder} that decodes a bytes stream to a String stream + * + * @param reduceToSingleBuffer whether this decoder should buffer all received items + * and decode a single consolidated String or re-emit items as they are provided + */ + public StringDecoder(boolean reduceToSingleBuffer) { super(new MimeType("text", "plain", DEFAULT_CHARSET)); + this.reduceToSingleBuffer = reduceToSingleBuffer; } @Override @@ -48,17 +75,26 @@ public class StringDecoder extends AbstractDecoder { } @Override - public Flux decode(Publisher inputStream, ResolvableType type, - MimeType mimeType, Object... hints) { + public SubscriberBarrier subscriberBarrier(Subscriber subscriber) { + if (reduceToSingleBuffer) { + return new ReduceSingleByteStreamBarrier(subscriber); + } + else { + return new SubscriberBarrier(subscriber); + } + } + + @Override + public Flux decodeInternal(Publisher inputStream, ResolvableType type, MimeType mimeType, Object... hints) { Charset charset; if (mimeType != null && mimeType.getCharSet() != null) { charset = mimeType.getCharSet(); } else { - charset = DEFAULT_CHARSET; + charset = DEFAULT_CHARSET; } - return Flux.from(inputStream).map(content -> new String(new Buffer(content).asBytes(), charset)); + return Flux.from(inputStream).map(content -> new String(content.duplicate().array(), charset)); } } 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/reactive/codec/decoder/StringDecoderTests.java index 038d26e2b1..0d693ec4af 100644 --- a/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.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,18 +16,19 @@ package org.springframework.reactive.codec.decoder; +import static java.util.stream.Collectors.*; +import static org.junit.Assert.*; + import java.nio.ByteBuffer; import java.util.List; import java.util.stream.StreamSupport; -import static java.util.stream.Collectors.toList; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; import org.junit.Test; -import org.reactivestreams.Publisher; import reactor.Flux; +import reactor.Mono; +import reactor.core.publisher.convert.RxJava1SingleConverter; import reactor.io.buffer.Buffer; +import rx.Single; import org.springframework.core.ResolvableType; import org.springframework.core.codec.support.StringDecoder; @@ -35,6 +36,7 @@ import org.springframework.http.MediaType; /** * @author Sebastien Deleuze + * @author Brian Clozel */ public class StringDecoderTests { @@ -50,11 +52,41 @@ public class StringDecoderTests { @Test public void decode() throws InterruptedException { Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); - Flux output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, String.class), null); + 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()); + assertEquals("foobar", results.get(0)); + } + + @Test + public void decodeDoNotBuffer() throws InterruptedException { + StringDecoder decoder = new StringDecoder(false); + Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + 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()); assertEquals("foo", results.get(0)); assertEquals("bar", results.get(1)); } + @Test + public void decodeMono() throws InterruptedException { + Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + Mono mono = Mono.from(this.decoder.decode(source, + ResolvableType.forClassWithGenerics(Mono.class, String.class), + MediaType.TEXT_PLAIN)); + String result = mono.get(); + assertEquals("foobar", result); + } + + @Test + public void decodeSingle() throws InterruptedException { + Flux source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer()); + Single single = RxJava1SingleConverter.from(this.decoder.decode(source, + ResolvableType.forClassWithGenerics(Single.class, String.class), + MediaType.TEXT_PLAIN)); + String result = single.toBlocking().value(); + assertEquals("foobar", result); + } + }