Add and use AbstractDecoderTestCase

Introduce new base test case for decoder tests, and use it.

Issue: SPR-17449

spring-core/src/test/java/org/springframework/core/codec/AbstractDecoderTestCase.java

@@ -0,0 +1,450 @@
+/*
+ * Copyright 2002-2018 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;
+
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+
+/**
+ * Abstract base class for {@link Decoder} unit tests. Subclasses need to implement
+ * {@link #canDecode()}, {@link #decode()} and {@link #decodeToMono()}, possibly using the wide
+ * variety of helper methods like {@link #testDecodeAll} or {@link #testDecodeToMonoAll}.
+ *
+ * @author Arjen Poutsma
+ * @since 5.1.3
+ */
+@SuppressWarnings("ProtectedField")
+public abstract class AbstractDecoderTestCase<D extends Decoder<?>>
+		extends AbstractLeakCheckingTestCase {
+
+	/**
+	 * The decoder to test.
+	 */
+	protected D decoder;
+
+	/**
+	 * Construct a new {@code AbstractDecoderTestCase} for the given decoder.
+	 * @param decoder the decoder
+	 */
+	protected AbstractDecoderTestCase(D decoder) {
+		Assert.notNull(decoder, "Encoder must not be null");
+
+		this.decoder = decoder;
+	}
+
+
+	/**
+	 * Subclasses should implement this method to test {@link Decoder#canDecode}.
+	 */
+	@Test
+	public abstract void canDecode() throws Exception;
+
+	/**
+	 * Subclasses should implement this method to test {@link Decoder#decode}, possibly using
+	 * {@link #testDecodeAll} or other helper methods.
+	 */
+	@Test
+	public abstract void decode() throws Exception;
+
+	/**
+	 * Subclasses should implement this method to test {@link Decoder#decodeToMono}, possibly using
+	 * {@link #testDecodeToMonoAll}.
+	 */
+	@Test
+	public abstract void decodeToMono() throws Exception;
+
+	// Flux
+
+	/**
+	 * Helper methods that tests for a variety of {@link Flux} decoding scenarios. This methods
+	 * invokes:
+	 * <ul>
+	 *     <li>{@link #testDecode(Publisher, ResolvableType, Consumer, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeError(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeCancel(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeEmpty(ResolvableType, MimeType, Map)}</li>
+	 * </ul>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputClass the desired output class
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param <T> the output type
+	 */
+	protected <T> void testDecodeAll(Publisher<DataBuffer> input, Class<? extends T> outputClass,
+			Consumer<StepVerifier.FirstStep<T>> stepConsumer) {
+		testDecodeAll(input, ResolvableType.forClass(outputClass), stepConsumer, null, null);
+	}
+
+	/**
+	 * Helper methods that tests for a variety of {@link Flux} decoding scenarios. This methods
+	 * invokes:
+	 * <ul>
+	 *     <li>{@link #testDecode(Publisher, ResolvableType, Consumer, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeError(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeCancel(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeEmpty(ResolvableType, MimeType, Map)}</li>
+	 * </ul>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 * @param <T> the output type
+	 */
+	protected <T> void testDecodeAll(Publisher<DataBuffer> input, ResolvableType outputType,
+			Consumer<StepVerifier.FirstStep<T>> stepConsumer,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+		testDecode(input, outputType, stepConsumer, mimeType, hints);
+		testDecodeError(input, outputType, mimeType, hints);
+		testDecodeCancel(input, outputType, mimeType, hints);
+		testDecodeEmpty(outputType, mimeType, hints);
+	}
+
+	/**
+	 * Test a standard {@link Decoder#decode decode} scenario. For example:
+	 * <pre class="code">
+	 * byte[] bytes1 = ...
+	 * byte[] bytes2 = ...
+	 *
+	 * Flux&lt;DataBuffer&gt; input = Flux.concat(
+	 *   dataBuffer(bytes1),
+	 *   dataBuffer(bytes2));
+	 *
+	 * testDecodeAll(input, byte[].class, step -&gt; step
+	 *   .consumeNextWith(expectBytes(bytes1))
+	 *   .consumeNextWith(expectBytes(bytes2))
+	 * 	 .verifyComplete());
+	 * </pre>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputClass the desired output class
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param <T> the output type
+	 */
+	protected <T> void testDecode(Publisher<DataBuffer> input, Class<? extends T> outputClass,
+			Consumer<StepVerifier.FirstStep<T>> stepConsumer) {
+		testDecode(input, ResolvableType.forClass(outputClass), stepConsumer, null, null);
+	}
+
+	/**
+	 * Test a standard {@link Decoder#decode decode} scenario. For example:
+	 * <pre class="code">
+	 * byte[] bytes1 = ...
+	 * byte[] bytes2 = ...
+	 *
+	 * Flux&lt;DataBuffer&gt; input = Flux.concat(
+	 *   dataBuffer(bytes1),
+	 *   dataBuffer(bytes2));
+	 *
+	 * testDecodeAll(input, byte[].class, step -&gt; step
+	 *   .consumeNextWith(expectBytes(bytes1))
+	 *   .consumeNextWith(expectBytes(bytes2))
+	 * 	 .verifyComplete());
+	 * </pre>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 * @param <T> the output type
+	 */
+	@SuppressWarnings("unchecked")
+	protected <T> void testDecode(Publisher<DataBuffer> input, ResolvableType outputType,
+			Consumer<StepVerifier.FirstStep<T>> stepConsumer,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Flux<T> result = (Flux<T>) this.decoder.decode(input, outputType, mimeType, hints);
+		StepVerifier.FirstStep<T> step = StepVerifier.create(result);
+		stepConsumer.accept(step);
+	}
+
+	/**
+	 * Test a {@link Decoder#decode decode} scenario where the input stream contains an error.
+	 * This test method will feed the first element of the {@code input} stream to the decoder,
+	 * followed by an {@link InputException}.
+	 * The result is expected to contain one "normal" element, followed by the error.
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 * @see InputException
+	 */
+	protected void testDecodeError(Publisher<DataBuffer> input, ResolvableType outputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		input = Flux.concat(
+				Flux.from(input).take(1),
+				Flux.error(new InputException()));
+
+		Flux<?> result = this.decoder.decode(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.expectNextCount(1)
+				.expectError(InputException.class)
+				.verify();
+	}
+
+	/**
+	 * Test a {@link Decoder#decode decode} scenario where the input stream is canceled.
+	 * This test method will feed the first element of the {@code input} stream to the decoder,
+	 * followed by a cancel signal.
+	 * The result is expected to contain one "normal" element.
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 */
+	protected void testDecodeCancel(Publisher<DataBuffer> input, ResolvableType outputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Flux<?> result = this.decoder.decode(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.expectNextCount(1)
+				.thenCancel()
+				.verify();
+	}
+
+	/**
+	 * Test a {@link Decoder#decode decode} scenario where the input stream is empty.
+	 * The output is expected to be empty as well.
+	 *
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 */
+	protected void testDecodeEmpty(ResolvableType outputType, @Nullable MimeType mimeType,
+			@Nullable Map<String, Object> hints) {
+
+		Flux<DataBuffer> input = Flux.empty();
+		Flux<?> result = this.decoder.decode(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.verifyComplete();
+	}
+
+	// Mono
+
+	/**
+	 * Helper methods that tests for a variety of {@link Mono} decoding scenarios. This methods
+	 * invokes:
+	 * <ul>
+	 *     <li>{@link #testDecodeToMono(Publisher, ResolvableType, Consumer, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoError(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoCancel(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoEmpty(ResolvableType, MimeType, Map)}</li>
+	 * </ul>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputClass the desired output class
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param <T> the output type
+	 */
+	protected <T> void testDecodeToMonoAll(Publisher<DataBuffer> input,
+			Class<? extends T> outputClass, Consumer<StepVerifier.FirstStep<T>> stepConsumer) {
+
+		testDecodeToMonoAll(input, ResolvableType.forClass(outputClass), stepConsumer, null, null);
+	}
+
+	/**
+	 * Helper methods that tests for a variety of {@link Mono} decoding scenarios. This methods
+	 * invokes:
+	 * <ul>
+	 *     <li>{@link #testDecodeToMono(Publisher, ResolvableType, Consumer, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoError(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoCancel(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoEmpty(ResolvableType, MimeType, Map)}</li>
+	 * </ul>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 * @param <T> the output type
+	 */
+	protected <T> void testDecodeToMonoAll(Publisher<DataBuffer> input, ResolvableType outputType,
+			Consumer<StepVerifier.FirstStep<T>> stepConsumer,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+		testDecodeToMono(input, outputType, stepConsumer, mimeType, hints);
+		testDecodeToMonoError(input, outputType, mimeType, hints);
+		testDecodeToMonoCancel(input, outputType, mimeType, hints);
+		testDecodeToMonoEmpty(outputType, mimeType, hints);
+	}
+
+	/**
+	 * Test a standard {@link Decoder#decodeToMono) decode} scenario. For example:
+	 * <pre class="code">
+	 * byte[] bytes1 = ...
+	 * byte[] bytes2 = ...
+	 * byte[] allBytes = ... // bytes1 + bytes2
+	 *
+	 * Flux&lt;DataBuffer&gt; input = Flux.concat(
+	 *   dataBuffer(bytes1),
+	 *   dataBuffer(bytes2));
+	 *
+	 * testDecodeAll(input, byte[].class, step -&gt; step
+	 *   .consumeNextWith(expectBytes(allBytes))
+	 * 	 .verifyComplete());
+	 * </pre>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputClass the desired output class
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param <T> the output type
+	 */
+	protected <T> void testDecodeToMono(Publisher<DataBuffer> input,
+			Class<? extends T> outputClass, Consumer<StepVerifier.FirstStep<T>> stepConsumer) {
+		testDecodeToMono(input, ResolvableType.forClass(outputClass), stepConsumer, null, null);
+	}
+
+	/**
+	 * Test a standard {@link Decoder#decodeToMono) decode} scenario. For example:
+	 * <pre class="code">
+	 * byte[] bytes1 = ...
+	 * byte[] bytes2 = ...
+	 * byte[] allBytes = ... // bytes1 + bytes2
+	 *
+	 * Flux&lt;DataBuffer&gt; input = Flux.concat(
+	 *   dataBuffer(bytes1),
+	 *   dataBuffer(bytes2));
+	 *
+	 * testDecodeAll(input, byte[].class, step -&gt; step
+	 *   .consumeNextWith(expectBytes(allBytes))
+	 * 	 .verifyComplete());
+	 * </pre>
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 * @param <T> the output type
+	 */
+	@SuppressWarnings("unchecked")
+	protected <T> void testDecodeToMono(Publisher<DataBuffer> input, ResolvableType outputType,
+			Consumer<StepVerifier.FirstStep<T>> stepConsumer,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Mono<T> result = (Mono<T>) this.decoder.decodeToMono(input, outputType, mimeType, hints);
+		StepVerifier.FirstStep<T> step = StepVerifier.create(result);
+		stepConsumer.accept(step);
+	}
+
+	/**
+	 * Test a {@link Decoder#decodeToMono decode} scenario where the input stream contains an error.
+	 * This test method will feed the first element of the {@code input} stream to the decoder,
+	 * followed by an {@link InputException}.
+	 * The result is expected to contain the error.
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 * @see InputException
+	 */
+	protected void testDecodeToMonoError(Publisher<DataBuffer> input, ResolvableType outputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		input = Flux.concat(
+				Flux.from(input).take(1),
+				Flux.error(new InputException()));
+
+		Mono<?> result = this.decoder.decodeToMono(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.expectError(InputException.class)
+				.verify();
+	}
+
+	/**
+	 * Test a {@link Decoder#decodeToMono decode} scenario where the input stream is canceled.
+	 * This test method will immediately cancel the output stream.
+	 *
+	 * @param input the input to be provided to the decoder
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 */
+	protected void testDecodeToMonoCancel(Publisher<DataBuffer> input, ResolvableType outputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Mono<?> result = this.decoder.decodeToMono(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.thenCancel()
+				.verify();
+	}
+
+	/**
+	 * Test a {@link Decoder#decodeToMono decode} scenario where the input stream is empty.
+	 * The output is expected to be empty as well.
+	 *
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 */
+	protected void testDecodeToMonoEmpty(ResolvableType outputType, @Nullable MimeType mimeType,
+			@Nullable Map<String, Object> hints) {
+
+		Flux<DataBuffer> input = Flux.empty();
+		Mono<?> result = this.decoder.decodeToMono(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.verifyComplete();
+	}
+
+	/**
+	 * Creates a deferred {@link DataBuffer} containing the given bytes.
+	 * @param bytes the bytes that are to be stored in the buffer
+	 * @return the deferred buffer
+	 */
+	protected Mono<DataBuffer> dataBuffer(byte[] bytes) {
+		return Mono.defer(() -> {
+			DataBuffer dataBuffer = this.bufferFactory.allocateBuffer(bytes.length);
+			dataBuffer.write(bytes);
+			return Mono.just(dataBuffer);
+		});
+	}
+
+	/**
+	 * Exception used in {@link #testDecodeError} and {@link #testDecodeToMonoError}
+	 */
+	@SuppressWarnings("serial")
+	public static class InputException extends RuntimeException {
+
+	}
+
+
+}

spring-core/src/test/java/org/springframework/core/codec/AbstractEncoderTestCase.java

@@ -20,17 +20,16 @@ import java.util.Map;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
-import org.junit.After;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DataBufferUtils;
-import org.springframework.core.io.buffer.LeakAwareDataBufferFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
@@ -46,13 +45,8 @@ import static org.junit.Assert.*;
  * @author Arjen Poutsma
  */
 @SuppressWarnings("ProtectedField")
-public abstract class AbstractEncoderTestCase<T, E extends Encoder<T>> {
+public abstract class AbstractEncoderTestCase<T, E extends Encoder<T>> extends
-
+		AbstractLeakCheckingTestCase {
-	/**
-	 * The data buffer factory used by the encoder.
-	 */
-	protected final DataBufferFactory bufferFactory =
-			new LeakAwareDataBufferFactory();
 
 	/**
 	 * The encoder to test.
@@ -110,15 +104,6 @@ public abstract class AbstractEncoderTestCase<T, E extends Encoder<T>> {
 		this.hints = hints;
 	}
 
-	/**
-	 * Checks whether any of the data buffers created by {@link #bufferFactory} have not been
-	 * released, throwing an assertion error if so.
-	 */
-	@After
-	public final void checkForLeaks() {
-		((LeakAwareDataBufferFactory) this.bufferFactory).checkForLeaks();
-	}
-
 	/**
 	 * Abstract template method that provides input for the encoder.
 	 * Used for {@link #encode()}, {@link #encodeError()}, and {@link #encodeCancel()}.

spring-core/src/test/java/org/springframework/core/codec/ByteArrayDecoderTests.java

@@ -16,16 +16,13 @@
 
 package org.springframework.core.codec;
 
-import java.util.Collections;
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
 
 import org.junit.Test;
-import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
@@ -34,11 +31,18 @@ import static org.junit.Assert.*;
 /**
  * @author Arjen Poutsma
  */
-public class ByteArrayDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ByteArrayDecoderTests extends AbstractDecoderTestCase<ByteArrayDecoder> {
 
-	private final ByteArrayDecoder decoder = new ByteArrayDecoder();
+	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
+	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
 
+
+	public ByteArrayDecoderTests() {
+		super(new ByteArrayDecoder());
+	}
+
+	@Override
 	@Test
 	public void canDecode() {
 		assertTrue(this.decoder.canDecode(ResolvableType.forClass(byte[].class),
@@ -49,50 +53,38 @@ public class ByteArrayDecoderTests extends AbstractDataBufferAllocatingTestCase
 				MimeTypeUtils.APPLICATION_JSON));
 	}
 
+	@Override
 	@Test
 	public void decode() {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.concat(
-		DataBuffer barBuffer = stringBuffer("bar");
+				dataBuffer(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				dataBuffer(this.barBytes));
-		Flux<byte[]> output = this.decoder.decode(source,
+
-				ResolvableType.forClassWithGenerics(Publisher.class, byte[].class),
+		testDecodeAll(input, byte[].class, step -> step
-				null, Collections.emptyMap());
+				.consumeNextWith(expectBytes(this.fooBytes))
-
+				.consumeNextWith(expectBytes(this.barBytes))
-		StepVerifier.create(output)
+				.verifyComplete());
-				.consumeNextWith(bytes -> assertArrayEquals("foo".getBytes(), bytes))
-				.consumeNextWith(bytes -> assertArrayEquals("bar".getBytes(), bytes))
-				.expectComplete()
-				.verify();
-	}
 
-	@Test
-	public void decodeError() {
-		DataBuffer fooBuffer = stringBuffer("foo");
-		Flux<DataBuffer> source =
-				Flux.just(fooBuffer).concatWith(Flux.error(new RuntimeException()));
-		Flux<byte[]> output = this.decoder.decode(source,
-				ResolvableType.forClassWithGenerics(Publisher.class, byte[].class),
-				null, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.consumeNextWith(bytes -> assertArrayEquals("foo".getBytes(), bytes))
-				.expectError()
-				.verify();
 	}
 
+	@Override
 	@Test
 	public void decodeToMono() {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.concat(
-		DataBuffer barBuffer = stringBuffer("bar");
+				dataBuffer(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				dataBuffer(this.barBytes));
-		Mono<byte[]> output = this.decoder.decodeToMono(source,
+
-				ResolvableType.forClassWithGenerics(Publisher.class, byte[].class),
+		byte[] expected = new byte[this.fooBytes.length + this.barBytes.length];
-				null, Collections.emptyMap());
+		System.arraycopy(this.fooBytes, 0, expected, 0, this.fooBytes.length);
-
+		System.arraycopy(this.barBytes, 0, expected, this.fooBytes.length, this.barBytes.length);
-		StepVerifier.create(output)
+
-				.consumeNextWith(bytes -> assertArrayEquals("foobar".getBytes(), bytes))
+		testDecodeToMonoAll(input, byte[].class, step -> step
-				.expectComplete()
+				.consumeNextWith(expectBytes(expected))
-				.verify();
+				.verifyComplete());
+	}
+
+	private Consumer<byte[]> expectBytes(byte[] expected) {
+		return bytes -> assertArrayEquals(expected, bytes);
 	}
 
 }

spring-core/src/test/java/org/springframework/core/codec/ByteBufferDecoderTests.java

@@ -17,16 +17,13 @@
 package org.springframework.core.codec;
 
 import java.nio.ByteBuffer;
-import java.util.Collections;
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
 
 import org.junit.Test;
-import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
@@ -35,10 +32,18 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class ByteBufferDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ByteBufferDecoderTests extends AbstractDecoderTestCase<ByteBufferDecoder> {
 
-	private final ByteBufferDecoder decoder = new ByteBufferDecoder();
+	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
+	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
+
+
+	public ByteBufferDecoderTests() {
+		super(new ByteBufferDecoder());
+	}
+
+	@Override
 	@Test
 	public void canDecode() {
 		assertTrue(this.decoder.canDecode(ResolvableType.forClass(ByteBuffer.class),
@@ -49,48 +54,38 @@ public class ByteBufferDecoderTests extends AbstractDataBufferAllocatingTestCase
 				MimeTypeUtils.APPLICATION_JSON));
 	}
 
+	@Override
 	@Test
 	public void decode() {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.concat(
-		DataBuffer barBuffer = stringBuffer("bar");
+				dataBuffer(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				dataBuffer(this.barBytes));
-		Flux<ByteBuffer> output = this.decoder.decode(source,
+
-				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
+		testDecodeAll(input, ByteBuffer.class, step -> step
-				null, Collections.emptyMap());
+				.consumeNextWith(expectByteBuffer(ByteBuffer.wrap(this.fooBytes)))
-
+				.consumeNextWith(expectByteBuffer(ByteBuffer.wrap(this.barBytes)))
-		StepVerifier.create(output)
+				.verifyComplete());
-				.expectNext(ByteBuffer.wrap("foo".getBytes()), ByteBuffer.wrap("bar".getBytes()))
+
-				.expectComplete()
-				.verify();
-	}
 
-	@Test
-	public void decodeError() {
-		DataBuffer fooBuffer = stringBuffer("foo");
-		Flux<DataBuffer> source =
-				Flux.just(fooBuffer).concatWith(Flux.error(new RuntimeException()));
-		Flux<ByteBuffer> output = this.decoder.decode(source,
-				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
-				null, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.expectNext(ByteBuffer.wrap("foo".getBytes()))
-				.expectError()
-				.verify();
 	}
 
+	@Override
 	@Test
 	public void decodeToMono() {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.concat(
-		DataBuffer barBuffer = stringBuffer("bar");
+				dataBuffer(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				dataBuffer(this.barBytes));
-		Mono<ByteBuffer> output = this.decoder.decodeToMono(source,
+		ByteBuffer expected = ByteBuffer.allocate(this.fooBytes.length + this.barBytes.length);
-				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
+		expected.put(this.fooBytes).put(this.barBytes).flip();
-				null, Collections.emptyMap());
+
-
+		testDecodeToMonoAll(input, ByteBuffer.class, step -> step
-		StepVerifier.create(output)
+				.consumeNextWith(expectByteBuffer(expected))
-				.expectNext(ByteBuffer.wrap("foobar".getBytes()))
+				.verifyComplete());
-				.expectComplete()
+
-				.verify();
 	}
+
+	private Consumer<ByteBuffer> expectByteBuffer(ByteBuffer expected) {
+		return actual -> assertEquals(expected, actual);
+	}
+
 }

spring-core/src/test/java/org/springframework/core/codec/DataBufferDecoderTests.java

@@ -17,18 +17,14 @@
 package org.springframework.core.codec;
 
 import java.nio.charset.StandardCharsets;
-import java.time.Duration;
+import java.util.function.Consumer;
-import java.util.Collections;
 
 import org.junit.Test;
-import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.support.DataBufferTestUtils;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.junit.Assert.*;
@@ -36,10 +32,18 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class DataBufferDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class DataBufferDecoderTests extends AbstractDecoderTestCase<DataBufferDecoder> {
 
-	private final DataBufferDecoder decoder = new DataBufferDecoder();
+	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
+	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
+
+
+	public DataBufferDecoderTests() {
+		super(new DataBufferDecoder());
+	}
+
+	@Override
 	@Test
 	public void canDecode() {
 		assertTrue(this.decoder.canDecode(ResolvableType.forClass(DataBuffer.class),
@@ -50,33 +54,40 @@ public class DataBufferDecoderTests extends AbstractDataBufferAllocatingTestCase
 				MimeTypeUtils.APPLICATION_JSON));
 	}
 
-	@Test
+	@Override
 	public void decode() {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.just(
-		DataBuffer barBuffer = stringBuffer("bar");
+				this.bufferFactory.wrap(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				this.bufferFactory.wrap(this.barBytes));
-		Flux<DataBuffer> output = this.decoder.decode(source,
-				ResolvableType.forClassWithGenerics(Publisher.class, DataBuffer.class),
-				null, Collections.emptyMap());
-
-		assertSame(source, output);
 
-		release(fooBuffer, barBuffer);
+		testDecodeAll(input, DataBuffer.class, step -> step
+				.consumeNextWith(expectDataBuffer(this.fooBytes))
+				.consumeNextWith(expectDataBuffer(this.barBytes))
+				.verifyComplete());
 	}
 
-	@Test
+	@Override
-	public void decodeToMono() {
+	public void decodeToMono() throws Exception {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.concat(
-		DataBuffer barBuffer = stringBuffer("bar");
+				dataBuffer(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				dataBuffer(this.barBytes));
-		Mono<DataBuffer> output = this.decoder.decodeToMono(source,
+
-				ResolvableType.forClassWithGenerics(Publisher.class, DataBuffer.class),
+		byte[] expected = new byte[this.fooBytes.length + this.barBytes.length];
-				null, Collections.emptyMap());
+		System.arraycopy(this.fooBytes, 0, expected, 0, this.fooBytes.length);
-
+		System.arraycopy(this.barBytes, 0, expected, this.fooBytes.length, this.barBytes.length);
-		DataBuffer outputBuffer = output.block(Duration.ofSeconds(5));
+
-		assertEquals("foobar", DataBufferTestUtils.dumpString(outputBuffer, StandardCharsets.UTF_8));
+		testDecodeToMonoAll(input, DataBuffer.class, step -> step
-
+				.consumeNextWith(expectDataBuffer(expected))
-		release(outputBuffer);
+				.verifyComplete());
 	}
 
+	private Consumer<DataBuffer> expectDataBuffer(byte[] expected) {
+		return actual -> {
+			byte[] actualBytes = new byte[actual.readableByteCount()];
+			actual.read(actualBytes);
+			assertArrayEquals(expected, actualBytes);
+
+			DataBufferUtils.release(actual);
+		};
+	}
 }

spring-core/src/test/java/org/springframework/core/codec/ResourceDecoderTests.java

@@ -17,17 +17,21 @@
 package org.springframework.core.codec;
 
 import java.io.IOException;
-import java.util.Collections;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
 
 import org.junit.Test;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
+import org.springframework.core.ResolvableType;
 import org.springframework.core.io.ByteArrayResource;
 import org.springframework.core.io.InputStreamResource;
 import org.springframework.core.io.Resource;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StreamUtils;
 
@@ -37,10 +41,18 @@ import static org.springframework.core.ResolvableType.forClass;
 /**
  * @author Arjen Poutsma
  */
-public class ResourceDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ResourceDecoderTests extends AbstractDecoderTestCase<ResourceDecoder> {
 
-	private final ResourceDecoder decoder = new ResourceDecoder();
+	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
+	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
+
+
+	public ResourceDecoderTests() {
+		super(new ResourceDecoder());
+	}
+
+	@Override
 	@Test
 	public void canDecode() {
 		assertTrue(this.decoder.canDecode(forClass(InputStreamResource.class), MimeTypeUtils.TEXT_PLAIN));
@@ -50,16 +62,15 @@ public class ResourceDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		assertFalse(this.decoder.canDecode(forClass(Object.class), MimeTypeUtils.APPLICATION_JSON));
 	}
 
+
+	@Override
 	@Test
 	public void decode() {
-		DataBuffer fooBuffer = stringBuffer("foo");
+		Flux<DataBuffer> input = Flux.concat(
-		DataBuffer barBuffer = stringBuffer("bar");
+				dataBuffer(this.fooBytes),
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
+				dataBuffer(this.barBytes));
-
-		Flux<Resource> result = this.decoder
-				.decode(source, forClass(Resource.class), null, Collections.emptyMap());
 
-		StepVerifier.create(result)
+		testDecodeAll(input, Resource.class, step -> step
 				.consumeNextWith(resource -> {
 					try {
 						byte[] bytes = StreamUtils.copyToByteArray(resource.getInputStream());
@@ -70,22 +81,42 @@ public class ResourceDecoderTests extends AbstractDataBufferAllocatingTestCase {
 					}
 				})
 				.expectComplete()
-				.verify();
+				.verify());
 	}
 
-	@Test
+	@Override
-	public void decodeError() {
+	protected void testDecodeError(Publisher<DataBuffer> input, ResolvableType outputType,
-		DataBuffer fooBuffer = stringBuffer("foo");
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
-		Flux<DataBuffer> source =
-				Flux.just(fooBuffer).concatWith(Flux.error(new RuntimeException()));
 
+		input = Flux.concat(
+				Flux.from(input).take(1),
+				Flux.error(new InputException()));
 
-		Flux<Resource> result = this.decoder
+		Flux<Resource> result = this.decoder.decode(input, outputType, mimeType, hints);
-				.decode(source, forClass(Resource.class), null, Collections.emptyMap());
 
 		StepVerifier.create(result)
-				.expectError()
+				.expectError(InputException.class)
 				.verify();
 	}
 
+	@Override
+	public void decodeToMono() throws Exception {
+		Flux<DataBuffer> input = Flux.concat(
+				dataBuffer(this.fooBytes),
+				dataBuffer(this.barBytes));
+
+		testDecodeToMonoAll(input, Resource.class, step -> step
+				.consumeNextWith(resource -> {
+					try {
+						byte[] bytes = StreamUtils.copyToByteArray(resource.getInputStream());
+						assertEquals("foobar", new String(bytes));
+					}
+					catch (IOException e) {
+						fail(e.getMessage());
+					}
+				})
+				.expectComplete()
+				.verify());
+	}
+
 }

spring-core/src/test/java/org/springframework/core/codec/StringDecoderTests.java

@@ -17,18 +17,20 @@
 package org.springframework.core.codec;
 
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.Test;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.lang.Nullable;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
@@ -42,25 +44,28 @@ import static org.junit.Assert.*;
  * @author Brian Clozel
  * @author Mark Paluch
  */
-public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class StringDecoderTests extends AbstractDecoderTestCase<StringDecoder> {
 
-	private StringDecoder decoder = StringDecoder.allMimeTypes();
+	private static final ResolvableType TYPE = ResolvableType.forClass(String.class);
 
+	public StringDecoderTests() {
+		super(StringDecoder.allMimeTypes());
+	}
 
+	@Override
 	@Test
 	public void canDecode() {
-
 		assertTrue(this.decoder.canDecode(
-				ResolvableType.forClass(String.class), MimeTypeUtils.TEXT_PLAIN));
+				TYPE, MimeTypeUtils.TEXT_PLAIN));
 
 		assertTrue(this.decoder.canDecode(
-				ResolvableType.forClass(String.class), MimeTypeUtils.TEXT_HTML));
+				TYPE, MimeTypeUtils.TEXT_HTML));
 
 		assertTrue(this.decoder.canDecode(
-				ResolvableType.forClass(String.class), MimeTypeUtils.APPLICATION_JSON));
+				TYPE, MimeTypeUtils.APPLICATION_JSON));
 
 		assertTrue(this.decoder.canDecode(
-				ResolvableType.forClass(String.class), MimeTypeUtils.parseMimeType("text/plain;charset=utf-8")));
+				TYPE, MimeTypeUtils.parseMimeType("text/plain;charset=utf-8")));
 
 
 		assertFalse(this.decoder.canDecode(
@@ -70,19 +75,33 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				ResolvableType.forClass(Object.class), MimeTypeUtils.APPLICATION_JSON));
 	}
 
+	@Override
 	@Test
-	public void decodeMultibyteCharacter() {
+	public void decode() {
 		String u = "ü";
 		String e = "é";
 		String o = "ø";
 		String s = String.format("%s\n%s\n%s", u, e, o);
-		Flux<DataBuffer> source = toDataBuffers(s, 1, UTF_8);
+		Flux<DataBuffer> input = toDataBuffers(s, 1, UTF_8);
 
-		Flux<String> output = this.decoder.decode(source, ResolvableType.forClass(String.class),
+		testDecodeAll(input, ResolvableType.forClass(String.class), step -> step
-				null, Collections.emptyMap());
-		StepVerifier.create(output)
 				.expectNext(u, e, o)
-				.verifyComplete();
+				.verifyComplete(), null, null);
+	}
+
+	@Override
+	protected void testDecodeError(Publisher<DataBuffer> input, ResolvableType outputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		input = Flux.concat(
+				Flux.from(input).take(1),
+				Flux.error(new InputException()));
+
+		Flux<String> result = this.decoder.decode(input, outputType, mimeType, hints);
+
+		StepVerifier.create(result)
+				.expectError(InputException.class)
+				.verify();
 	}
 
 	@Test
@@ -92,13 +111,11 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		String o = "ø";
 		String s = String.format("%s\n%s\n%s", u, e, o);
 		Flux<DataBuffer> source = toDataBuffers(s, 2, UTF_16BE);
-
 		MimeType mimeType = MimeTypeUtils.parseMimeType("text/plain;charset=utf-16be");
-		Flux<String> output = this.decoder.decode(source, ResolvableType.forClass(String.class),
+
-				mimeType, Collections.emptyMap());
+		testDecode(source, TYPE, step -> step
-		StepVerifier.create(output)
 				.expectNext(u, e, o)
-				.verifyComplete();
+				.verifyComplete(), mimeType, null);
 	}
 
 	private Flux<DataBuffer> toDataBuffers(String s, int length, Charset charset) {
@@ -115,7 +132,7 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeNewLine() {
-		Flux<DataBuffer> source = Flux.just(
+		Flux<DataBuffer> input = Flux.just(
 				stringBuffer("\r\nabc\n"),
 				stringBuffer("def"),
 				stringBuffer("ghi\r\n\n"),
@@ -126,10 +143,7 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				stringBuffer("xyz")
 		);
 
-		Flux<String> output = this.decoder.decode(source, ResolvableType.forClass(String.class),
+		testDecode(input, String.class, step -> step
-				null, Collections.emptyMap());
-
-		StepVerifier.create(output)
 				.expectNext("")
 				.expectNext("abc")
 				.expectNext("defghi")
@@ -138,15 +152,15 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				.expectNext("pqr")
 				.expectNext("stuvwxyz")
 				.expectComplete()
-				.verify();
+				.verify());
 	}
 
 	@Test
 	public void decodeNewLineIncludeDelimiters() {
 
-		decoder = StringDecoder.allMimeTypes(StringDecoder.DEFAULT_DELIMITERS, false);
+		this.decoder = StringDecoder.allMimeTypes(StringDecoder.DEFAULT_DELIMITERS, false);
 
-		Flux<DataBuffer> source = Flux.just(
+		Flux<DataBuffer> input = Flux.just(
 				stringBuffer("\r\nabc\n"),
 				stringBuffer("def"),
 				stringBuffer("ghi\r\n\n"),
@@ -157,10 +171,7 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				stringBuffer("xyz")
 		);
 
-		Flux<String> output = this.decoder.decode(source, ResolvableType.forClass(String.class),
+		testDecode(input, String.class, step -> step
-				null, Collections.emptyMap());
-
-		StepVerifier.create(output)
 				.expectNext("\r\n")
 				.expectNext("abc\n")
 				.expectNext("defghi\r\n")
@@ -169,27 +180,23 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 				.expectNext("pqr\n")
 				.expectNext("stuvwxyz")
 				.expectComplete()
-				.verify();
+				.verify());
 	}
 
 	@Test
 	public void decodeEmptyFlux() {
-		Flux<DataBuffer> source = Flux.empty();
+		Flux<DataBuffer> input = Flux.empty();
-		Flux<String> output = this.decoder.decode(source, ResolvableType.forClass(String.class),
-				null, Collections.emptyMap());
 
-		StepVerifier.create(output)
+		testDecode(input, String.class, step -> step
-				.expectNextCount(0)
 				.expectComplete()
-				.verify();
+				.verify());
-
 	}
 
 	@Test
 	public void decodeEmptyDataBuffer() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(""));
+		Flux<DataBuffer> input = Flux.just(stringBuffer(""));
-		Flux<String> output = this.decoder.decode(source,
+		Flux<String> output = this.decoder.decode(input,
-				ResolvableType.forClass(String.class), null, Collections.emptyMap());
+				TYPE, null, Collections.emptyMap());
 
 		StepVerifier.create(output)
 				.expectNext("")
@@ -197,45 +204,35 @@ public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	}
 
-	@Test
+	@Override
-	public void decodeError() {
-		DataBuffer fooBuffer = stringBuffer("foo\n");
-		DataBuffer barBuffer = stringBuffer("bar");
-		Flux<DataBuffer> source =
-				Flux.just(fooBuffer, barBuffer).concatWith(Flux.error(new RuntimeException()));
-
-		Flux<String> output = this.decoder.decode(source,
-				ResolvableType.forClass(String.class), null, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.expectNext("foo")
-				.expectError()
-				.verify();
-
-	}
-
 	@Test
 	public void decodeToMono() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("foo"), stringBuffer("bar"), stringBuffer("baz"));
+		Flux<DataBuffer> input = Flux.just(
-		Mono<String> output = this.decoder.decodeToMono(source,
+				stringBuffer("foo"),
-				ResolvableType.forClass(String.class), null, Collections.emptyMap());
+				stringBuffer("bar"),
+				stringBuffer("baz"));
 
-		StepVerifier.create(output)
+		testDecodeToMonoAll(input, String.class, step -> step
 				.expectNext("foobarbaz")
 				.expectComplete()
-				.verify();
+				.verify());
 	}
 
 	@Test
 	public void decodeToMonoWithEmptyFlux() throws InterruptedException {
-		Flux<DataBuffer> source = Flux.empty();
+		Flux<DataBuffer> input = Flux.empty();
-		Mono<String> output = this.decoder.decodeToMono(source,
-				ResolvableType.forClass(String.class), null, Collections.emptyMap());
 
-		StepVerifier.create(output)
+		testDecodeToMono(input, String.class, step -> step
-				.expectNextCount(0)
 				.expectComplete()
-				.verify();
+				.verify());
+	}
+
+	private DataBuffer stringBuffer(String value) {
+		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+		buffer.write(bytes);
+		return buffer;
 	}
 
+
 }

spring-core/src/test/java/org/springframework/core/io/buffer/AbstractLeakCheckingTestCase.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2002-2018 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.io.buffer;
+
+import org.junit.After;
+
+/**
+ * Abstract base class for unit tests that allocate data buffers via a {@link DataBufferFactory}.
+ * After each unit test, this base class checks whether all created buffers have been released,
+ * throwing an {@link AssertionError} if not.
+ *
+ * @author Arjen Poutsma
+ * @since 5.1.3
+ * @see LeakAwareDataBufferFactory
+ */
+public abstract class AbstractLeakCheckingTestCase {
+
+	/**
+	 * The data buffer factory.
+	 */
+	@SuppressWarnings("ProtectedField")
+	protected final LeakAwareDataBufferFactory bufferFactory = new LeakAwareDataBufferFactory();
+
+	/**
+	 * Checks whether any of the data buffers created by {@link #bufferFactory} have not been
+	 * released, throwing an assertion error if so.
+	 */
+	@After
+	public final void checkForLeaks() {
+		this.bufferFactory.checkForLeaks();
+	}
+
+}

spring-core/src/test/java/org/springframework/core/io/buffer/LeakAwareDataBufferFactory.java

@@ -26,28 +26,13 @@ import org.springframework.util.Assert;
 
 /**
  * Implementation of the {@code DataBufferFactory} interface that keep track of memory leaks.
- * Useful for unit tests that handle data buffers. Simply call {@link #checkForLeaks()} in
+ * Useful for unit tests that handle data buffers. Simply inherit from
- * a JUnit {@link After} method, and any buffers have not been released will result in an
+ * {@link AbstractLeakCheckingTestCase} or call {@link #checkForLeaks()} in
+ * a JUnit {@link After} method yourself, and any buffers have not been released will result in an
  * {@link AssertionError}.
- * <pre class="code">
- * public class MyUnitTest {
  *
- * 	private final LeakAwareDataBufferFactory bufferFactory =
- * 	  new LeakAwareDataBufferFactory();
- *
- *  &#064;Test
- * 	public void doSomethingWithBufferFactory() {
- * 		...
- * 	}
- *
- * 	&#064;After
- * 	public void checkForLeaks() {
- * 		bufferFactory.checkForLeaks();
- * 	}
- *
- * }
- * </pre>
  * @author Arjen Poutsma
+ * @see LeakAwareDataBufferFactory
  */
 public class LeakAwareDataBufferFactory implements DataBufferFactory {
 

spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java

@@ -124,15 +124,15 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
 					try {
 						Message.Builder builder = getMessageBuilder(elementType.toClass());
 						builder.mergeFrom(CodedInputStream.newInstance(dataBuffer.asByteBuffer()), this.extensionRegistry);
-						Message message = builder.build();
+						return builder.build();
-						DataBufferUtils.release(dataBuffer);
-						return message;
 					}
 					catch (IOException ex) {
 						throw new DecodingException("I/O error while parsing input stream", ex);
 					}
 					catch (Exception ex) {
 						throw new DecodingException("Could not read Protobuf message: " + ex.getMessage(), ex);
+					} finally {
+						DataBufferUtils.release(dataBuffer);
 					}
 				}
 		);

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

@@ -16,6 +16,7 @@
 
 package org.springframework.http.codec;
 
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 
@@ -26,7 +27,7 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
@@ -40,7 +41,7 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class FormHttpMessageReaderTests extends AbstractDataBufferAllocatingTestCase {
+public class FormHttpMessageReaderTests extends AbstractLeakCheckingTestCase {
 
 	private final FormHttpMessageReader reader = new FormHttpMessageReader();
 
@@ -127,4 +128,11 @@ public class FormHttpMessageReaderTests extends AbstractDataBufferAllocatingTest
 					.body(body);
 	}
 
+	private DataBuffer stringBuffer(String value) {
+		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+		buffer.write(bytes);
+		return buffer;
+	}
+
 }

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

@@ -16,14 +16,19 @@
 
 package org.springframework.http.codec;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import org.junit.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
@@ -35,7 +40,7 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class FormHttpMessageWriterTests extends AbstractDataBufferAllocatingTestCase {
+public class FormHttpMessageWriterTests extends AbstractLeakCheckingTestCase {
 
 	private final FormHttpMessageWriter writer = new FormHttpMessageWriter();
 
@@ -89,4 +94,14 @@ public class FormHttpMessageWriterTests extends AbstractDataBufferAllocatingTest
 		assertEquals(expected.length(), headers.getContentLength());
 	}
 
+	private Consumer<DataBuffer> stringConsumer(String expected) {
+		return dataBuffer -> {
+			String value =
+					DataBufferTestUtils.dumpString(dataBuffer, StandardCharsets.UTF_8);
+			DataBufferUtils.release(dataBuffer);
+			assertEquals(expected, value);
+		};
+	}
+
+
 }

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

@@ -16,6 +16,7 @@
 
 package org.springframework.http.codec;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Collections;
 
@@ -25,7 +26,7 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
@@ -38,7 +39,7 @@ import static org.junit.Assert.*;
  *
  * @author Sebastien Deleuze
  */
-public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAllocatingTestCase {
+public class ServerSentEventHttpMessageReaderTests extends AbstractLeakCheckingTestCase {
 
 	private ServerSentEventHttpMessageReader messageReader =
 			new ServerSentEventHttpMessageReader(new Jackson2JsonDecoder());
@@ -188,4 +189,12 @@ public class ServerSentEventHttpMessageReaderTests extends AbstractDataBufferAll
 				.verify();
 	}
 
+	private DataBuffer stringBuffer(String value) {
+		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+		buffer.write(bytes);
+		return buffer;
+	}
+
+
 }

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

@@ -23,7 +23,6 @@ import java.util.List;
 import java.util.Map;
 
 import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -35,11 +34,10 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.AbstractDecoderTestCase;
 import org.springframework.core.codec.CodecException;
 import org.springframework.core.codec.DecodingException;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.codec.Pojo;
 import org.springframework.util.MimeType;
 
@@ -62,12 +60,19 @@ import static org.springframework.http.codec.json.JacksonViewBean.MyJacksonView3
  * @author Sebastien Deleuze
  * @author Rossen Stoyanchev
  */
-public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class Jackson2JsonDecoderTests extends AbstractDecoderTestCase<Jackson2JsonDecoder> {
 
+	private Pojo pojo1 = new Pojo("f1", "b1");
+
+	private Pojo pojo2 = new Pojo("f2", "b2");
+
+	public Jackson2JsonDecoderTests() {
+		super(new Jackson2JsonDecoder());
+	}
+
+	@Override
 	@Test
 	public void canDecode() {
-		Jackson2JsonDecoder decoder = new Jackson2JsonDecoder();
-
 		assertTrue(decoder.canDecode(forClass(Pojo.class), APPLICATION_JSON));
 		assertTrue(decoder.canDecode(forClass(Pojo.class), APPLICATION_JSON_UTF8));
 		assertTrue(decoder.canDecode(forClass(Pojo.class), APPLICATION_STREAM_JSON));
@@ -93,173 +98,112 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 		decoder.getMimeTypes().add(new MimeType("text", "ecmascript"));
 	}
 
+	@Override
 	@Test
-	public void decodePojo() throws Exception {
+	public void decode() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
+		Flux<DataBuffer> input = Flux.concat(
-		ResolvableType elementType = forClass(Pojo.class);
+				stringBuffer("[{\"bar\":\"b1\",\"foo\":\"f1\"},"),
-		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null,
+				stringBuffer("{\"bar\":\"b2\",\"foo\":\"f2\"}]"));
-				emptyMap());
+
-
+		testDecodeAll(input, Pojo.class, step -> step
-		StepVerifier.create(flux)
+				.expectNext(pojo1)
-				.expectNext(new Pojo("foofoo", "barbar"))
+				.expectNext(pojo2)
-				.verifyComplete();
+				.verifyComplete());
-	}
-
-	@Test
-	public void decodePojoWithError() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foo\":}"));
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null,
-				emptyMap());
-
-		StepVerifier.create(flux).verifyError(CodecException.class);
 	}
 
-	@Test
+	@Override
-	public void decodeToList() throws Exception {
+	public void decodeToMono() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(
+		Flux<DataBuffer> input = Flux.concat(
-				"[{\"bar\":\"b1\",\"foo\":\"f1\"},{\"bar\":\"b2\",\"foo\":\"f2\"}]"));
+				stringBuffer("[{\"bar\":\"b1\",\"foo\":\"f1\"},"),
+				stringBuffer("{\"bar\":\"b2\",\"foo\":\"f2\"}]"));
 
 		ResolvableType elementType = ResolvableType.forClassWithGenerics(List.class, Pojo.class);
-		Mono<Object> mono = new Jackson2JsonDecoder().decodeToMono(source, elementType,
-				null, emptyMap());
 
-		StepVerifier.create(mono)
+		testDecodeToMonoAll(input, elementType, step -> step
 				.expectNext(asList(new Pojo("f1", "b1"), new Pojo("f2", "b2")))
 				.expectComplete()
-				.verify();
+				.verify(), null, null);
 	}
 
-	@Test
-	public void decodeArrayToFlux() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(
-				"[{\"bar\":\"b1\",\"foo\":\"f1\"},{\"bar\":\"b2\",\"foo\":\"f2\"}]"));
-
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null,
-				emptyMap());
-
-		StepVerifier.create(flux)
-				.expectNext(new Pojo("f1", "b1"))
-				.expectNext(new Pojo("f2", "b2"))
-				.verifyComplete();
-	}
 
 	@Test
-	public void decodeStreamToFlux() throws Exception {
+	public void decodeEmptyArrayToFlux() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"bar\":\"b1\",\"foo\":\"f1\"}"),
+		Flux<DataBuffer> input = Flux.from(stringBuffer("[]"));
-				stringBuffer("{\"bar\":\"b2\",\"foo\":\"f2\"}"));
-
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, APPLICATION_STREAM_JSON,
-				emptyMap());
-
-		StepVerifier.create(flux)
-				.expectNext(new Pojo("f1", "b1"))
-				.expectNext(new Pojo("f2", "b2"))
-				.verifyComplete();
-	}
 
-	@Test
+		testDecode(input, Pojo.class, step -> step.verifyComplete());
-	public void decodeEmptyArrayToFlux() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("[]"));
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null, emptyMap());
-
-		StepVerifier.create(flux)
-				.expectNextCount(0)
-				.verifyComplete();
 	}
 
 	@Test
-	public void fieldLevelJsonView() throws Exception {
+	public void fieldLevelJsonView() {
-		Flux<DataBuffer> source = Flux.just(
+		Flux<DataBuffer> input = Flux.from(
 				stringBuffer("{\"withView1\" : \"with\", \"withView2\" : \"with\", \"withoutView\" : \"without\"}"));
 		ResolvableType elementType = forClass(JacksonViewBean.class);
 		Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView1.class);
-		Flux<JacksonViewBean> flux = new Jackson2JsonDecoder()
-				.decode(source, elementType, null, hints).cast(JacksonViewBean.class);
 
-		StepVerifier.create(flux)
+		testDecode(input, elementType, step -> step
-				.consumeNextWith(b -> {
+				.consumeNextWith(o -> {
-					assertTrue(b.getWithView1().equals("with"));
+					JacksonViewBean b = (JacksonViewBean) o;
+					assertEquals("with", b.getWithView1());
 					assertNull(b.getWithView2());
 					assertNull(b.getWithoutView());
-				})
+				}), null, hints);
-				.verifyComplete();
 	}
 
 	@Test
-	public void classLevelJsonView() throws Exception {
+	public void classLevelJsonView() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(
+		Flux<DataBuffer> input = Flux.from(stringBuffer(
 				"{\"withView1\" : \"with\", \"withView2\" : \"with\", \"withoutView\" : \"without\"}"));
 		ResolvableType elementType = forClass(JacksonViewBean.class);
 		Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView3.class);
-		Flux<JacksonViewBean> flux = new Jackson2JsonDecoder()
-				.decode(source, elementType, null, hints).cast(JacksonViewBean.class);
 
-		StepVerifier.create(flux)
+		testDecode(input, elementType, step -> step
-				.consumeNextWith(b -> {
+				.consumeNextWith(o -> {
+					JacksonViewBean b = (JacksonViewBean) o;
+					assertEquals("without", b.getWithoutView());
 					assertNull(b.getWithView1());
 					assertNull(b.getWithView2());
-					assertTrue(b.getWithoutView().equals("without"));
 				})
-				.verifyComplete();
+				.verifyComplete(), null, hints);
-	}
-
-	@Test
-	public void decodeEmptyBodyToMono() throws Exception {
-		Flux<DataBuffer> source = Flux.empty();
-		ResolvableType elementType = forClass(Pojo.class);
-		Mono<Object> mono = new Jackson2JsonDecoder().decodeToMono(source, elementType, null, emptyMap());
-
-		StepVerifier.create(mono)
-				.expectNextCount(0)
-				.verifyComplete();
-	}
-
-	@Test
-	public void invalidData() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer( "{\"foofoo\": \"foofoo\", \"barbar\": \"barbar\"}"));
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = new Jackson2JsonDecoder(new ObjectMapper()).decode(source, elementType, null, emptyMap());
-		StepVerifier.create(flux).verifyErrorMatches(ex -> ex instanceof DecodingException);
 	}
 
 	@Test
-	public void error() throws Exception {
+	public void invalidData() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"foofoo\": \"foofoo\", \"barbar\":"))
+		Flux<DataBuffer> input =
-				.concatWith(Flux.error(new RuntimeException()));
+				Flux.from(stringBuffer("{\"foofoo\": \"foofoo\", \"barbar\": \"barbar\""));
-		ResolvableType elementType = forClass(Pojo.class);
+		testDecode(input, Pojo.class, step -> step
-		Flux<Object> flux = new Jackson2JsonDecoder(new ObjectMapper()).decode(source, elementType, null, emptyMap());
+				.verifyError(DecodingException.class));
-
-		StepVerifier.create(flux)
-				.expectError(RuntimeException.class)
-				.verify();
 	}
 
 	@Test
-	public void noDefaultConstructor() throws Exception {
+	public void noDefaultConstructor() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer( "{\"property1\":\"foo\",\"property2\":\"bar\"}"));
+		Flux<DataBuffer> input =
+				Flux.from(stringBuffer("{\"property1\":\"foo\",\"property2\":\"bar\"}"));
 		ResolvableType elementType = forClass(BeanWithNoDefaultConstructor.class);
-		Flux<Object> flux = new Jackson2JsonDecoder().decode(source, elementType, null, emptyMap());
+		Flux<Object> flux = new Jackson2JsonDecoder().decode(input, elementType, null, emptyMap());
 		StepVerifier.create(flux).verifyError(CodecException.class);
 	}
 
 	@Test  // SPR-15975
 	public void  customDeserializer() {
-		DataBuffer buffer = new DefaultDataBufferFactory().wrap("{\"test\": 1}".getBytes());
+		Mono<DataBuffer> input = stringBuffer("{\"test\": 1}");
 
-		Jackson2JsonDecoder decoder = new Jackson2JsonDecoder(new ObjectMapper());
+		testDecode(input, TestObject.class, step -> step
-		Flux<TestObject> decoded = decoder.decode(Mono.just(buffer),
+				.consumeNextWith(o -> assertEquals(1, o.getTest()))
-				ResolvableType.forClass(TestObject.class), null, null).cast(TestObject.class);
+				.verifyComplete()
+		);
+	}
 
-		StepVerifier.create(decoded)
+	private Mono<DataBuffer> stringBuffer(String value) {
-				.assertNext(v -> assertEquals(1, v.getTest()))
+		return Mono.defer(() -> {
-				.verifyComplete();
+			byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+			DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+			buffer.write(bytes);
+			return Mono.just(buffer);
+		});
 	}
 
 
+
 	private static class BeanWithNoDefaultConstructor {
 
 		private final String property1;
@@ -272,11 +216,11 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 		}
 
 		public String getProperty1() {
-			return property1;
+			return this.property1;
 		}
 
 		public String getProperty2() {
-			return property2;
+			return this.property2;
 		}
 
 	}
@@ -285,7 +229,7 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 	public static class TestObject {
 		private int test;
 		public int getTest() {
-			return test;
+			return this.test;
 		}
 		public void setTest(int test) {
 			this.test = test;
@@ -302,7 +246,7 @@ public class Jackson2JsonDecoderTests extends AbstractDataBufferAllocatingTestCa
 
 		@Override
 		public TestObject deserialize(JsonParser p,
-				DeserializationContext ctxt) throws IOException, JsonProcessingException {
+				DeserializationContext ctxt) throws IOException {
 			JsonNode node = p.readValueAsTree();
 			TestObject result = new TestObject();
 			result.setTest(node.get("test").asInt());

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

@@ -16,26 +16,22 @@
 
 package org.springframework.http.codec.json;
 
+import java.util.Arrays;
 import java.util.List;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.codec.CodecException;
+import org.springframework.core.codec.AbstractDecoderTestCase;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.codec.Pojo;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 import org.springframework.util.MimeType;
 
-import static java.util.Arrays.asList;
+import static org.junit.Assert.*;
-import static java.util.Collections.emptyMap;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
 import static org.springframework.core.ResolvableType.forClass;
 import static org.springframework.http.MediaType.APPLICATION_JSON;
 
@@ -44,13 +40,22 @@ import static org.springframework.http.MediaType.APPLICATION_JSON;
  *
  * @author Sebastien Deleuze
  */
-public class Jackson2SmileDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class Jackson2SmileDecoderTests extends AbstractDecoderTestCase<Jackson2SmileDecoder> {
 
 	private final static MimeType SMILE_MIME_TYPE = new MimeType("application", "x-jackson-smile");
 	private final static MimeType STREAM_SMILE_MIME_TYPE = new MimeType("application", "stream+x-jackson-smile");
 
-	private final Jackson2SmileDecoder decoder = new Jackson2SmileDecoder();
+	private Pojo pojo1 = new Pojo("f1", "b1");
 
+	private Pojo pojo2 = new Pojo("f2", "b2");
+
+	private ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
+
+	public Jackson2SmileDecoderTests() {
+		super(new Jackson2SmileDecoder());
+	}
+
+	@Override
 	@Test
 	public void canDecode() {
 		assertTrue(decoder.canDecode(forClass(Pojo.class), SMILE_MIME_TYPE));
@@ -61,76 +66,42 @@ public class Jackson2SmileDecoderTests extends AbstractDataBufferAllocatingTestC
 		assertFalse(decoder.canDecode(forClass(Pojo.class), APPLICATION_JSON));
 	}
 
-	@Test
+	@Override
-	public void decodePojo() throws Exception {
+	public void decode() {
-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
+		Flux<DataBuffer> input = Flux.just(this.pojo1, this.pojo2)
-		Pojo pojo = new Pojo("foo", "bar");
+				.map(this::writeObject)
-		byte[] serializedPojo = mapper.writer().writeValueAsBytes(pojo);
+				.flatMap(this::dataBuffer);
-
-		Flux<DataBuffer> source = Flux.just(this.bufferFactory.wrap(serializedPojo));
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = decoder.decode(source, elementType, null, emptyMap());
-
-		StepVerifier.create(flux)
-				.expectNext(pojo)
-				.verifyComplete();
-	}
 
-	@Test
+		testDecodeAll(input, Pojo.class, step -> step
-	public void decodePojoWithError() throws Exception {
+				.expectNext(pojo1)
-		Flux<DataBuffer> source = Flux.just(stringBuffer("123"));
+				.expectNext(pojo2)
-		ResolvableType elementType = forClass(Pojo.class);
+				.verifyComplete());
-		Flux<Object> flux = decoder.decode(source, elementType, null, emptyMap());
 
-		StepVerifier.create(flux).verifyError(CodecException.class);
 	}
 
-	@Test
+	private byte[] writeObject(Object o) {
-	public void decodeToList() throws Exception {
+		try {
-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
+			return this.mapper.writer().writeValueAsBytes(o);
-		List<Pojo> list = asList(new Pojo("f1", "b1"), new Pojo("f2", "b2"));
+		}
-		byte[] serializedList = mapper.writer().writeValueAsBytes(list);
+		catch (JsonProcessingException e) {
-		Flux<DataBuffer> source = Flux.just(this.bufferFactory.wrap(serializedList));
+			throw new AssertionError(e);
+		}
 
-		ResolvableType elementType = ResolvableType.forClassWithGenerics(List.class, Pojo.class);
-		Mono<Object> mono = decoder.decodeToMono(source, elementType, null, emptyMap());
-
-		StepVerifier.create(mono)
-				.expectNext(list)
-				.expectComplete()
-				.verify();
 	}
 
-	@Test
+	@Override
-	public void decodeListToFlux() throws Exception {
+	public void decodeToMono() {
-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
+		List<Pojo> expected = Arrays.asList(pojo1, pojo2);
-		List<Pojo> list = asList(new Pojo("f1", "b1"), new Pojo("f2", "b2"));
-		byte[] serializedList = mapper.writer().writeValueAsBytes(list);
-		Flux<DataBuffer> source = Flux.just(this.bufferFactory.wrap(serializedList));
-
-		ResolvableType elementType = forClass(Pojo.class);
-		Flux<Object> flux = decoder.decode(source, elementType, null, emptyMap());
-
-		StepVerifier.create(flux)
-				.expectNext(new Pojo("f1", "b1"))
-				.expectNext(new Pojo("f2", "b2"))
-				.verifyComplete();
-	}
 
-	@Test
+		Flux<DataBuffer> input = Flux.just(expected)
-	public void decodeStreamToFlux() throws Exception {
+				.map(this::writeObject)
-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
+				.flatMap(this::dataBuffer);
-		List<Pojo> list = asList(new Pojo("f1", "b1"), new Pojo("f2", "b2"));
+
-		byte[] serializedList = mapper.writer().writeValueAsBytes(list);
+		ResolvableType elementType = ResolvableType.forClassWithGenerics(List.class, Pojo.class);
-		Flux<DataBuffer> source = Flux.just(this.bufferFactory.wrap(serializedList));
+		testDecodeToMono(input, elementType, step -> step
-
+				.expectNext(expected)
-		ResolvableType elementType = forClass(Pojo.class);
+				.expectComplete()
-		Flux<Object> flux = decoder.decode(source, elementType, STREAM_SMILE_MIME_TYPE, emptyMap());
+				.verify(), null, null);
-
-		StepVerifier.create(flux)
-				.expectNext(new Pojo("f1", "b1"))
-				.expectNext(new Pojo("f2", "b2"))
-				.verifyComplete();
 	}
 
 }

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

@@ -18,6 +18,7 @@ package org.springframework.http.codec.json;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.function.Consumer;
 
@@ -33,7 +34,7 @@ import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.codec.DecodingException;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 
 import static java.util.Arrays.asList;
@@ -43,7 +44,7 @@ import static java.util.Collections.singletonList;
  * @author Arjen Poutsma
  * @author Rossen Stoyanchev
  */
-public class Jackson2TokenizerTests extends AbstractDataBufferAllocatingTestCase {
+public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 
 	private ObjectMapper objectMapper;
 
@@ -191,11 +192,14 @@ public class Jackson2TokenizerTests extends AbstractDataBufferAllocatingTestCase
 				.verify();
 	}
 
-	@Test(expected = DecodingException.class) // SPR-16521
+	@Test // SPR-16521
 	public void jsonEOFExceptionIsWrappedAsDecodingError() {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"status\": \"noClosingQuote}"));
 		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, false);
-		tokens.blockLast();
+
+		StepVerifier.create(tokens)
+				.expectError(DecodingException.class)
+				.verify();
 	}
 
 
@@ -222,6 +226,14 @@ public class Jackson2TokenizerTests extends AbstractDataBufferAllocatingTestCase
 		builder.verifyComplete();
 	}
 
+	private DataBuffer stringBuffer(String value) {
+		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+		buffer.write(bytes);
+		return buffer;
+	}
+
+
 
 	private static class JSONAssertConsumer implements Consumer<String> {
 

spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriterTests.java

@@ -33,7 +33,7 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
@@ -51,7 +51,7 @@ import static org.junit.Assert.*;
  * @author Sebastien Deleuze
  * @author Rossen Stoyanchev
  */
-public class MultipartHttpMessageWriterTests extends AbstractDataBufferAllocatingTestCase {
+public class MultipartHttpMessageWriterTests extends AbstractLeakCheckingTestCase {
 
 	private final MultipartHttpMessageWriter writer =
 			new MultipartHttpMessageWriter(ClientCodecConfigurer.create().getWriters());

spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufDecoderTests.java

@@ -17,17 +17,17 @@
 package org.springframework.http.codec.protobuf;
 
 import java.io.IOException;
+import java.util.Arrays;
 
 import com.google.protobuf.Message;
-import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.AbstractDecoderTestCase;
 import org.springframework.core.codec.DecodingException;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.MediaType;
@@ -38,37 +38,36 @@ import org.springframework.util.MimeType;
 import static java.util.Collections.emptyMap;
 import static org.junit.Assert.*;
 import static org.springframework.core.ResolvableType.forClass;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 
 /**
  * Unit tests for {@link ProtobufDecoder}.
  *
  * @author Sebastien Deleuze
  */
-public class ProtobufDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ProtobufDecoderTests extends AbstractDecoderTestCase<ProtobufDecoder> {
 
 	private final static MimeType PROTOBUF_MIME_TYPE = new MimeType("application", "x-protobuf");
 
 	private final SecondMsg secondMsg = SecondMsg.newBuilder().setBlah(123).build();
 
-	private final Msg testMsg = Msg.newBuilder().setFoo("Foo").setBlah(secondMsg).build();
+	private final Msg testMsg1 = Msg.newBuilder().setFoo("Foo").setBlah(secondMsg).build();
 
 	private final SecondMsg secondMsg2 = SecondMsg.newBuilder().setBlah(456).build();
 
 	private final Msg testMsg2 = Msg.newBuilder().setFoo("Bar").setBlah(secondMsg2).build();
 
-	private ProtobufDecoder decoder;
+	public ProtobufDecoderTests() {
-
+		super(new ProtobufDecoder());
-
-	@Before
-	public void setup() {
-		this.decoder = new ProtobufDecoder();
 	}
 
+
 	@Test(expected = IllegalArgumentException.class)
 	public void extensionRegistryNull() {
 		new ProtobufDecoder(null);
 	}
 
+	@Override
 	@Test
 	public void canDecode() {
 		assertTrue(this.decoder.canDecode(forClass(Msg.class), null));
@@ -78,115 +77,119 @@ public class ProtobufDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		assertFalse(this.decoder.canDecode(forClass(Object.class), PROTOBUF_MIME_TYPE));
 	}
 
+	@Override
 	@Test
 	public void decodeToMono() {
-		DataBuffer data = byteBuffer(testMsg.toByteArray());
+		Mono<DataBuffer> input = dataBuffer(this.testMsg1);
-		ResolvableType elementType = forClass(Msg.class);
-
-		Mono<Message> mono = this.decoder.decodeToMono(Flux.just(data), elementType, null, emptyMap());
-
-		StepVerifier.create(mono)
-				.expectNext(testMsg)
-				.verifyComplete();
-	}
-
-	@Test
-	public void decodeToMonoWithLargerDataBuffer() {
-		DataBuffer buffer = this.bufferFactory.allocateBuffer(1024);
-		buffer.write(testMsg.toByteArray());
-		ResolvableType elementType = forClass(Msg.class);
-
-		Mono<Message> mono = this.decoder.decodeToMono(Flux.just(buffer), elementType, null, emptyMap());
 
-		StepVerifier.create(mono)
+		testDecodeToMonoAll(input, Msg.class, step -> step
-				.expectNext(testMsg)
+				.expectNext(this.testMsg1)
-				.verifyComplete();
+				.verifyComplete());
 	}
 
 	@Test
 	public void decodeChunksToMono() {
-		DataBuffer buffer = byteBuffer(testMsg.toByteArray());
+		byte[] full = this.testMsg1.toByteArray();
-		Flux<DataBuffer> chunks = Flux.just(
+		byte[] chunk1 = Arrays.copyOfRange(full, 0, full.length / 2);
-				DataBufferUtils.retain(buffer.slice(0, 4)),
+		byte[] chunk2 = Arrays.copyOfRange(full, chunk1.length, full.length);
-				DataBufferUtils.retain(buffer.slice(4, buffer.readableByteCount() - 4)));
+
-		ResolvableType elementType = forClass(Msg.class);
+		Flux<DataBuffer> input = Flux.just(chunk1, chunk2)
-		release(buffer);
+				.flatMap(bytes -> Mono.defer(() -> {
-
+					DataBuffer dataBuffer = this.bufferFactory.allocateBuffer(bytes.length);
-		Mono<Message> mono = this.decoder.decodeToMono(chunks, elementType, null,
+					dataBuffer.write(bytes);
-				emptyMap());
+					return Mono.just(dataBuffer);
-
+				}));
-		StepVerifier.create(mono)
+
-				.expectNext(testMsg)
+		testDecodeToMono(input, Msg.class, step -> step
-				.verifyComplete();
+				.expectNext(this.testMsg1)
+				.verifyComplete());
 	}
 
+	@Override
 	@Test
-	public void decode() throws IOException {
+	public void decode() {
-		DataBuffer buffer = this.bufferFactory.allocateBuffer();
+		Flux<DataBuffer> input = Flux.just(this.testMsg1, this.testMsg2)
-		testMsg.writeDelimitedTo(buffer.asOutputStream());
+				.flatMap(msg -> Mono.defer(() -> {
-		DataBuffer buffer2 = this.bufferFactory.allocateBuffer();
+					DataBuffer buffer = this.bufferFactory.allocateBuffer();
-		testMsg2.writeDelimitedTo(buffer2.asOutputStream());
+					try {
-
+						msg.writeDelimitedTo(buffer.asOutputStream());
-		Flux<DataBuffer> source = Flux.just(buffer, buffer2);
+						return Mono.just(buffer);
-		ResolvableType elementType = forClass(Msg.class);
+					}
-
+					catch (IOException e) {
-		Flux<Message> messages = this.decoder.decode(source, elementType, null, emptyMap());
+						release(buffer);
-
+						return Mono.error(e);
-		StepVerifier.create(messages)
+					}
-				.expectNext(testMsg)
+				}));
-				.expectNext(testMsg2)
+
-				.verifyComplete();
+		testDecodeAll(input, Msg.class, step -> step
+				.expectNext(this.testMsg1)
+				.expectNext(this.testMsg2)
+				.verifyComplete());
 	}
 
 	@Test
 	public void decodeSplitChunks() throws IOException {
-		DataBuffer buffer = this.bufferFactory.allocateBuffer();
+		Flux<DataBuffer> input = Flux.just(this.testMsg1, this.testMsg2)
-		testMsg.writeDelimitedTo(buffer.asOutputStream());
+				.flatMap(msg -> Mono.defer(() -> {
-		DataBuffer buffer2 = this.bufferFactory.allocateBuffer();
+					DataBuffer buffer = this.bufferFactory.allocateBuffer();
-		testMsg2.writeDelimitedTo(buffer2.asOutputStream());
+					try {
-
+						msg.writeDelimitedTo(buffer.asOutputStream());
-		Flux<DataBuffer> chunks = Flux.just(
+						return Mono.just(buffer);
-				DataBufferUtils.retain(buffer.slice(0, 4)),
+					}
-				DataBufferUtils.retain(buffer.slice(4, buffer.readableByteCount() - 4)),
+					catch (IOException e) {
-				DataBufferUtils.retain(buffer2.slice(0, 2)),
+						release(buffer);
-				DataBufferUtils.retain(buffer2
+						return Mono.error(e);
-						.slice(2, buffer2.readableByteCount() - 2)));
+					}
-		release(buffer, buffer2);
+				}))
-
+				.flatMap(buffer -> {
-		ResolvableType elementType = forClass(Msg.class);
+					int len = buffer.readableByteCount() / 2;
-		Flux<Message> messages = this.decoder.decode(chunks, elementType, null, emptyMap());
+					Flux<DataBuffer> result = Flux.just(
-
+							DataBufferUtils.retain(buffer.slice(0, len)),
-		StepVerifier.create(messages)
+							DataBufferUtils
-				.expectNext(testMsg)
+									.retain(buffer.slice(len, buffer.readableByteCount() - len))
-				.expectNext(testMsg2)
+					);
-				.verifyComplete();
+					release(buffer);
+					return result;
+				});
+
+		testDecode(input, Msg.class, step -> step
+				.expectNext(this.testMsg1)
+				.expectNext(this.testMsg2)
+				.verifyComplete());
 	}
 
 	@Test
 	public void decodeMergedChunks() throws IOException {
-		DataBuffer buffer = bufferFactory.allocateBuffer();
+		DataBuffer buffer = this.bufferFactory.allocateBuffer();
-		testMsg.writeDelimitedTo(buffer.asOutputStream());
+		this.testMsg1.writeDelimitedTo(buffer.asOutputStream());
-		testMsg.writeDelimitedTo(buffer.asOutputStream());
+		this.testMsg1.writeDelimitedTo(buffer.asOutputStream());
 
 		ResolvableType elementType = forClass(Msg.class);
 		Flux<Message> messages = this.decoder.decode(Mono.just(buffer), elementType, null, emptyMap());
 
 		StepVerifier.create(messages)
-				.expectNext(testMsg)
+				.expectNext(testMsg1)
-				.expectNext(testMsg)
+				.expectNext(testMsg1)
 				.verifyComplete();
 	}
 
 	@Test
 	public void exceedMaxSize() {
 		this.decoder.setMaxMessageSize(1);
-		Flux<DataBuffer> source = Flux.just(byteBuffer(testMsg.toByteArray()));
+		Mono<DataBuffer> input = dataBuffer(this.testMsg1);
-		ResolvableType elementType = forClass(Msg.class);
-		Flux<Message> messages = this.decoder.decode(source, elementType, null,
-				emptyMap());
 
-		StepVerifier.create(messages)
+		testDecode(input, Msg.class, step -> step
-				.verifyError(DecodingException.class);
+				.verifyError(DecodingException.class));
+	}
+
+	private Mono<DataBuffer> dataBuffer(Msg msg) {
+		return Mono.defer(() -> {
+			byte[] bytes = msg.toByteArray();
+			DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+			buffer.write(bytes);
+			return Mono.just(buffer);
+		});
 	}
 
+
 }

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

@@ -16,6 +16,7 @@
 
 package org.springframework.http.codec.xml;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.List;
 import javax.xml.namespace.QName;
@@ -27,7 +28,7 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
@@ -43,7 +44,7 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class Jaxb2XmlDecoderTests extends AbstractLeakCheckingTestCase {
 
 	private static final String POJO_ROOT = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
 			"<pojo>" +
@@ -87,7 +88,7 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 	@Test
 	public void splitOneBranches() {
 		Flux<XMLEvent> xmlEvents = this.xmlEventDecoder
-				.decode(Flux.just(stringBuffer(POJO_ROOT)), null, null, Collections.emptyMap());
+				.decode(stringBuffer(POJO_ROOT), null, null, Collections.emptyMap());
 		Flux<List<XMLEvent>> result = this.decoder.split(xmlEvents, new QName("pojo"));
 
 		StepVerifier.create(result)
@@ -109,7 +110,7 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 	@Test
 	public void splitMultipleBranches() throws Exception {
 		Flux<XMLEvent> xmlEvents = this.xmlEventDecoder
-				.decode(Flux.just(stringBuffer(POJO_CHILD)), null, null, Collections.emptyMap());
+				.decode(stringBuffer(POJO_CHILD), null, null, Collections.emptyMap());
 		Flux<List<XMLEvent>> result = this.decoder.split(xmlEvents, new QName("pojo"));
 
 
@@ -157,7 +158,7 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeSingleXmlRootElement() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_ROOT));
+		Mono<DataBuffer> source = stringBuffer(POJO_ROOT);
 		Mono<Object> output = this.decoder.decodeToMono(source, ResolvableType.forClass(Pojo.class),
 				null, Collections.emptyMap());
 
@@ -169,7 +170,7 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeSingleXmlTypeElement() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_ROOT));
+		Mono<DataBuffer> source = stringBuffer(POJO_ROOT);
 		Mono<Object> output = this.decoder.decodeToMono(source, ResolvableType.forClass(TypePojo.class),
 				null, Collections.emptyMap());
 
@@ -181,7 +182,7 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeMultipleXmlRootElement() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_CHILD));
+		Mono<DataBuffer> source = stringBuffer(POJO_CHILD);
 		Flux<Object> output = this.decoder.decode(source, ResolvableType.forClass(Pojo.class),
 				null, Collections.emptyMap());
 
@@ -194,7 +195,7 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeMultipleXmlTypeElement() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_CHILD));
+		Mono<DataBuffer> source = stringBuffer(POJO_CHILD);
 		Flux<Object> output = this.decoder.decode(source, ResolvableType.forClass(TypePojo.class),
 				null, Collections.emptyMap());
 
@@ -207,8 +208,9 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeError() throws Exception {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("<pojo>"))
+		Flux<DataBuffer> source = Flux.concat(
-				.concatWith(Flux.error(new RuntimeException()));
+				stringBuffer("<pojo>"),
+				Flux.error(new RuntimeException()));
 
 		Mono<Object> output = this.decoder.decodeToMono(source, ResolvableType.forClass(Pojo.class),
 				null, Collections.emptyMap());
@@ -239,6 +241,16 @@ public class Jaxb2XmlDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	}
 
+	private Mono<DataBuffer> stringBuffer(String value) {
+		return Mono.defer(() -> {
+			byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+			DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+			buffer.write(bytes);
+			return Mono.just(buffer);
+		});
+	}
+
+
 	@javax.xml.bind.annotation.XmlType(name = "pojo")
 	public static class TypePojo {
 

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

@@ -16,14 +16,16 @@
 
 package org.springframework.http.codec.xml;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import javax.xml.stream.events.XMLEvent;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 
 import static org.junit.Assert.*;
@@ -31,7 +33,7 @@ import static org.junit.Assert.*;
 /**
  * @author Arjen Poutsma
  */
-public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
+public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 
 	private static final String XML = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
 			"<pojo>" +
@@ -45,7 +47,7 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 	public void toXMLEventsAalto() {
 
 		Flux<XMLEvent> events =
-				this.decoder.decode(Flux.just(stringBuffer(XML)), null, null, Collections.emptyMap());
+				this.decoder.decode(stringBuffer(XML), null, null, Collections.emptyMap());
 
 		StepVerifier.create(events)
 				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
@@ -66,7 +68,7 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		decoder.useAalto = false;
 
 		Flux<XMLEvent> events =
-				this.decoder.decode(Flux.just(stringBuffer(XML)), null, null, Collections.emptyMap());
+				this.decoder.decode(stringBuffer(XML), null, null, Collections.emptyMap());
 
 		StepVerifier.create(events)
 				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
@@ -85,8 +87,9 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void decodeErrorAalto() {
-		Flux<DataBuffer> source = Flux.just(stringBuffer("<pojo>"))
+		Flux<DataBuffer> source = Flux.concat(
-				.concatWith(Flux.error(new RuntimeException()));
+				stringBuffer("<pojo>"),
+				Flux.error(new RuntimeException()));
 
 		Flux<XMLEvent> events =
 				this.decoder.decode(source, null, null, Collections.emptyMap());
@@ -102,8 +105,9 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 	public void decodeErrorNonAalto() {
 		decoder.useAalto = false;
 
-		Flux<DataBuffer> source = Flux.just(stringBuffer("<pojo>"))
+		Flux<DataBuffer> source = Flux.concat(
-				.concatWith(Flux.error(new RuntimeException()));
+				stringBuffer("<pojo>"),
+				Flux.error(new RuntimeException()));
 
 		Flux<XMLEvent> events =
 				this.decoder.decode(source, null, null, Collections.emptyMap());
@@ -128,4 +132,13 @@ public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 		assertEquals(expectedData, event.asCharacters().getData());
 	}
 
+	private Mono<DataBuffer> stringBuffer(String value) {
+		return Mono.defer(() -> {
+			byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+			DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+			buffer.write(bytes);
+			return Mono.just(buffer);
+		});
+	}
+
 }