diff --git a/spring-core/src/test/java/org/springframework/core/codec/AbstractEncoderTestCase.java b/spring-core/src/test/java/org/springframework/core/codec/AbstractEncoderTestCase.java
index 94d42cc2dc..9ae9ee1406 100644
--- a/spring-core/src/test/java/org/springframework/core/codec/AbstractEncoderTestCase.java
+++ b/spring-core/src/test/java/org/springframework/core/codec/AbstractEncoderTestCase.java
@@ -18,7 +18,6 @@ package org.springframework.core.codec;
 
 import java.util.Map;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import org.junit.Test;
 import org.reactivestreams.Publisher;
@@ -28,7 +27,6 @@ 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.lang.Nullable;
 import org.springframework.util.Assert;
@@ -36,193 +34,245 @@ import org.springframework.util.MimeType;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.*;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 
 /**
  * Abstract base class for {@link Encoder} unit tests. Subclasses need to implement
- * {@link #input()} and {@link #outputConsumers()}, from which {@link #encode()},
- * {@link #encodeError()} and {@link #encodeCancel()} are run.
+ * {@link #canEncode()} and {@link #encode()}, possibly using the wide
+ *  * variety of helper methods like {@link #testEncodeAll}.
  *
  * @author Arjen Poutsma
+ * @since 5.1.3
  */
 @SuppressWarnings("ProtectedField")
-public abstract class AbstractEncoderTestCase<T, E extends Encoder<T>> extends
-		AbstractLeakCheckingTestCase {
+public abstract class AbstractEncoderTestCase<E extends Encoder<?>>
+		extends	AbstractLeakCheckingTestCase {
 
 	/**
 	 * The encoder to test.
 	 */
 	protected final E encoder;
 
+
 	/**
-	 * The type used for
-	 * {@link Encoder#encode(Publisher, DataBufferFactory, ResolvableType, MimeType, Map)}.
+	 * Construct a new {@code AbstractEncoderTestCase} for the given parameters.
+	 * @param encoder the encoder
 	 */
-	protected final ResolvableType elementType;
+	protected AbstractEncoderTestCase(E encoder) {
+
+		Assert.notNull(encoder, "Encoder must not be null");
+
+		this.encoder = encoder;
+	}
+
 
 	/**
-	 * The mime type used for
-	 * {@link Encoder#encode(Publisher, DataBufferFactory, ResolvableType, MimeType, Map)}.
-	 * May be {@code null}.
+	 * Subclasses should implement this method to test {@link Encoder#canEncode}.
 	 */
-	@Nullable
-	protected final MimeType mimeType;
+	@Test
+	public abstract void canEncode() throws Exception;
 
 	/**
-	 * The hints used for
-	 * {@link Encoder#encode(Publisher, DataBufferFactory, ResolvableType, MimeType, Map)}.
-	 * May be {@code null}.
+	 * Subclasses should implement this method to test {@link Encoder#encode}, possibly using
+	 * {@link #testEncodeAll} or other helper methods.
 	 */
-	@Nullable
-	protected final Map<String, Object> hints;
+	@Test
+	public abstract void encode() throws Exception;
 
 
 	/**
-	 * Construct a new {@code AbstractEncoderTestCase} for the given encoder and element class.
-	 * @param encoder the encoder
-	 * @param elementClass the element class
+	 * Helper methods that tests for a variety of encoding scenarios. This methods
+	 * invokes:
+	 * <ul>
+	 *     <li>{@link #testEncode(Publisher, ResolvableType, Consumer, MimeType, Map)}</li>
+	 *     <li>{@link #testEncodeError(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testEncodeCancel(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testEncodeEmpty(ResolvableType, MimeType, Map)}</li>
+	 * </ul>
+	 *
+	 * @param input the input to be provided to the encoder
+	 * @param inputClass the input class
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param <T> the output type
 	 */
-	protected AbstractEncoderTestCase(E encoder, Class<?> elementClass) {
-		this(encoder, ResolvableType.forClass(elementClass), null, null);
+	protected <T> void testEncodeAll(Publisher<? extends T> input, Class<? extends T> inputClass,
+			Consumer<StepVerifier.FirstStep<DataBuffer>> stepConsumer) {
+		testEncodeAll(input, ResolvableType.forClass(inputClass), stepConsumer, null, null);
 	}
 
 	/**
-	 * Construct a new {@code AbstractEncoderTestCase} for the given parameters.
-	 * @param encoder the encoder
-	 * @param elementType the element type
-	 * @param mimeType the mime type. May be {@code null}.
-	 * @param hints the hints. May be {@code null}.
+	 * Helper methods that tests for a variety of decoding scenarios. This methods
+	 * invokes:
+	 * <ul>
+	 *     <li>{@link #testEncode(Publisher, ResolvableType, Consumer, MimeType, Map)}</li>
+	 *     <li>{@link #testEncodeError(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testEncodeCancel(Publisher, ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testEncodeEmpty(ResolvableType, MimeType, Map)}</li>
+	 * </ul>
+	 *
+	 * @param input the input to be provided to the encoder
+	 * @param inputType the input 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 AbstractEncoderTestCase(E encoder, ResolvableType elementType,
+	protected <T> void testEncodeAll(Publisher<? extends T> input, ResolvableType inputType,
+			Consumer<StepVerifier.FirstStep<DataBuffer>> stepConsumer,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+		testEncode(input, inputType, stepConsumer, mimeType, hints);
+		testEncodeError(input, inputType, mimeType, hints);
+		testEncodeCancel(input, inputType, mimeType, hints);
+		testEncodeEmpty(inputType, mimeType, hints);
+	}
 
-		Assert.notNull(encoder, "Encoder must not be null");
-		Assert.notNull(elementType, "ElementType must not be null");
+	/**
+	 * Test a standard {@link Encoder#encode encode} scenario.
+	 *
+	 * @param input the input to be provided to the encoder
+	 * @param inputClass the input class
+	 * @param stepConsumer a consumer to {@linkplain StepVerifier verify} the output
+	 * @param <T> the output type
+	 */
+	protected <T> void testEncode(Publisher<? extends T> input, Class<? extends T> inputClass,
+			Consumer<StepVerifier.FirstStep<DataBuffer>> stepConsumer) {
+		testEncode(input, ResolvableType.forClass(inputClass), stepConsumer, null, null);
+	}
 
-		this.encoder = encoder;
-		this.elementType = elementType;
-		this.mimeType = mimeType;
-		this.hints = hints;
+	/**
+	 * Test a standard {@link Encoder#encode encode} scenario.
+	 *
+	 * @param input the input to be provided to the encoder
+	 * @param inputType the input 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 testEncode(Publisher<? extends T> input, ResolvableType inputType,
+			Consumer<StepVerifier.FirstStep<DataBuffer>> stepConsumer,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Flux<DataBuffer> result = encoder().encode(input, this.bufferFactory, inputType,
+				mimeType, hints);
+		StepVerifier.FirstStep<DataBuffer> step = StepVerifier.create(result);
+		stepConsumer.accept(step);
 	}
 
 	/**
-	 * Abstract template method that provides input for the encoder.
-	 * Used for {@link #encode()}, {@link #encodeError()}, and {@link #encodeCancel()}.
+	 * Test a {@link Encoder#encode encode} scenario where the input stream contains an error.
+	 * This test method will feed the first element of the {@code input} stream to the encoder,
+	 * 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 encoder
+	 * @param inputType the input 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 abstract Flux<T> input();
+	protected void testEncodeError(Publisher<?> input, ResolvableType inputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		input = Flux.concat(
+				Flux.from(input).take(1),
+				Flux.error(new InputException()));
+
+		Flux<DataBuffer> result = encoder().encode(input, this.bufferFactory, inputType,
+				mimeType, hints);
+
+		StepVerifier.create(result)
+				.consumeNextWith(DataBufferUtils::release)
+				.expectError(InputException.class)
+				.verify();
+	}
 
 	/**
-	 * Abstract template method that verifies the output of the encoder.
-	 * The returned stream should contain a buffer consumer for each expected output, given
-	 * the {@linkplain #input()}.
+	 * Test a {@link Encoder#encode encode} 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 encoder
+	 * @param inputType the input 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 abstract Stream<Consumer<DataBuffer>> outputConsumers();
+	protected void testEncodeCancel(Publisher<?> input, ResolvableType inputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-	private Stream<Consumer<DataBuffer>> outputAndReleaseConsumers() {
-		return outputConsumers()
-				.map(consumer -> consumer.andThen(DataBufferUtils::release));
+		Flux<DataBuffer> result = encoder().encode(input, this.bufferFactory, inputType, mimeType,
+				hints);
+
+		StepVerifier.create(result)
+				.consumeNextWith(DataBufferUtils::release)
+				.thenCancel()
+				.verify();
 	}
 
 	/**
-	 * Create a result consumer that expects the given String in UTF-8 encoding.
-	 * @param expected the expected string
-	 * @return a consumer that expects the given data buffer to be equal to {@code expected}
+	 * Test a {@link Encoder#encode encode} scenario where the input stream is empty.
+	 * The output is expected to be empty as well.
+	 *
+	 * @param inputType the input 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 final Consumer<DataBuffer> resultConsumer(String expected) {
-		return dataBuffer -> {
-			byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
-			dataBuffer.read(resultBytes);
-			String actual = new String(resultBytes, UTF_8);
-			assertEquals(expected, actual);
-		};
+	protected void testEncodeEmpty(ResolvableType inputType, @Nullable MimeType mimeType,
+			@Nullable Map<String, Object> hints) {
 
+		Flux<?> input = Flux.empty();
+		Flux<DataBuffer> result = encoder().encode(input, this.bufferFactory, inputType,
+				mimeType, hints);
+
+		StepVerifier.create(result)
+				.verifyComplete();
 	}
 
 	/**
 	 * Create a result consumer that expects the given bytes.
-	 * @param expected the expected string
+	 * @param expected the expected bytes
 	 * @return a consumer that expects the given data buffer to be equal to {@code expected}
 	 */
-	protected final Consumer<DataBuffer> resultConsumer(byte[] expected) {
+	protected final Consumer<DataBuffer> expectBytes(byte[] expected) {
 		return dataBuffer -> {
 			byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
 			dataBuffer.read(resultBytes);
+			release(dataBuffer);
 			assertArrayEquals(expected, resultBytes);
 		};
 	}
 
 	/**
-	 * Tests whether passing {@link #input()} to the encoder can be consumed with
-	 * {@link #outputConsumers()}.
+	 * Create a result consumer that expects the given string, using the UTF-8 encoding.
+	 * @param expected the expected string
+	 * @return a consumer that expects the given data buffer to be equal to {@code expected}
 	 */
-	@Test
-	public final void encode() {
-		Flux<T> input = input();
-
-		Flux<DataBuffer> output = this.encoder.encode(input, this.bufferFactory,
-				this.elementType, this.mimeType, this.hints);
-
-		StepVerifier.Step<DataBuffer> step = StepVerifier.create(output);
-
-		outputAndReleaseConsumers().forEach(step::consumeNextWith);
+	protected Consumer<DataBuffer> expectString(String expected) {
+		return dataBuffer -> {
+			byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
+			dataBuffer.read(resultBytes);
+			release(dataBuffer);
+			String actual = new String(resultBytes, UTF_8);
+			assertEquals(expected, actual);
+		};
 
-		step.expectComplete()
-				.verify();
 	}
 
-	/**
-	 * Tests whether passing an error to the encoder can be consumed with
-	 * {@link #outputConsumers()}.
-	 */
-	@Test
-	public final void encodeError() {
-
-		boolean singleValue = this.encoder instanceof AbstractSingleValueEncoder;
-
-		Flux<T> input;
-		if (singleValue) {
-			input = Flux.error(new RuntimeException());
-		}
-		else {
-			input = Flux.concat(
-					input().take(1),
-					Flux.error(new RuntimeException()));
-		}
-
-		Flux<DataBuffer> output = this.encoder.encode(input, this.bufferFactory,
-				this.elementType, this.mimeType, this.hints);
-
-		if (singleValue) {
-			StepVerifier.create(output)
-					.expectError(RuntimeException.class)
-					.verify();
-		}
-		else {
-			Consumer<DataBuffer> firstResultConsumer = outputAndReleaseConsumers().findFirst()
-					.orElseThrow(IllegalArgumentException::new);
-			StepVerifier.create(output)
-					.consumeNextWith(firstResultConsumer)
-					.expectError(RuntimeException.class)
-					.verify();
-		}
+	@SuppressWarnings("unchecked")
+	private <T> Encoder<T> encoder() {
+		return (Encoder<T>) this.encoder;
+
 	}
 
 	/**
-	 * Tests whether canceling the output of the encoder can be consumed with
-	 * {@link #outputConsumers()}.
+	 * Exception used in {@link #testEncodeError}.
 	 */
-	@Test
-	public final void encodeCancel() {
-		Flux<T> input = input();
+	@SuppressWarnings("serial")
+	public static class InputException extends RuntimeException {
 
-		Flux<DataBuffer> output = this.encoder.encode(input, this.bufferFactory,
-				this.elementType, this.mimeType, this.hints);
-
-		Consumer<DataBuffer> firstResultConsumer = outputAndReleaseConsumers().findFirst()
-				.orElseThrow(IllegalArgumentException::new);
-		StepVerifier.create(output)
-				.consumeNextWith(firstResultConsumer)
-				.thenCancel()
-				.verify();
 	}
 
+
 }
diff --git a/spring-core/src/test/java/org/springframework/core/codec/ByteArrayEncoderTests.java b/spring-core/src/test/java/org/springframework/core/codec/ByteArrayEncoderTests.java
index c48c71eb88..833fbc2c3a 100644
--- a/spring-core/src/test/java/org/springframework/core/codec/ByteArrayEncoderTests.java
+++ b/spring-core/src/test/java/org/springframework/core/codec/ByteArrayEncoderTests.java
@@ -17,14 +17,11 @@
 package org.springframework.core.codec;
 
 import java.nio.charset.StandardCharsets;
-import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.junit.Assert.*;
@@ -32,30 +29,18 @@ import static org.junit.Assert.*;
 /**
  * @author Arjen Poutsma
  */
-public class ByteArrayEncoderTests extends AbstractEncoderTestCase<byte[], ByteArrayEncoder> {
+public class ByteArrayEncoderTests extends AbstractEncoderTestCase<ByteArrayEncoder> {
 
 	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
 	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
 
 	public ByteArrayEncoderTests() {
-		super(new ByteArrayEncoder(), byte[].class);
+		super(new ByteArrayEncoder());
 	}
 
-	@Override
-	protected Flux<byte[]> input() {
-		return Flux.just(this.fooBytes,
-				this.barBytes);
-	}
 
 	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.fooBytes))
-				.add(resultConsumer(this.barBytes))
-				.build();
-	}
-
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(byte[].class),
@@ -69,4 +54,13 @@ public class ByteArrayEncoderTests extends AbstractEncoderTestCase<byte[], ByteA
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
+	@Override
+	public void encode() {
+		Flux<byte[]> input = Flux.just(this.fooBytes, this.barBytes);
+
+		testEncodeAll(input, byte[].class, step -> step
+				.consumeNextWith(expectBytes(this.fooBytes))
+				.consumeNextWith(expectBytes(this.barBytes))
+				.verifyComplete());
+	}
 }
diff --git a/spring-core/src/test/java/org/springframework/core/codec/ByteBufferEncoderTests.java b/spring-core/src/test/java/org/springframework/core/codec/ByteBufferEncoderTests.java
index cf21cb4657..22e82a0b71 100644
--- a/spring-core/src/test/java/org/springframework/core/codec/ByteBufferEncoderTests.java
+++ b/spring-core/src/test/java/org/springframework/core/codec/ByteBufferEncoderTests.java
@@ -18,14 +18,11 @@ package org.springframework.core.codec;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.junit.Assert.*;
@@ -33,30 +30,17 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class ByteBufferEncoderTests extends AbstractEncoderTestCase<ByteBuffer, ByteBufferEncoder> {
+public class ByteBufferEncoderTests extends AbstractEncoderTestCase<ByteBufferEncoder> {
 
 	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
 	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
 
 	public ByteBufferEncoderTests() {
-		super(new ByteBufferEncoder(), ByteBuffer.class);
+		super(new ByteBufferEncoder());
 	}
 
 	@Override
-	protected Flux<ByteBuffer> input() {
-		return Flux.just(this.fooBytes, this.barBytes)
-				.map(ByteBuffer::wrap);
-	}
-
-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.fooBytes))
-				.add(resultConsumer(this.barBytes))
-				.build();
-	}
-
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(ByteBuffer.class),
@@ -70,5 +54,17 @@ public class ByteBufferEncoderTests extends AbstractEncoderTestCase<ByteBuffer,
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
+	@Override
+	@Test
+	public void encode() {
+		Flux<ByteBuffer> input = Flux.just(this.fooBytes, this.barBytes)
+				.map(ByteBuffer::wrap);
 
+		testEncodeAll(input, ByteBuffer.class, step -> step
+				.consumeNextWith(expectBytes(this.fooBytes))
+				.consumeNextWith(expectBytes(this.barBytes))
+				.verifyComplete());
+
+
+	}
 }
diff --git a/spring-core/src/test/java/org/springframework/core/codec/CharSequenceEncoderTests.java b/spring-core/src/test/java/org/springframework/core/codec/CharSequenceEncoderTests.java
index 312fdf87d9..636b45bfc1 100644
--- a/spring-core/src/test/java/org/springframework/core/codec/CharSequenceEncoderTests.java
+++ b/spring-core/src/test/java/org/springframework/core/codec/CharSequenceEncoderTests.java
@@ -16,14 +16,9 @@
 
 package org.springframework.core.codec;
 
-import java.util.function.Consumer;
-import java.util.stream.Stream;
-
-import org.junit.Test;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.junit.Assert.*;
@@ -32,31 +27,19 @@ import static org.junit.Assert.*;
  * @author Sebastien Deleuze
  */
 public class CharSequenceEncoderTests
-		extends AbstractEncoderTestCase<CharSequence, CharSequenceEncoder> {
+		extends AbstractEncoderTestCase<CharSequenceEncoder> {
 
 	private final String foo = "foo";
 
 	private final String bar = "bar";
 
 	public CharSequenceEncoderTests() {
-		super(CharSequenceEncoder.textPlainOnly(), CharSequence.class);
+		super(CharSequenceEncoder.textPlainOnly());
 	}
 
-	@Override
-	protected Flux<CharSequence> input() {
-		return Flux.just(this.foo, this.bar);
-	}
 
 	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.foo))
-				.add(resultConsumer(this.bar))
-				.build();
-	}
-
-	@Test
-	public void canWrite() {
+	public void canEncode() throws Exception {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(String.class),
 				MimeTypeUtils.TEXT_PLAIN));
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(StringBuilder.class),
@@ -71,4 +54,17 @@ public class CharSequenceEncoderTests
 		// SPR-15464
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
+
+	@Override
+	public void encode() {
+		Flux<CharSequence> input = Flux.just(this.foo, this.bar);
+
+		testEncodeAll(input, CharSequence.class, step -> step
+				.consumeNextWith(expectString(this.foo))
+				.consumeNextWith(expectString(this.bar))
+				.verifyComplete());
+
+
+	}
+
 }
diff --git a/spring-core/src/test/java/org/springframework/core/codec/DataBufferEncoderTests.java b/spring-core/src/test/java/org/springframework/core/codec/DataBufferEncoderTests.java
index d378ce1b8f..7ff4449f3a 100644
--- a/spring-core/src/test/java/org/springframework/core/codec/DataBufferEncoderTests.java
+++ b/spring-core/src/test/java/org/springframework/core/codec/DataBufferEncoderTests.java
@@ -17,11 +17,10 @@
 package org.springframework.core.codec;
 
 import java.nio.charset.StandardCharsets;
-import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
@@ -32,36 +31,18 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class DataBufferEncoderTests extends AbstractEncoderTestCase<DataBuffer, DataBufferEncoder> {
+public class DataBufferEncoderTests extends AbstractEncoderTestCase<DataBufferEncoder> {
 
 	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
 	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
 
 	public DataBufferEncoderTests() {
-		super(new DataBufferEncoder(), DataBuffer.class);
+		super(new DataBufferEncoder());
 	}
 
-	@Override
-	protected Flux<DataBuffer> input() {
-//		DefaultDataBufferFactory bufferFactory = new DefaultDataBufferFactory();
-		return Flux.just(this.fooBytes, this.barBytes)
-				.map(bytes -> {
-					DataBuffer dataBuffer = bufferFactory.allocateBuffer(bytes.length);
-					dataBuffer.write(bytes);
-					return dataBuffer;
-				});
-	}
 
 	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.fooBytes))
-				.add(resultConsumer(this.barBytes))
-				.build();
-
-	}
-
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(DataBuffer.class),
@@ -75,4 +56,21 @@ public class DataBufferEncoderTests extends AbstractEncoderTestCase<DataBuffer,
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
+	@Override
+	public void encode() throws Exception {
+		Flux<DataBuffer> input = Flux.just(this.fooBytes, this.barBytes)
+				.flatMap(bytes -> Mono.defer(() -> {
+					DataBuffer dataBuffer = this.bufferFactory.allocateBuffer(bytes.length);
+					dataBuffer.write(bytes);
+					return Mono.just(dataBuffer);
+				}));
+
+		testEncodeAll(input, DataBuffer.class, step -> step
+				.consumeNextWith(expectBytes(this.fooBytes))
+				.consumeNextWith(expectBytes(this.barBytes))
+				.verifyComplete());
+
+	}
+
+
 }
diff --git a/spring-core/src/test/java/org/springframework/core/codec/ResourceEncoderTests.java b/spring-core/src/test/java/org/springframework/core/codec/ResourceEncoderTests.java
index 81cffb1b7c..953e48cd6c 100644
--- a/spring-core/src/test/java/org/springframework/core/codec/ResourceEncoderTests.java
+++ b/spring-core/src/test/java/org/springframework/core/codec/ResourceEncoderTests.java
@@ -16,17 +16,20 @@
 
 package org.springframework.core.codec;
 
-import java.util.function.Consumer;
-import java.util.stream.Stream;
+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.DataBuffer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -35,27 +38,16 @@ import static org.junit.Assert.*;
 /**
  * @author Arjen Poutsma
  */
-public class ResourceEncoderTests extends AbstractEncoderTestCase<Resource, ResourceEncoder> {
+public class ResourceEncoderTests extends AbstractEncoderTestCase<ResourceEncoder> {
 
 	private final byte[] bytes = "foo".getBytes(UTF_8);
 
-	public ResourceEncoderTests() {
-		super(new ResourceEncoder(), Resource.class);
-	}
 
-	@Override
-	protected Flux<Resource> input() {
-		return Flux.just(new ByteArrayResource(this.bytes));
+	public ResourceEncoderTests() {
+		super(new ResourceEncoder());
 	}
 
 	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.bytes))
-				.build();
-	}
-
-
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(InputStreamResource.class),
@@ -71,4 +63,28 @@ public class ResourceEncoderTests extends AbstractEncoderTestCase<Resource, Reso
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
+	@Override
+	public void encode() {
+		Flux<Resource> input = Flux.just(new ByteArrayResource(this.bytes));
+
+		testEncodeAll(input, Resource.class, step -> step
+				.consumeNextWith(expectBytes(this.bytes))
+				.verifyComplete());
+	}
+
+	@Override
+	protected void testEncodeError(Publisher<?> input, ResolvableType outputType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Flux<Resource> i = Flux.error(new InputException());
+
+		Flux<DataBuffer> result = ((Encoder<Resource>) this.encoder).encode(i,
+				this.bufferFactory, outputType,
+				mimeType, hints);
+
+		StepVerifier.create(result)
+				.expectError(InputException.class)
+				.verify();
+	}
+
 }
diff --git a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
index 9c1568799e..3eda72a113 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2JsonEncoderTests.java
@@ -20,8 +20,6 @@ import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
-import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.annotation.JsonTypeName;
@@ -29,18 +27,15 @@ 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.AbstractEncoderTestCase;
-import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
 import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.util.MimeType;
 
-import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.junit.Assert.*;
 import static org.springframework.http.MediaType.APPLICATION_JSON;
@@ -55,30 +50,14 @@ import static org.springframework.http.codec.json.JacksonViewBean.MyJacksonView3
 /**
  * @author Sebastien Deleuze
  */
-public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Jackson2JsonEncoder> {
+public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Jackson2JsonEncoder> {
 
 
 	public Jackson2JsonEncoderTests() {
-		super(new Jackson2JsonEncoder(), ResolvableType.forClass(Pojo.class),
-				APPLICATION_STREAM_JSON, null);
-	}
-
-	@Override
-	protected Flux<Object> input() {
-		return Flux.just(new Pojo("foo", "bar"),
-				new Pojo("foofoo", "barbar"),
-				new Pojo("foofoofoo", "barbarbar"));
+		super(new Jackson2JsonEncoder());
 	}
 
 	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
-				.add(resultConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
-				.add(resultConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
-				.build();
-	}
-
 	@Test
 	public void canEncode() {
 		ResolvableType pojoType = ResolvableType.forClass(Pojo.class);
@@ -94,6 +73,22 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		assertFalse(this.encoder.canEncode(ResolvableType.forClass(Object.class), APPLICATION_OCTET_STREAM));
 	}
 
+	@Override
+	public void encode() throws Exception {
+		Flux<Object> input = Flux.just(new Pojo("foo", "bar"),
+				new Pojo("foofoo", "barbar"),
+				new Pojo("foofoofoo", "barbarbar"));
+
+		testEncodeAll(input, ResolvableType.forClass(Pojo.class), step -> step
+				.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
+				.consumeNextWith(expectString("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
+				.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
+				.verifyComplete(),
+				APPLICATION_STREAM_JSON, null);
+
+
+	}
+
 	@Test // SPR-15866
 	public void canEncodeWithCustomMimeType() {
 		MimeType textJavascript = new MimeType("text", "javascript", StandardCharsets.UTF_8);
@@ -121,33 +116,29 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 
 	@Test
 	public void encodeNonStream() {
-		Flux<Pojo> source = Flux.just(
+		Flux<Pojo> input = Flux.just(
 				new Pojo("foo", "bar"),
 				new Pojo("foofoo", "barbar"),
 				new Pojo("foofoofoo", "barbarbar")
 		);
-		ResolvableType type = ResolvableType.forClass(Pojo.class);
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null, emptyMap());
 
-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("[" +
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(expectString("[" +
 						"{\"foo\":\"foo\",\"bar\":\"bar\"}," +
 						"{\"foo\":\"foofoo\",\"bar\":\"barbar\"}," +
 						"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}]")
 						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+				.verifyComplete());
 	}
 
 	@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.bufferFactory, type, null, emptyMap());
+		Flux<ParentClass> input = Flux.just(new Foo(), new Bar());
 
-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("[{\"type\":\"foo\"},{\"type\":\"bar\"}]")
+		testEncode(input, ParentClass.class, step -> step
+				.consumeNextWith(expectString("[{\"type\":\"foo\"},{\"type\":\"bar\"}]")
 						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+				.verifyComplete());
 	}
 
 
@@ -156,22 +147,21 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		MediaType fooMediaType = new MediaType("application", "foo");
 		MediaType barMediaType = new MediaType("application", "bar");
 		this.encoder.setStreamingMediaTypes(Arrays.asList(fooMediaType, barMediaType));
-		Flux<Pojo> source = Flux.just(
+		Flux<Pojo> input = Flux.just(
 				new Pojo("foo", "bar"),
 				new Pojo("foofoo", "barbar"),
 				new Pojo("foofoofoo", "barbarbar")
 		);
-		ResolvableType type = ResolvableType.forClass(Pojo.class);
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, barMediaType, emptyMap());
 
-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n")
+		testEncode(input, ResolvableType.forClass(Pojo.class), step -> step
+				.consumeNextWith(expectString("{\"foo\":\"foo\",\"bar\":\"bar\"}\n")
 						.andThen(DataBufferUtils::release))
-				.consumeNextWith(resultConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n")
+				.consumeNextWith(expectString("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n")
 						.andThen(DataBufferUtils::release))
-				.consumeNextWith(resultConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n")
+				.consumeNextWith(expectString("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n")
 						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+				.verifyComplete(),
+				barMediaType, null);
 	}
 
 	@Test
@@ -180,15 +170,16 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		bean.setWithView1("with");
 		bean.setWithView2("with");
 		bean.setWithoutView("without");
+		Mono<JacksonViewBean> input = Mono.just(bean);
 
 		ResolvableType type = ResolvableType.forClass(JacksonViewBean.class);
 		Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView1.class);
-		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);
 
-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("{\"withView1\":\"with\"}")
-						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+		testEncode(input, type, step -> step
+						.consumeNextWith(expectString("{\"withView1\":\"with\"}")
+								.andThen(DataBufferUtils::release))
+						.verifyComplete(),
+				null, hints);
 	}
 
 	@Test
@@ -197,17 +188,19 @@ public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Ja
 		bean.setWithView1("with");
 		bean.setWithView2("with");
 		bean.setWithoutView("without");
+		Mono<JacksonViewBean> input = Mono.just(bean);
 
 		ResolvableType type = ResolvableType.forClass(JacksonViewBean.class);
 		Map<String, Object> hints = singletonMap(JSON_VIEW_HINT, MyJacksonView3.class);
-		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);
 
-		StepVerifier.create(output)
-				.consumeNextWith(resultConsumer("{\"withoutView\":\"without\"}")
-						.andThen(DataBufferUtils::release))
-				.verifyComplete();
+		testEncode(input, type, step -> step
+						.consumeNextWith(expectString("{\"withoutView\":\"without\"}")
+								.andThen(DataBufferUtils::release))
+						.verifyComplete(),
+				null, hints);
 	}
 
+
 	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 	private static class ParentClass {
 	}
diff --git a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2SmileEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2SmileEncoderTests.java
index c6f106a1b4..a3d7caa8e7 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2SmileEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/json/Jackson2SmileEncoderTests.java
@@ -18,20 +18,18 @@ package org.springframework.http.codec.json;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.type.CollectionType;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
-import reactor.test.StepVerifier;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractEncoderTestCase;
 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.codec.Pojo;
 import org.springframework.http.codec.ServerSentEvent;
@@ -39,6 +37,7 @@ import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 import org.springframework.util.MimeType;
 
 import static org.junit.Assert.*;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 import static org.springframework.http.MediaType.APPLICATION_XML;
 
 /**
@@ -46,7 +45,7 @@ import static org.springframework.http.MediaType.APPLICATION_XML;
  *
  * @author Sebastien Deleuze
  */
-public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, Jackson2SmileEncoder> {
+public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Jackson2SmileEncoder> {
 
 	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");
@@ -55,30 +54,9 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J
 
 	private final ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
 
-	private Pojo pojo1 = new Pojo("foo", "bar");
-
-	private Pojo pojo2 = new Pojo("foofoo", "barbar");
-
-	private Pojo pojo3 = new Pojo("foofoofoo", "barbarbar");
-
 	public Jackson2SmileEncoderTests() {
-		super(new Jackson2SmileEncoder(), ResolvableType.forClass(Pojo.class),
-				STREAM_SMILE_MIME_TYPE, null);
-
-	}
-
-	@Override
-	protected Flux<Object> input() {
-		return Flux.just(this.pojo1, this.pojo2, this.pojo3);
-	}
+		super(new Jackson2SmileEncoder());
 
-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(pojoConsumer(this.pojo1))
-				.add(pojoConsumer(this.pojo2))
-				.add(pojoConsumer(this.pojo3))
-				.build();
 	}
 
 	public Consumer<DataBuffer> pojoConsumer(Pojo expected) {
@@ -87,7 +65,7 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J
 				Pojo actual = this.mapper.reader().forType(Pojo.class)
 						.readValue(DataBufferTestUtils.dumpBytes(dataBuffer));
 				assertEquals(expected, actual);
-				DataBufferUtils.release(dataBuffer);
+				release(dataBuffer);
 			}
 			catch (IOException ex) {
 				throw new UncheckedIOException(ex);
@@ -96,6 +74,7 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J
 	}
 
 
+	@Override
 	@Test
 	public void canEncode() {
 		ResolvableType pojoType = ResolvableType.forClass(Pojo.class);
@@ -116,32 +95,62 @@ public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, J
 		assertFalse(this.encoder.canEncode(sseType, SMILE_MIME_TYPE));
 	}
 
+	@Override
+	@Test
+	public void encode() {
+		List<Pojo> list = Arrays.asList(
+				new Pojo("foo", "bar"),
+				new Pojo("foofoo", "barbar"),
+				new Pojo("foofoofoo", "barbarbar"));
+
+		Flux<Pojo> input = Flux.fromIterable(list);
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(expect(list, List.class)));
+	}
+
 	@Test
-	public void encodeNonStream() {
-		Flux<DataBuffer> output = this.encoder.encode(input(), this.bufferFactory, elementType,
-				null, null);
-
-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> {
-					try {
-						CollectionType type = mapper.getTypeFactory()
-								.constructCollectionType(List.class, Pojo.class);
-						List<Pojo> value = mapper.reader().forType(type)
-								.readValue(dataBuffer.asInputStream());
-						assertEquals(3, value.size());
-						assertEquals(pojo1, value.get(0));
-						assertEquals(pojo2, value.get(1));
-						assertEquals(pojo3, value.get(2));
-					}
-					catch (IOException ex) {
-						throw new UncheckedIOException(ex);
-					}
-					finally {
-						DataBufferUtils.release(dataBuffer);
-					}
-				})
-				.verifyComplete();
+	public void encodeError() throws Exception {
+		Mono<Pojo> input = Mono.error(new InputException());
+
+		testEncode(input, Pojo.class, step -> step
+				.expectError(InputException.class)
+				.verify());
+
+	}
+
+	@Test
+	public void encodeAsStream() throws Exception {
+		Pojo pojo1 = new Pojo("foo", "bar");
+		Pojo pojo2 = new Pojo("foofoo", "barbar");
+		Pojo pojo3 = new Pojo("foofoofoo", "barbarbar");
+		Flux<Pojo> input = Flux.just(pojo1, pojo2, pojo3);
+		ResolvableType type = ResolvableType.forClass(Pojo.class);
+
+		testEncodeAll(input, type, step -> step
+				.consumeNextWith(expect(pojo1, Pojo.class))
+				.consumeNextWith(expect(pojo2, Pojo.class))
+				.consumeNextWith(expect(pojo3, Pojo.class))
+				.verifyComplete(),
+		STREAM_SMILE_MIME_TYPE, null);
+	}
+
+
+	private <T> Consumer<DataBuffer> expect(T expected, Class<T> expectedType) {
+		return dataBuffer -> {
+			try {
+				Object actual = this.mapper.reader().forType(expectedType)
+						.readValue(dataBuffer.asInputStream());
+				assertEquals(expected, actual);
+			}
+			catch (IOException e) {
+				throw new UncheckedIOException(e);
+			}
+			finally {
+				release(dataBuffer);
+			}
+		};
+
 	}
 
 
diff --git a/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufEncoderTests.java
index c358f2d9ba..15f8d5f7da 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/protobuf/ProtobufEncoderTests.java
@@ -19,14 +19,15 @@ package org.springframework.http.codec.protobuf;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
 
 import com.google.protobuf.Message;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.codec.AbstractEncoderTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.MediaType;
 import org.springframework.protobuf.Msg;
 import org.springframework.protobuf.SecondMsg;
@@ -40,7 +41,7 @@ import static org.springframework.core.ResolvableType.forClass;
  *
  * @author Sebastien Deleuze
  */
-public class ProtobufEncoderTests extends AbstractEncoderTestCase<Message, ProtobufEncoder> {
+public class ProtobufEncoderTests extends AbstractEncoderTestCase<ProtobufEncoder> {
 
 	private final static MimeType PROTOBUF_MIME_TYPE = new MimeType("application", "x-protobuf");
 
@@ -50,42 +51,64 @@ public class ProtobufEncoderTests extends AbstractEncoderTestCase<Message, Proto
 	private Msg msg2 =
 			Msg.newBuilder().setFoo("Bar").setBlah(SecondMsg.newBuilder().setBlah(456).build()).build();
 
+
 	public ProtobufEncoderTests() {
-		super(new ProtobufEncoder(), Msg.class);
+		super(new ProtobufEncoder());
 	}
 
 	@Override
-	protected Flux<Message> input() {
-		return Flux.just(this.msg1, this.msg2);
+	@Test
+	public void canEncode() {
+		assertTrue(this.encoder.canEncode(forClass(Msg.class), null));
+		assertTrue(this.encoder.canEncode(forClass(Msg.class), PROTOBUF_MIME_TYPE));
+		assertTrue(this.encoder.canEncode(forClass(Msg.class), MediaType.APPLICATION_OCTET_STREAM));
+		assertFalse(this.encoder.canEncode(forClass(Msg.class), MediaType.APPLICATION_JSON));
+		assertFalse(this.encoder.canEncode(forClass(Object.class), PROTOBUF_MIME_TYPE));
 	}
 
 	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(resultConsumer(this.msg1))
-				.add(resultConsumer(this.msg2))
-				.build();
+	@Test
+	public void encode() {
+		Mono<Message> input = Mono.just(this.msg1);
+
+		testEncodeAll(input, Msg.class, step -> step
+				.consumeNextWith(dataBuffer -> {
+					try {
+						assertEquals(this.msg1, Msg.parseFrom(dataBuffer.asInputStream()));
+
+					}
+					catch (IOException ex) {
+						throw new UncheckedIOException(ex);
+					}
+					finally {
+						DataBufferUtils.release(dataBuffer);
+					}
+				})
+				.verifyComplete());
+	}
+
+	@Test
+	public void encodeStream() {
+		Flux<Message> input = Flux.just(this.msg1, this.msg2);
 
+		testEncodeAll(input, Msg.class, step -> step
+				.consumeNextWith(expect(this.msg1))
+				.consumeNextWith(expect(this.msg2))
+				.verifyComplete());
 	}
 
-	protected final Consumer<DataBuffer> resultConsumer(Msg msg) {
+	protected final Consumer<DataBuffer> expect(Msg msg) {
 		return dataBuffer -> {
 			try {
 				assertEquals(msg, Msg.parseDelimitedFrom(dataBuffer.asInputStream()));
+
 			}
 			catch (IOException ex) {
 				throw new UncheckedIOException(ex);
 			}
+			finally {
+				DataBufferUtils.release(dataBuffer);
+			}
 		};
 	}
-
-	@Test
-	public void canEncode() {
-		assertTrue(this.encoder.canEncode(forClass(Msg.class), null));
-		assertTrue(this.encoder.canEncode(forClass(Msg.class), PROTOBUF_MIME_TYPE));
-		assertTrue(this.encoder.canEncode(forClass(Msg.class), MediaType.APPLICATION_OCTET_STREAM));
-		assertFalse(this.encoder.canEncode(forClass(Msg.class), MediaType.APPLICATION_JSON));
-		assertFalse(this.encoder.canEncode(forClass(Object.class), PROTOBUF_MIME_TYPE));
-	}
-
 }
diff --git a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
index 63f4abff4f..7d73e7d82d 100644
--- a/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
+++ b/spring-web/src/test/java/org/springframework/http/codec/xml/Jaxb2XmlEncoderTests.java
@@ -16,38 +16,39 @@
 
 package org.springframework.http.codec.xml;
 
-import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
 import javax.xml.bind.annotation.XmlElement;
 import javax.xml.bind.annotation.XmlElements;
 import javax.xml.bind.annotation.XmlRootElement;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractEncoderTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.Pojo;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.*;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 import static org.xmlunit.matchers.CompareMatcher.isSimilarTo;
 
 /**
  * @author Sebastien Deleuze
  * @author Arjen Poutsma
  */
-public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Object, Jaxb2XmlEncoder> {
+public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Jaxb2XmlEncoder> {
 
 	public Jaxb2XmlEncoderTests() {
-		super(new Jaxb2XmlEncoder(), Pojo.class);
+		super(new Jaxb2XmlEncoder());
 	}
 
+	@Override
 	@Test
 	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(Pojo.class),
@@ -69,23 +70,46 @@ public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Object, Jaxb2X
 	}
 
 	@Override
-	protected Flux<Object> input() {
-		return Flux.just(new Container());
+	@Test
+	public void encode() {
+		Mono<Pojo> input = Mono.just(new Pojo("foofoo", "barbar"));
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(
+						expectXml("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
+								"<pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"))
+				.verifyComplete());
 	}
 
-	@Override
-	protected Stream<Consumer<DataBuffer>> outputConsumers() {
-		return Stream.<Consumer<DataBuffer>>builder()
-				.add(dataBuffer -> {
-					String s = DataBufferTestUtils
-							.dumpString(dataBuffer, StandardCharsets.UTF_8);
-					assertThat(s,
-							isSimilarTo("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
-									"<container><foo><name>name1</name></foo><bar><title>title1</title></bar></container>"));
-				})
-				.build();
+	@Test
+	public void encodeError() {
+		Flux<Pojo> input = Flux.error(RuntimeException::new);
+
+		testEncode(input, Pojo.class, step -> step
+				.expectError(RuntimeException.class)
+				.verify());
 	}
 
+	@Test
+	public void encodeElementsWithCommonType() {
+		Mono<Container> input = Mono.just(new Container());
+
+		testEncode(input, Pojo.class, step -> step
+				.consumeNextWith(
+						expectXml("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
+								"<container><foo><name>name1</name></foo><bar><title>title1</title></bar></container>"))
+				.verifyComplete());
+	}
+
+	protected Consumer<DataBuffer> expectXml(String expected) {
+		return dataBuffer -> {
+			byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
+			dataBuffer.read(resultBytes);
+			release(dataBuffer);
+			String actual = new String(resultBytes, UTF_8);
+			assertThat(actual, isSimilarTo(expected));
+		};
+	}
 
 	public static class Model {}