Polish and simplify JacksonJsonEncoder

spring-web-reactive/src/main/java/org/springframework/http/codec/json/JacksonJsonEncoder.java

@@ -46,14 +46,16 @@ import org.springframework.util.MimeType;
  */
 public class JacksonJsonEncoder extends AbstractEncoder<Object> {
 
-	private final ObjectMapper mapper;
-
 	private static final ByteBuffer START_ARRAY_BUFFER = ByteBuffer.wrap(new byte[]{'['});
 
 	private static final ByteBuffer SEPARATOR_BUFFER = ByteBuffer.wrap(new byte[]{','});
 
 	private static final ByteBuffer END_ARRAY_BUFFER = ByteBuffer.wrap(new byte[]{']'});
 
+
+	private final ObjectMapper mapper;
+
+
 	public JacksonJsonEncoder() {
 		this(new ObjectMapper());
 	}
@@ -66,54 +68,44 @@ public class JacksonJsonEncoder extends AbstractEncoder<Object> {
 		this.mapper = mapper;
 	}
 
+
 	@Override
-	public Flux<DataBuffer> encode(Publisher<?> inputStream,
-			DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType,
-			Object... hints) {
+	public Flux<DataBuffer> encode(Publisher<?> inputStream, DataBufferFactory bufferFactory,
+			ResolvableType elementType, MimeType mimeType, Object... hints) {
+
 		Assert.notNull(inputStream, "'inputStream' must not be null");
 		Assert.notNull(bufferFactory, "'bufferFactory' must not be null");
 		Assert.notNull(elementType, "'elementType' must not be null");
+
 		if (inputStream instanceof Mono) {
-			// single object
-			return Flux.from(inputStream)
-					.map(value -> serialize(value, bufferFactory, elementType));
+			return Flux.from(inputStream).map(value -> encodeValue(value, bufferFactory, elementType));
 		}
-		else {
-			// array
-			Mono<DataBuffer> startArray =
-					Mono.just(bufferFactory.wrap(START_ARRAY_BUFFER));
-			Flux<DataBuffer> arraySeparators =
-					Mono.fromSupplier(() -> bufferFactory.wrap(SEPARATOR_BUFFER)).repeat();
-			Mono<DataBuffer> endArray =
-					Mono.just(bufferFactory.wrap(END_ARRAY_BUFFER));
-
-			Flux<DataBuffer> serializedObjects = Flux.from(inputStream)
-					.map(value -> serialize(value, bufferFactory, elementType));
 
-			Flux<DataBuffer> array = Flux.zip(serializedObjects, arraySeparators)
-					.flatMap(tuple -> Flux.just(tuple.getT1(), tuple.getT2()));
+		Flux<DataBuffer> array = Flux.from(inputStream)
+				.flatMap(value ->
+						Flux.just(encodeValue(value, bufferFactory, elementType),
+								bufferFactory.wrap(SEPARATOR_BUFFER)));
 
-			Flux<DataBuffer> arrayWithoutLastSeparator = array.skipLast(1);
-
-			return Flux.concat(startArray, arrayWithoutLastSeparator, endArray);
-		}
+		return Flux.concat(Mono.just(bufferFactory.wrap(START_ARRAY_BUFFER)), array.skipLast(1),
+				Mono.just(bufferFactory.wrap(END_ARRAY_BUFFER)));
 	}
 
-	private DataBuffer serialize(Object value, DataBufferFactory dataBufferFactory,
-			ResolvableType type) {
+	private DataBuffer encodeValue(Object value, DataBufferFactory bufferFactory, ResolvableType type) {
 		TypeFactory typeFactory = this.mapper.getTypeFactory();
 		JavaType javaType = typeFactory.constructType(type.getType());
 		ObjectWriter writer = this.mapper.writerFor(javaType);
-		DataBuffer buffer = dataBufferFactory.allocateBuffer();
+
+		DataBuffer buffer = bufferFactory.allocateBuffer();
 		OutputStream outputStream = buffer.asOutputStream();
+
 		try {
 			writer.writeValue(outputStream, value);
 		}
 		catch (IOException e) {
 			throw new CodecException("Error while writing the data", e);
 		}
+
 		return buffer;
 	}
 
-
 }

spring-web-reactive/src/test/java/org/springframework/core/codec/ByteBufferEncoderTests.java

@@ -61,7 +61,7 @@ public class ByteBufferEncoderTests extends AbstractDataBufferAllocatingTestCase
 		Flux<ByteBuffer> source =
 				Flux.just(ByteBuffer.wrap(fooBytes), ByteBuffer.wrap(barBytes));
 
-		Flux<DataBuffer> output = this.encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
 				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
 				null);
 		TestSubscriber

spring-web-reactive/src/test/java/org/springframework/core/codec/ResourceEncoderTests.java

@@ -62,7 +62,7 @@ public class ResourceEncoderTests extends AbstractDataBufferAllocatingTestCase {
 
 		Mono<Resource> source = Mono.just(resource);
 
-		Flux<DataBuffer> output = this.encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
 				ResolvableType.forClass(Resource.class),
 						null);
 

spring-web-reactive/src/test/java/org/springframework/core/codec/StringEncoderTests.java

@@ -59,7 +59,7 @@ public class StringEncoderTests extends AbstractDataBufferAllocatingTestCase {
 	@Test
 	public void write() throws InterruptedException {
 		Flux<String> output = Flux.from(
-				this.encoder.encode(Flux.just("foo"), this.dataBufferFactory, null, null))
+				this.encoder.encode(Flux.just("foo"), this.bufferFactory, null, null))
 						.map(chunk -> {
 							byte[] b = new byte[chunk.readableByteCount()];
 							chunk.read(b);

spring-web-reactive/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java

@@ -37,7 +37,7 @@ import static org.junit.Assert.assertEquals;
 public abstract class AbstractDataBufferAllocatingTestCase {
 
 	@Parameterized.Parameter
-	public DataBufferFactory dataBufferFactory;
+	public DataBufferFactory bufferFactory;
 
 	@Parameterized.Parameters(name = "{0}")
 	public static Object[][] dataBufferFactories() {
@@ -53,12 +53,12 @@ public abstract class AbstractDataBufferAllocatingTestCase {
 	}
 
 	protected DataBuffer createDataBuffer(int capacity) {
-		return this.dataBufferFactory.allocateBuffer(capacity);
+		return this.bufferFactory.allocateBuffer(capacity);
 	}
 
 	protected DataBuffer stringBuffer(String value) {
 		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
-		DataBuffer buffer = this.dataBufferFactory.allocateBuffer(bytes.length);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
 		buffer.write(bytes);
 		return buffer;
 	}

spring-web-reactive/src/test/java/org/springframework/core/io/buffer/support/DataBufferTestUtilsTests.java

@@ -33,7 +33,7 @@ public class DataBufferTestUtilsTests extends AbstractDataBufferAllocatingTestCa
 
 	@Test
 	public void dumpBytes() {
-		DataBuffer buffer = this.dataBufferFactory.allocateBuffer(4);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(4);
 		byte[] source = {'a', 'b', 'c', 'd'};
 		buffer.write(source);
 
@@ -46,7 +46,7 @@ public class DataBufferTestUtilsTests extends AbstractDataBufferAllocatingTestCa
 
 	@Test
 	public void dumpString() {
-		DataBuffer buffer = this.dataBufferFactory.allocateBuffer(4);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(4);
 		String source = "abcd";
 		buffer.write(source.getBytes(StandardCharsets.UTF_8));
 

spring-web-reactive/src/test/java/org/springframework/core/io/buffer/support/DataBufferUtilsTests.java

@@ -42,7 +42,7 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
 				.toURI();
 		FileChannel channel = FileChannel.open(Paths.get(uri), StandardOpenOption.READ);
 
-		Flux<DataBuffer> flux = DataBufferUtils.read(channel, this.dataBufferFactory, 4);
+		Flux<DataBuffer> flux = DataBufferUtils.read(channel, this.bufferFactory, 4);
 
 		TestSubscriber
 				.subscribe(flux)
@@ -61,7 +61,7 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
 				.toURI();
 		FileChannel channel = FileChannel.open(Paths.get(uri), StandardOpenOption.READ);
 
-		Flux<DataBuffer> flux = DataBufferUtils.read(channel, this.dataBufferFactory, 3);
+		Flux<DataBuffer> flux = DataBufferUtils.read(channel, this.bufferFactory, 3);
 
 		TestSubscriber
 				.subscribe(flux)
@@ -80,7 +80,7 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
 		InputStream is = DataBufferUtilsTests.class
 				.getResourceAsStream("DataBufferUtilsTests.txt");
 
-		Flux<DataBuffer> flux = DataBufferUtils.read(is, this.dataBufferFactory, 4);
+		Flux<DataBuffer> flux = DataBufferUtils.read(is, this.bufferFactory, 4);
 
 		TestSubscriber
 				.subscribe(flux)

spring-web-reactive/src/test/java/org/springframework/http/codec/SseEventEncoderTests.java

@@ -64,7 +64,7 @@ public class SseEventEncoderTests extends AbstractDataBufferAllocatingTestCase {
 		event.setComment("bla\nbla bla\nbla bla bla");
 		event.setReconnectTime(123L);
 		Mono<SseEvent> source = Mono.just(event);
-		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = encoder.encode(source, this.bufferFactory,
 						ResolvableType.forClass(SseEvent.class), new MimeType("text", "event-stream"));
 		TestSubscriber
 				.subscribe(output)
@@ -84,7 +84,7 @@ public class SseEventEncoderTests extends AbstractDataBufferAllocatingTestCase {
 	public void encodeString() {
 		SseEventEncoder encoder = new SseEventEncoder(Arrays.asList(new JacksonJsonEncoder()));
 		Flux<String> source = Flux.just("foo", "bar");
-		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = encoder.encode(source, this.bufferFactory,
 				ResolvableType.forClass(String.class), new MimeType("text", "event-stream"));
 		TestSubscriber
 				.subscribe(output)
@@ -103,7 +103,7 @@ public class SseEventEncoderTests extends AbstractDataBufferAllocatingTestCase {
 	public void encodeMultilineString() {
 		SseEventEncoder encoder = new SseEventEncoder(Arrays.asList(new JacksonJsonEncoder()));
 		Flux<String> source = Flux.just("foo\nbar", "foo\nbaz");
-		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = encoder.encode(source, this.bufferFactory,
 				ResolvableType.forClass(String.class), new MimeType("text", "event-stream"));
 		TestSubscriber
 				.subscribe(output)
@@ -122,7 +122,7 @@ public class SseEventEncoderTests extends AbstractDataBufferAllocatingTestCase {
 	public void encodePojo() {
 		SseEventEncoder encoder = new SseEventEncoder(Arrays.asList(new JacksonJsonEncoder()));
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
-		Flux<DataBuffer> output = encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = encoder.encode(source, this.bufferFactory,
 						ResolvableType.forClass(Pojo.class), new MimeType("text", "event-stream"));
 		TestSubscriber
 				.subscribe(output)

spring-web-reactive/src/test/java/org/springframework/http/codec/json/JacksonJsonEncoderTests.java

@@ -39,11 +39,13 @@ public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCas
 
 	private JacksonJsonEncoder encoder;
 
+
 	@Before
 	public void createEncoder() {
 		this.encoder = new JacksonJsonEncoder();
 	}
 
+
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(null, MediaType.APPLICATION_JSON));
@@ -57,13 +59,10 @@ public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCas
 				new Pojo("foofoo", "barbar"),
 				new Pojo("foofoofoo", "barbarbar")
 		);
-
 		ResolvableType type = ResolvableType.forClass(Pojo.class);
-		Flux<DataBuffer> output =
-				this.encoder.encode(source, this.dataBufferFactory, type, null);
+		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null);
 
-		TestSubscriber
-				.subscribe(output)
+		TestSubscriber.subscribe(output)
 				.assertComplete()
 				.assertNoError()
 				.assertValuesWith(
@@ -80,13 +79,10 @@ public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCas
 	@Test
 	public void encodeWithType() {
 		Flux<ParentClass> source = Flux.just(new Foo(), new Bar());
-
 		ResolvableType type = ResolvableType.forClass(ParentClass.class);
-		Flux<DataBuffer> output =
-				this.encoder.encode(source, this.dataBufferFactory, type, null);
+		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null);
 
-		TestSubscriber
-				.subscribe(output)
+		TestSubscriber.subscribe(output)
 				.assertComplete()
 				.assertNoError()
 				.assertValuesWith(stringConsumer("["),
@@ -96,6 +92,7 @@ public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCas
 						stringConsumer("]"));
 	}
 
+	
 	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
 	private static class ParentClass {
 	}

spring-web-reactive/src/test/java/org/springframework/http/codec/xml/Jaxb2EncoderTests.java

@@ -71,7 +71,7 @@ public class Jaxb2EncoderTests extends AbstractDataBufferAllocatingTestCase {
 	@Test
 	public void encode() {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
-		Flux<DataBuffer> output = this.encoder.encode(source, this.dataBufferFactory,
+		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
 				ResolvableType.forClass(Pojo.class),
 						MediaType.APPLICATION_XML);
 		TestSubscriber