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.

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<T> extends AbstractDecoder<T> {
+
+	public AbstractRawByteStreamDecoder(MimeType... supportedMimeTypes) {
+		super(supportedMimeTypes);
+	}
+
+	@Override
+	public Flux<T> decode(Publisher<ByteBuffer> 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
+	 *
+	 * <p>Implementations should provide their own {@link SubscriberBarrier} or use one of the
+	 * provided implementations by this class
+	 */
+	public abstract SubscriberBarrier<ByteBuffer, ByteBuffer> subscriberBarrier(Subscriber<? super ByteBuffer> subscriber);
+
+	public abstract Flux<T> decodeInternal(Publisher<ByteBuffer> 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<ByteBuffer, ByteBuffer> {
+
+		@SuppressWarnings("rawtypes")
+		static final AtomicLongFieldUpdater<ReduceSingleByteStreamBarrier> REQUESTED =
+				AtomicLongFieldUpdater.newUpdater(ReduceSingleByteStreamBarrier.class, "requested");
+
+		static final AtomicIntegerFieldUpdater<ReduceSingleByteStreamBarrier> TERMINATED =
+				AtomicIntegerFieldUpdater.newUpdater(ReduceSingleByteStreamBarrier.class, "terminated");
+
+
+		private volatile long requested;
+
+		private volatile int terminated;
+
+		private ByteBuffer buffer;
+
+		public ReduceSingleByteStreamBarrier(Subscriber<? super ByteBuffer> 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<ByteBuffer, ByteBuffer> {
+
+		@SuppressWarnings("rawtypes")
+		static final AtomicLongFieldUpdater<SplitLinesByteStreamBarrier> REQUESTED =
+				AtomicLongFieldUpdater.newUpdater(SplitLinesByteStreamBarrier.class, "requested");
+
+		static final AtomicIntegerFieldUpdater<SplitLinesByteStreamBarrier> TERMINATED =
+				AtomicIntegerFieldUpdater.newUpdater(SplitLinesByteStreamBarrier.class, "terminated");
+
+
+		private volatile long requested;
+
+		private volatile int terminated;
+
+		private ByteBuffer buffer;
+
+		public SplitLinesByteStreamBarrier(Subscriber<? super ByteBuffer> 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();
+			}
+		}
+	}
+
+}

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.
  *
+ * <p>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<String> {
+public class StringDecoder extends AbstractRawByteStreamDecoder<String> {
 
 	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
+	 *
+	 * <p>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<String> {
 	}
 
 	@Override
-	public Flux<String> decode(Publisher<ByteBuffer> inputStream, ResolvableType type,
-			MimeType mimeType, Object... hints) {
+	public SubscriberBarrier<ByteBuffer, ByteBuffer> subscriberBarrier(Subscriber<? super ByteBuffer> subscriber) {
+		if (reduceToSingleBuffer) {
+			return new ReduceSingleByteStreamBarrier(subscriber);
+		}
+		else {
+			return new SubscriberBarrier(subscriber);
+		}
 
+	}
+
+	@Override
+	public Flux<String> decodeInternal(Publisher<ByteBuffer> 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));
 	}
 
 }

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<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
-		Flux<String> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, String.class), null);
+		Flux<String> output = this.decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null);
+		List<String> 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<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
+		Flux<String> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Flux.class, String.class), null);
 		List<String> 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<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
+		Mono<String> 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<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
+		Single<String> single = RxJava1SingleConverter.from(this.decoder.decode(source,
+				ResolvableType.forClassWithGenerics(Single.class, String.class),
+				MediaType.TEXT_PLAIN));
+		String result = single.toBlocking().value();
+		assertEquals("foobar", result);
+	}
+
 }