Use Mono semantics for JSON object/array serialization

Before this commit, a handler method returning a stream with a JSON
content-type was producing a JSON object for single element streams
or a JSON array for multiple elements streams.

This kind of dynamic change of the output based on the number of
elements was difficult to handle on client side and not consistent
with Spring MVC behavior.

With this commit, we achieve a more consistent behavior by using
the Mono semantics to control this behavior. Mono (and Promise/Single)
are serialized to JSON object and Flux (and Observable/Stream) are
serialized to JSON array.

spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonEncoder.java

@@ -23,6 +23,7 @@ import java.nio.charset.StandardCharsets;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.reactivestreams.Publisher;
 import reactor.Flux;
+import reactor.Mono;
 import reactor.io.buffer.Buffer;
 
 import org.springframework.core.ResolvableType;
@@ -64,22 +65,23 @@ public class JacksonJsonEncoder extends AbstractEncoder<Object> {
 	public Flux<ByteBuffer> encode(Publisher<? extends Object> inputStream,
 			ResolvableType type, MimeType mimeType, Object... hints) {
 
-		Flux<ByteBuffer> stream = Flux.from(inputStream).map(value -> {
-			Buffer buffer = new Buffer();
-			BufferOutputStream outputStream = new BufferOutputStream(buffer);
-			try {
-				this.mapper.writeValue(outputStream, value);
-			}
-			catch (IOException e) {
-				throw new CodecException("Error while writing the data", e);
-			}
-			buffer.flip();
-			return buffer.byteBuffer();
-		});
-		if (this.postProcessor != null) {
-			stream = this.postProcessor.encode(stream, type, mimeType, hints);
-		};
-		return stream;
+		Publisher<ByteBuffer> 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);
+		try {
+			this.mapper.writeValue(outputStream, value);
+		}
+		catch (IOException e) {
+			throw new CodecException("Error while writing the data", e);
+		}
+		buffer.flip();
+		return buffer.byteBuffer();
 	}
 
 }

spring-web-reactive/src/main/java/org/springframework/core/codec/support/JsonObjectEncoder.java

@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import reactor.Flux;
+import reactor.Mono;
 import reactor.core.subscriber.SubscriberBarrier;
 import reactor.core.support.BackpressureUtils;
 import reactor.io.buffer.Buffer;
@@ -32,8 +33,9 @@ import org.springframework.core.ResolvableType;
 import org.springframework.util.MimeType;
 
 /**
- * Encode a byte stream of individual JSON element to a byte stream representing
- * a single JSON array when if it contains more than one element.
+ * Encode a byte stream of individual JSON element to a byte stream representing:
+ *  - the same JSON object than the input stream if it is a {@link Mono}
+ *  - a JSON array for other kinds of {@link Publisher}
  *
  * @author Sebastien Deleuze
  * @author Stephane Maldini
@@ -48,22 +50,24 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
 	}
 
 	@Override
-	public Flux<ByteBuffer> encode(Publisher<? extends ByteBuffer> messageStream,
+	public Flux<ByteBuffer> encode(Publisher<? extends ByteBuffer> inputStream,
 			ResolvableType type, MimeType mimeType, Object... hints) {
 
-		//noinspection Convert2MethodRef
-		return Flux.from(messageStream).lift(bbs -> new JsonEncoderBarrier(bbs));
+		if (inputStream instanceof Mono) {
+			return Flux.from(inputStream);
+		}
+		return Flux.from(inputStream).lift(s -> new JsonArrayEncoderBarrier(s));
 	}
 
 
-	private static class JsonEncoderBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
+	private static class JsonArrayEncoderBarrier extends SubscriberBarrier<ByteBuffer, ByteBuffer> {
 
 		@SuppressWarnings("rawtypes")
-		static final AtomicLongFieldUpdater<JsonEncoderBarrier> REQUESTED =
-				AtomicLongFieldUpdater.newUpdater(JsonEncoderBarrier.class, "requested");
+		static final AtomicLongFieldUpdater<JsonArrayEncoderBarrier> REQUESTED =
+				AtomicLongFieldUpdater.newUpdater(JsonArrayEncoderBarrier.class, "requested");
 
-		static final AtomicIntegerFieldUpdater<JsonEncoderBarrier> TERMINATED =
-				AtomicIntegerFieldUpdater.newUpdater(JsonEncoderBarrier.class, "terminated");
+		static final AtomicIntegerFieldUpdater<JsonArrayEncoderBarrier> TERMINATED =
+				AtomicIntegerFieldUpdater.newUpdater(JsonArrayEncoderBarrier.class, "terminated");
 
 
 		private ByteBuffer prev = null;
@@ -75,7 +79,7 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
 		private volatile int terminated;
 
 
-		public JsonEncoderBarrier(Subscriber<? super ByteBuffer> subscriber) {
+		public JsonArrayEncoderBarrier(Subscriber<? super ByteBuffer> subscriber) {
 			super(subscriber);
 		}
 
@@ -94,20 +98,19 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
 		@Override
 		protected void doNext(ByteBuffer next) {
 			this.count++;
-			if (this.count == 1) {
-				this.prev = next;
-				super.doRequest(1);
-				return;
-			}
 
 			ByteBuffer tmp = this.prev;
 			this.prev = next;
 			Buffer buffer = new Buffer();
-			if (this.count == 2) {
+			if (this.count == 1) {
 				buffer.append("[");
 			}
-			buffer.append(tmp);
-			buffer.append(",");
+			if (tmp != null) {
+				buffer.append(tmp);
+			}
+			if (this.count > 1) {
+				buffer.append(",");
+			}
 			buffer.flip();
 
 			BackpressureUtils.getAndSub(REQUESTED, this, 1L);
@@ -118,9 +121,7 @@ public class JsonObjectEncoder extends AbstractEncoder<ByteBuffer> {
 			if(BackpressureUtils.getAndSub(REQUESTED, this, 1L) > 0) {
 				Buffer buffer = new Buffer();
 				buffer.append(this.prev);
-				if (this.count > 1) {
-					buffer.append("]");
-				}
+				buffer.append("]");
 				buffer.flip();
 				subscriber.onNext(buffer.byteBuffer());
 				super.doComplete();

spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JsonObjectEncoderTests.java

@@ -18,13 +18,12 @@ package org.springframework.reactive.codec.encoder;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.List;
 
 import static org.junit.Assert.assertEquals;
 import org.junit.Test;
+import reactor.Flux;
+import reactor.Mono;
 import reactor.io.buffer.Buffer;
-import reactor.rx.Stream;
-import reactor.rx.Streams;
 
 import org.springframework.core.codec.support.JsonObjectEncoder;
 
@@ -34,46 +33,59 @@ import org.springframework.core.codec.support.JsonObjectEncoder;
 public class JsonObjectEncoderTests {
 
 	@Test
-	public void encodeSingleElement() throws InterruptedException {
+	public void encodeSingleElementFlux() throws InterruptedException {
 		JsonObjectEncoder encoder = new JsonObjectEncoder();
-		Stream<ByteBuffer> source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
-		List<String> results = Streams.from(encoder.encode(source, null, null)).map(chunk -> {
+		Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
+		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.remaining()];
 			chunk.get(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toList().get();
+		}).toIterable();
+		String result = String.join("", results);
+		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"}]", result);
+	}
+
+	@Test
+	public void encodeSingleElementMono() throws InterruptedException {
+		JsonObjectEncoder encoder = new JsonObjectEncoder();
+		Mono<ByteBuffer> source = Mono.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
+		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
+			byte[] b = new byte[chunk.remaining()];
+			chunk.get(b);
+			return new String(b, StandardCharsets.UTF_8);
+		}).toIterable();
 		String result = String.join("", results);
 		assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", result);
 	}
 
 	@Test
-	public void encodeTwoElements() throws InterruptedException {
+	public void encodeTwoElementsFlux() throws InterruptedException {
 		JsonObjectEncoder encoder = new JsonObjectEncoder();
-		Stream<ByteBuffer> source = Streams.just(
+		Flux<ByteBuffer> source = Flux.just(
 				Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(),
 				Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer());
-		List<String> results = Streams.from(encoder.encode(source, null, null)).map(chunk -> {
+		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.remaining()];
 			chunk.get(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toList().get();
+		}).toIterable();
 		String result = String.join("", results);
 		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]", result);
 	}
 
 	@Test
-	public void encodeThreeElements() throws InterruptedException {
+	public void encodeThreeElementsFlux() throws InterruptedException {
 		JsonObjectEncoder encoder = new JsonObjectEncoder();
-		Stream<ByteBuffer> source = Streams.just(
+		Flux<ByteBuffer> source = Flux.just(
 				Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer(),
 				Buffer.wrap("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}").byteBuffer(),
 				Buffer.wrap("{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}").byteBuffer()
 		);
-		List<String> results = Streams.from(encoder.encode(source, null, null)).map(chunk -> {
+		Iterable<String> results = Flux.from(encoder.encode(source, null, null)).map(chunk -> {
 			byte[] b = new byte[chunk.remaining()];
 			chunk.get(b);
 			return new String(b, StandardCharsets.UTF_8);
-		}).toList().get();
+		}).toIterable();
 		String result = String.join("", results);
 		assertEquals("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"},{\"foo\": \"foofoofoofoo\", \"bar\": \"barbarbarbar\"}]", result);
 	}