Add and use AbstractEncoderTestCase

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

Issue: SPR-17449

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

@@ -0,0 +1,243 @@
+/*
+ * 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 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.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;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.*;
+
+/**
+ * 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.
+ *
+ * @author Arjen Poutsma
+ */
+@SuppressWarnings("ProtectedField")
+public abstract class AbstractEncoderTestCase<T, E extends Encoder<T>> {
+
+	/**
+	 * The data buffer factory used by the encoder.
+	 */
+	protected final DataBufferFactory bufferFactory =
+			new LeakAwareDataBufferFactory();
+
+	/**
+	 * The encoder to test.
+	 */
+	protected final E encoder;
+
+	/**
+	 * The type used for
+	 * {@link Encoder#encode(Publisher, DataBufferFactory, ResolvableType, MimeType, Map)}.
+	 */
+	protected final ResolvableType elementType;
+
+	/**
+	 * The mime type used for
+	 * {@link Encoder#encode(Publisher, DataBufferFactory, ResolvableType, MimeType, Map)}.
+	 * May be {@code null}.
+	 */
+	@Nullable
+	protected final MimeType mimeType;
+
+	/**
+	 * The hints used for
+	 * {@link Encoder#encode(Publisher, DataBufferFactory, ResolvableType, MimeType, Map)}.
+	 * May be {@code null}.
+	 */
+	@Nullable
+	protected final Map<String, Object> hints;
+
+
+	/**
+	 * Construct a new {@code AbstractEncoderTestCase} for the given encoder and element class.
+	 * @param encoder the encoder
+	 * @param elementClass the element class
+	 */
+	protected AbstractEncoderTestCase(E encoder, Class<?> elementClass) {
+		this(encoder, ResolvableType.forClass(elementClass), 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}.
+	 */
+	protected AbstractEncoderTestCase(E encoder, ResolvableType elementType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		Assert.notNull(encoder, "Encoder must not be null");
+		Assert.notNull(elementType, "ElementType must not be null");
+
+		this.encoder = encoder;
+		this.elementType = elementType;
+		this.mimeType = mimeType;
+		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()}.
+	 */
+	protected abstract Flux<T> input();
+
+	/**
+	 * 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()}.
+	 */
+	protected abstract Stream<Consumer<DataBuffer>> outputConsumers();
+
+	private Stream<Consumer<DataBuffer>> outputAndReleaseConsumers() {
+		return outputConsumers()
+				.map(consumer -> consumer.andThen(DataBufferUtils::release));
+	}
+
+	/**
+	 * 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}
+	 */
+	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);
+		};
+
+	}
+
+	/**
+	 * Create a result consumer that expects the given bytes.
+	 * @param expected the expected string
+	 * @return a consumer that expects the given data buffer to be equal to {@code expected}
+	 */
+	protected final Consumer<DataBuffer> resultConsumer(byte[] expected) {
+		return dataBuffer -> {
+			byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
+			dataBuffer.read(resultBytes);
+			assertArrayEquals(expected, resultBytes);
+		};
+	}
+
+	/**
+	 * Tests whether passing {@link #input()} to the encoder can be consumed with
+	 * {@link #outputConsumers()}.
+	 */
+	@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);
+
+		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();
+		}
+	}
+
+	/**
+	 * Tests whether canceling the output of the encoder can be consumed with
+	 * {@link #outputConsumers()}.
+	 */
+	@Test
+	public final void encodeCancel() {
+		Flux<T> input = input();
+
+		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();
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * 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.
@@ -16,35 +16,44 @@
 
 package org.springframework.core.codec;
 
-import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.Collections;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
 
-import org.junit.Before;
 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.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 /**
  * @author Arjen Poutsma
  */
-public class ByteArrayEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ByteArrayEncoderTests extends AbstractEncoderTestCase<byte[], ByteArrayEncoder> {
 
-	private ByteArrayEncoder encoder;
+	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
-	@Before
-	public void createEncoder() {
-		this.encoder = new ByteArrayEncoder();
+	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
+
+	public ByteArrayEncoderTests() {
+		super(new ByteArrayEncoder(), byte[].class);
+	}
+
+	@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
@@ -60,29 +69,4 @@ public class ByteArrayEncoderTests extends AbstractDataBufferAllocatingTestCase
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
-	@Test
-	public void encode() {
-		byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
-		byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
-		Flux<byte[]> source = Flux.just(fooBytes, barBytes);
-
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
-				null, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.consumeNextWith(b -> {
-					byte[] buf = new byte[3];
-					b.read(buf);
-					assertArrayEquals(fooBytes, buf);
-				})
-				.consumeNextWith(b -> {
-					byte[] buf = new byte[3];
-					b.read(buf);
-					assertArrayEquals(barBytes, buf);
-				})
-				.expectComplete()
-				.verify();
-	}
-
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * 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.
@@ -18,33 +18,43 @@ package org.springframework.core.codec;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.Collections;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
 
-import org.junit.Before;
 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.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 /**
  * @author Sebastien Deleuze
  */
-public class ByteBufferEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ByteBufferEncoderTests extends AbstractEncoderTestCase<ByteBuffer, ByteBufferEncoder> {
 
-	private ByteBufferEncoder encoder;
+	private final byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
 
-	@Before
-	public void createEncoder() {
-		this.encoder = new ByteBufferEncoder();
+	private final byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
+
+	public ByteBufferEncoderTests() {
+		super(new ByteBufferEncoder(), ByteBuffer.class);
+	}
+
+	@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
@@ -60,29 +70,5 @@ public class ByteBufferEncoderTests extends AbstractDataBufferAllocatingTestCase
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
-	@Test
-	public void encode() {
-		byte[] fooBytes = "foo".getBytes(StandardCharsets.UTF_8);
-		byte[] barBytes = "bar".getBytes(StandardCharsets.UTF_8);
-		Flux<ByteBuffer> source =
-				Flux.just(ByteBuffer.wrap(fooBytes), ByteBuffer.wrap(barBytes));
-
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
-				null, Collections.emptyMap());
-		StepVerifier.create(output)
-				.consumeNextWith(b -> {
-					byte[] buf = new byte[3];
-					b.read(buf);
-					assertArrayEquals(fooBytes, buf);
-				})
-				.consumeNextWith(b -> {
-					byte[] buf = new byte[3];
-					b.read(buf);
-					assertArrayEquals(barBytes, buf);
-				})
-				.expectComplete()
-				.verify();
-	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * 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.
@@ -16,34 +16,43 @@
 
 package org.springframework.core.codec;
 
-import java.util.Collections;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
 
-import org.junit.Before;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
 import reactor.core.publisher.Flux;
-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;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 /**
  * @author Sebastien Deleuze
  */
-@RunWith(Parameterized.class)
-public class CharSequenceEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class CharSequenceEncoderTests
+		extends AbstractEncoderTestCase<CharSequence, CharSequenceEncoder> {
 
-	private CharSequenceEncoder encoder;
+	private final String foo = "foo";
 
-	@Before
-	public void createEncoder() {
-		this.encoder = CharSequenceEncoder.textPlainOnly();
+	private final String bar = "bar";
+
+	public CharSequenceEncoderTests() {
+		super(CharSequenceEncoder.textPlainOnly(), CharSequence.class);
+	}
+
+	@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
@@ -62,27 +71,4 @@ public class CharSequenceEncoderTests extends AbstractDataBufferAllocatingTestCa
 		// SPR-15464
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
-
-	@Test
-	public void writeString() {
-		Flux<String> stringFlux = Flux.just("foo");
-		Flux<DataBuffer> output = Flux.from(
-				this.encoder.encode(stringFlux, this.bufferFactory, null, null, Collections.emptyMap()));
-		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("foo"))
-				.expectComplete()
-				.verify();
-	}
-
-	@Test
-	public void writeStringBuilder() {
-		Flux<StringBuilder> stringBuilderFlux = Flux.just(new StringBuilder("foo"));
-		Flux<DataBuffer> output = Flux.from(
-				this.encoder.encode(stringBuilderFlux, this.bufferFactory, null, null, Collections.emptyMap()));
-		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("foo"))
-				.expectComplete()
-				.verify();
-	}
-
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * 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.
@@ -16,15 +16,14 @@
 
 package org.springframework.core.codec;
 
-import java.nio.ByteBuffer;
-import java.util.Collections;
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import org.junit.Test;
-import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.util.MimeTypeUtils;
 
@@ -33,9 +32,35 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class DataBufferEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class DataBufferEncoderTests extends AbstractEncoderTestCase<DataBuffer, DataBufferEncoder> {
 
-	private final DataBufferEncoder encoder = new 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);
+	}
+
+	@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() {
@@ -50,19 +75,4 @@ public class DataBufferEncoderTests extends AbstractDataBufferAllocatingTestCase
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
-	@Test
-	public void encode() {
-		DataBuffer fooBuffer = stringBuffer("foo");
-		DataBuffer barBuffer = stringBuffer("bar");
-		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
-
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
-				null, Collections.emptyMap());
-
-		assertSame(source, output);
-
-		release(fooBuffer, barBuffer);
-	}
-
 }

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

@@ -16,34 +16,48 @@
 
 package org.springframework.core.codec;
 
-import java.nio.charset.StandardCharsets;
-import java.util.Collections;
+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 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.util.MimeTypeUtils;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.*;
 
 /**
  * @author Arjen Poutsma
  */
-public class ResourceEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ResourceEncoderTests extends AbstractEncoderTestCase<Resource, 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));
+	}
+
+	@Override
+	protected Stream<Consumer<DataBuffer>> outputConsumers() {
+		return Stream.<Consumer<DataBuffer>>builder()
+				.add(resultConsumer(this.bytes))
+				.build();
+	}
 
-	private final ResourceEncoder encoder = new ResourceEncoder();
 
 	@Test
-	public void canEncode() throws Exception {
+	public void canEncode() {
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(InputStreamResource.class),
 				MimeTypeUtils.TEXT_PLAIN));
 		assertTrue(this.encoder.canEncode(ResolvableType.forClass(ByteArrayResource.class),
@@ -57,21 +71,4 @@ public class ResourceEncoderTests extends AbstractDataBufferAllocatingTestCase {
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
-	@Test
-	public void encode() throws Exception {
-		String s = "foo";
-		Resource resource = new ByteArrayResource(s.getBytes(StandardCharsets.UTF_8));
-
-		Mono<Resource> source = Mono.just(resource);
-
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClass(Resource.class),
-				null, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer(s))
-				.expectComplete()
-				.verify();
-	}
-
 }

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

@@ -16,25 +16,27 @@
 
 package org.springframework.core.codec;
 
-import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.function.Consumer;
 
-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.io.ByteArrayResource;
 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.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.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.core.io.support.ResourceRegion;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.*;
 
 /**
@@ -42,14 +44,12 @@ import static org.junit.Assert.*;
  *
  * @author Brian Clozel
  */
-public class ResourceRegionEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ResourceRegionEncoderTests  {
 
-	private ResourceRegionEncoder encoder;
+	private ResourceRegionEncoder encoder = new ResourceRegionEncoder();
+
+	private DataBufferFactory bufferFactory = new LeakAwareDataBufferFactory();
 
-	@Before
-	public void setUp() {
-		this.encoder = new ResourceRegionEncoder();
-	}
 
 	@Test
 	public void canEncode() {
@@ -68,18 +68,8 @@ public class ResourceRegionEncoderTests extends AbstractDataBufferAllocatingTest
 
 	@Test
 	public void shouldEncodeResourceRegionFileResource() throws Exception {
-		shouldEncodeResourceRegion(
-				new ClassPathResource("ResourceRegionEncoderTests.txt", getClass()));
-	}
-
-	@Test
-	public void shouldEncodeResourceRegionByteArrayResource() throws Exception {
-		String content = "Spring Framework test resource content.";
-		shouldEncodeResourceRegion(new ByteArrayResource(content.getBytes(StandardCharsets.UTF_8)));
-	}
-
-	private void shouldEncodeResourceRegion(Resource resource) {
-		ResourceRegion region = new ResourceRegion(resource, 0, 6);
+		ResourceRegion region = new ResourceRegion(
+				new ClassPathResource("ResourceRegionEncoderTests.txt", getClass()), 0, 6);
 		Flux<DataBuffer> result = this.encoder.encode(Mono.just(region), this.bufferFactory,
 				ResolvableType.forClass(ResourceRegion.class),
 				MimeTypeUtils.APPLICATION_OCTET_STREAM,
@@ -93,42 +83,7 @@ public class ResourceRegionEncoderTests extends AbstractDataBufferAllocatingTest
 
 	@Test
 	public void shouldEncodeMultipleResourceRegionsFileResource() throws Exception {
-		shouldEncodeMultipleResourceRegions(
-				new ClassPathResource("ResourceRegionEncoderTests.txt", getClass()));
-	}
-
-	@Test
-	public void shouldEncodeMultipleResourceRegionsByteArrayResource() throws Exception {
-		String content = "Spring Framework test resource content.";
-		shouldEncodeMultipleResourceRegions(
-				new ByteArrayResource(content.getBytes(StandardCharsets.UTF_8)));
-	}
-
-	@Test
-	public void nonExisting() {
 		Resource resource = new ClassPathResource("ResourceRegionEncoderTests.txt", getClass());
-		Resource nonExisting = new ClassPathResource("does not exist", getClass());
-		Flux<ResourceRegion> regions = Flux.just(
-				new ResourceRegion(resource, 0, 6),
-				new ResourceRegion(nonExisting, 0, 6));
-
-		String boundary = MimeTypeUtils.generateMultipartBoundaryString();
-
-		Flux<DataBuffer> result = this.encoder.encode(regions, this.bufferFactory,
-				ResolvableType.forClass(ResourceRegion.class),
-				MimeType.valueOf("text/plain"),
-				Collections.singletonMap(ResourceRegionEncoder.BOUNDARY_STRING_HINT, boundary));
-
-		StepVerifier.create(result)
-				.consumeNextWith(stringConsumer("\r\n--" + boundary + "\r\n"))
-				.consumeNextWith(stringConsumer("Content-Type: text/plain\r\n"))
-				.consumeNextWith(stringConsumer("Content-Range: bytes 0-5/39\r\n\r\n"))
-				.consumeNextWith(stringConsumer("Spring"))
-				.expectError(EncodingException.class)
-				.verify();
-	}
-
-	private void shouldEncodeMultipleResourceRegions(Resource resource) {
 		Flux<ResourceRegion> regions = Flux.just(
 				new ResourceRegion(resource, 0, 6),
 				new ResourceRegion(resource, 7, 9),
@@ -137,7 +92,7 @@ public class ResourceRegionEncoderTests extends AbstractDataBufferAllocatingTest
 		);
 		String boundary = MimeTypeUtils.generateMultipartBoundaryString();
 
-		Flux<DataBuffer> result = this.encoder.encode(regions, super.bufferFactory,
+		Flux<DataBuffer> result = this.encoder.encode(regions, this.bufferFactory,
 				ResolvableType.forClass(ResourceRegion.class),
 				MimeType.valueOf("text/plain"),
 				Collections.singletonMap(ResourceRegionEncoder.BOUNDARY_STRING_HINT, boundary)
@@ -165,4 +120,38 @@ public class ResourceRegionEncoderTests extends AbstractDataBufferAllocatingTest
 				.verify();
 	}
 
+	@Test
+	public void nonExisting() {
+		Resource resource = new ClassPathResource("ResourceRegionEncoderTests.txt", getClass());
+		Resource nonExisting = new ClassPathResource("does not exist", getClass());
+		Flux<ResourceRegion> regions = Flux.just(
+				new ResourceRegion(resource, 0, 6),
+				new ResourceRegion(nonExisting, 0, 6));
+
+		String boundary = MimeTypeUtils.generateMultipartBoundaryString();
+
+		Flux<DataBuffer> result = this.encoder.encode(regions, this.bufferFactory,
+				ResolvableType.forClass(ResourceRegion.class),
+				MimeType.valueOf("text/plain"),
+				Collections.singletonMap(ResourceRegionEncoder.BOUNDARY_STRING_HINT, boundary));
+
+		StepVerifier.create(result)
+				.consumeNextWith(stringConsumer("\r\n--" + boundary + "\r\n"))
+				.consumeNextWith(stringConsumer("Content-Type: text/plain\r\n"))
+				.consumeNextWith(stringConsumer("Content-Range: bytes 0-5/39\r\n\r\n"))
+				.consumeNextWith(stringConsumer("Spring"))
+				.expectError(EncodingException.class)
+				.verify();
+	}
+
+	protected Consumer<DataBuffer> stringConsumer(String expected) {
+		return dataBuffer -> {
+			String value =
+					DataBufferTestUtils.dumpString(dataBuffer, UTF_8);
+			DataBufferUtils.release(dataBuffer);
+			assertEquals(expected, value);
+		};
+	}
+
+
 }

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

@@ -20,38 +20,64 @@ 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;
-import static java.util.Collections.*;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.Test;
-import static org.junit.Assert.assertEquals;
-import static org.springframework.http.MediaType.*;
-import static org.springframework.http.codec.json.Jackson2JsonEncoder.*;
-import static org.springframework.http.codec.json.JacksonViewBean.*;
-import org.springframework.util.MimeType;
 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.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 org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+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;
+import static org.springframework.http.MediaType.APPLICATION_JSON_UTF8;
+import static org.springframework.http.MediaType.APPLICATION_OCTET_STREAM;
+import static org.springframework.http.MediaType.APPLICATION_STREAM_JSON;
+import static org.springframework.http.MediaType.APPLICATION_XML;
+import static org.springframework.http.codec.json.Jackson2JsonEncoder.JSON_VIEW_HINT;
+import static org.springframework.http.codec.json.JacksonViewBean.MyJacksonView1;
+import static org.springframework.http.codec.json.JacksonViewBean.MyJacksonView3;
 
 /**
  * @author Sebastien Deleuze
  */
-public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class Jackson2JsonEncoderTests extends AbstractEncoderTestCase<Object, Jackson2JsonEncoder> {
 
-	private final Jackson2JsonEncoder encoder = new 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"));
+	}
+
+	@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() {
@@ -94,7 +120,7 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 	}
 
 	@Test
-	public void encode() throws Exception {
+	public void encodeNonStream() {
 		Flux<Pojo> source = Flux.just(
 				new Pojo("foo", "bar"),
 				new Pojo("foofoo", "barbar"),
@@ -104,43 +130,29 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, null, emptyMap());
 
 		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("[" +
+				.consumeNextWith(resultConsumer("[" +
 						"{\"foo\":\"foo\",\"bar\":\"bar\"}," +
 						"{\"foo\":\"foofoo\",\"bar\":\"barbar\"}," +
-						"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}]"))
+						"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}]")
+						.andThen(DataBufferUtils::release))
 				.verifyComplete();
 	}
 
 	@Test
-	public void encodeWithType() throws Exception {
+	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());
 
 		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("[{\"type\":\"foo\"},{\"type\":\"bar\"}]"))
+				.consumeNextWith(resultConsumer("[{\"type\":\"foo\"},{\"type\":\"bar\"}]")
+						.andThen(DataBufferUtils::release))
 				.verifyComplete();
 	}
 
-	@Test
-	public void encodeAsStream() throws Exception {
-		Flux<Pojo> source = 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, APPLICATION_STREAM_JSON, emptyMap());
-
-		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
-				.consumeNextWith(stringConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
-				.consumeNextWith(stringConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
-				.verifyComplete();
-	}
 
 	@Test  // SPR-15727
-	public void encodeAsStreamWithCustomStreamingType() throws Exception {
+	public void encodeAsStreamWithCustomStreamingType() {
 		MediaType fooMediaType = new MediaType("application", "foo");
 		MediaType barMediaType = new MediaType("application", "bar");
 		this.encoder.setStreamingMediaTypes(Arrays.asList(fooMediaType, barMediaType));
@@ -153,14 +165,17 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory, type, barMediaType, emptyMap());
 
 		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n"))
-				.consumeNextWith(stringConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n"))
-				.consumeNextWith(stringConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n"))
+				.consumeNextWith(resultConsumer("{\"foo\":\"foo\",\"bar\":\"bar\"}\n")
+						.andThen(DataBufferUtils::release))
+				.consumeNextWith(resultConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}\n")
+						.andThen(DataBufferUtils::release))
+				.consumeNextWith(resultConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}\n")
+						.andThen(DataBufferUtils::release))
 				.verifyComplete();
 	}
 
 	@Test
-	public void fieldLevelJsonView() throws Exception {
+	public void fieldLevelJsonView() {
 		JacksonViewBean bean = new JacksonViewBean();
 		bean.setWithView1("with");
 		bean.setWithView2("with");
@@ -171,12 +186,13 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);
 
 		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("{\"withView1\":\"with\"}"))
+				.consumeNextWith(resultConsumer("{\"withView1\":\"with\"}")
+						.andThen(DataBufferUtils::release))
 				.verifyComplete();
 	}
 
 	@Test
-	public void classLevelJsonView() throws Exception {
+	public void classLevelJsonView() {
 		JacksonViewBean bean = new JacksonViewBean();
 		bean.setWithView1("with");
 		bean.setWithView2("with");
@@ -187,12 +203,12 @@ public class Jackson2JsonEncoderTests extends AbstractDataBufferAllocatingTestCa
 		Flux<DataBuffer> output = this.encoder.encode(Mono.just(bean), this.bufferFactory, type, null, hints);
 
 		StepVerifier.create(output)
-				.consumeNextWith(stringConsumer("{\"withoutView\":\"without\"}"))
+				.consumeNextWith(resultConsumer("{\"withoutView\":\"without\"}")
+						.andThen(DataBufferUtils::release))
 				.verifyComplete();
 	}
 
-
-	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 	private static class ParentClass {
 	}
 

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

@@ -19,14 +19,17 @@ package org.springframework.http.codec.json;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 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 org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+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;
@@ -35,9 +38,7 @@ import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 import org.springframework.util.MimeType;
 
-import static java.util.Collections.emptyMap;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 import static org.springframework.http.MediaType.APPLICATION_XML;
 
 /**
@@ -45,13 +46,55 @@ import static org.springframework.http.MediaType.APPLICATION_XML;
  *
  * @author Sebastien Deleuze
  */
-public class Jackson2SmileEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class Jackson2SmileEncoderTests extends AbstractEncoderTestCase<Object, 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");
 
 	private final Jackson2SmileEncoder encoder = new Jackson2SmileEncoder();
 
+	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);
+	}
+
+	@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) {
+		return dataBuffer -> {
+			try {
+				Pojo actual = this.mapper.reader().forType(Pojo.class)
+						.readValue(DataBufferTestUtils.dumpBytes(dataBuffer));
+				assertEquals(expected, actual);
+				DataBufferUtils.release(dataBuffer);
+			}
+			catch (IOException ex) {
+				throw new UncheckedIOException(ex);
+			}
+		};
+	}
+
 
 	@Test
 	public void canEncode() {
@@ -74,48 +117,33 @@ public class Jackson2SmileEncoderTests extends AbstractDataBufferAllocatingTestC
 	}
 
 	@Test
-	public void encode() throws Exception {
-		Flux<Pojo> source = 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());
+	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 -> readPojo(mapper, List.class, dataBuffer))
+				.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();
 	}
 
-	@Test
-	public void encodeAsStream() throws Exception {
-		Flux<Pojo> source = 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, STREAM_SMILE_MIME_TYPE, emptyMap());
 
-		ObjectMapper mapper = Jackson2ObjectMapperBuilder.smile().build();
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> readPojo(mapper, Pojo.class, dataBuffer))
-				.consumeNextWith(dataBuffer -> readPojo(mapper, Pojo.class, dataBuffer))
-				.consumeNextWith(dataBuffer -> readPojo(mapper, Pojo.class, dataBuffer))
-				.verifyComplete();
-	}
-
-	public <T> T readPojo(ObjectMapper mapper, Class<T> valueType, DataBuffer dataBuffer) {
-		try {
-			T value = mapper.reader().forType(valueType).readValue(DataBufferTestUtils.dumpBytes(dataBuffer));
-			DataBufferUtils.release(dataBuffer);
-			return value;
-		}
-		catch (IOException ex) {
-			throw new UncheckedIOException(ex);
-		}
-	}
 
 }

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

@@ -18,23 +18,20 @@ 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 reactor.test.StepVerifier;
 
-import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+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;
 import org.springframework.util.MimeType;
 
-import static java.util.Collections.emptyMap;
 import static org.junit.Assert.*;
 import static org.springframework.core.ResolvableType.forClass;
 
@@ -43,13 +40,44 @@ import static org.springframework.core.ResolvableType.forClass;
  *
  * @author Sebastien Deleuze
  */
-public class ProtobufEncoderTests extends AbstractDataBufferAllocatingTestCase {
+public class ProtobufEncoderTests extends AbstractEncoderTestCase<Message, ProtobufEncoder> {
 
 	private final static MimeType PROTOBUF_MIME_TYPE = new MimeType("application", "x-protobuf");
 
-	private final Msg testMsg = Msg.newBuilder().setFoo("Foo").setBlah(SecondMsg.newBuilder().setBlah(123).build()).build();
+	private Msg msg1 =
+			Msg.newBuilder().setFoo("Foo").setBlah(SecondMsg.newBuilder().setBlah(123).build()).build();
 
-	private final ProtobufEncoder encoder = new ProtobufEncoder();
+	private Msg msg2 =
+			Msg.newBuilder().setFoo("Bar").setBlah(SecondMsg.newBuilder().setBlah(456).build()).build();
+
+	public ProtobufEncoderTests() {
+		super(new ProtobufEncoder(), Msg.class);
+	}
+
+	@Override
+	protected Flux<Message> input() {
+		return Flux.just(this.msg1, this.msg2);
+	}
+
+	@Override
+	protected Stream<Consumer<DataBuffer>> outputConsumers() {
+		return Stream.<Consumer<DataBuffer>>builder()
+				.add(resultConsumer(this.msg1))
+				.add(resultConsumer(this.msg2))
+				.build();
+
+	}
+
+	protected final Consumer<DataBuffer> resultConsumer(Msg msg) {
+		return dataBuffer -> {
+			try {
+				assertEquals(msg, Msg.parseDelimitedFrom(dataBuffer.asInputStream()));
+			}
+			catch (IOException ex) {
+				throw new UncheckedIOException(ex);
+			}
+		};
+	}
 
 	@Test
 	public void canEncode() {
@@ -60,63 +88,4 @@ public class ProtobufEncoderTests extends AbstractDataBufferAllocatingTestCase {
 		assertFalse(this.encoder.canEncode(forClass(Object.class), PROTOBUF_MIME_TYPE));
 	}
 
-	@Test
-	public void encode() {
-		Mono<Message> message = Mono.just(this.testMsg);
-		ResolvableType elementType = forClass(Msg.class);
-		Flux<DataBuffer> output = this.encoder.encode(message, this.bufferFactory, elementType, PROTOBUF_MIME_TYPE, emptyMap());
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> {
-					try {
-						assertEquals(this.testMsg, Msg.parseFrom(dataBuffer.asInputStream()));
-						DataBufferUtils.release(dataBuffer);
-					}
-					catch (IOException ex) {
-						throw new UncheckedIOException(ex);
-					}
-				})
-				.verifyComplete();
-	}
-
-	@Test
-	public void encodeError() {
-		Flux<Msg> messages = Flux.just(this.testMsg)
-				.concatWith(Flux.error(new RuntimeException()));
-
-		ResolvableType elementType = forClass(Msg.class);
-		Flux<DataBuffer> output = this.encoder.encode(messages, this.bufferFactory, elementType, PROTOBUF_MIME_TYPE, emptyMap());
-		StepVerifier.create(output)
-				.consumeNextWith(DataBufferUtils::release)
-				.expectError(RuntimeException.class)
-				.verify();
-	}
-
-	@Test
-	public void encodeStream() {
-		Msg testMsg2 = Msg.newBuilder().setFoo("Bar").setBlah(SecondMsg.newBuilder().setBlah(456).build()).build();
-		Flux<Message> messages = Flux.just(this.testMsg, testMsg2);
-		ResolvableType elementType = forClass(Msg.class);
-		Flux<DataBuffer> output = this.encoder.encode(messages, this.bufferFactory, elementType, PROTOBUF_MIME_TYPE, emptyMap());
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> {
-					try {
-						assertEquals(this.testMsg, Msg.parseDelimitedFrom(dataBuffer.asInputStream()));
-						DataBufferUtils.release(dataBuffer);
-					}
-					catch (IOException ex) {
-						throw new UncheckedIOException(ex);
-					}
-				})
-				.consumeNextWith(dataBuffer -> {
-					try {
-						assertEquals(testMsg2, Msg.parseDelimitedFrom(dataBuffer.asInputStream()));
-						DataBufferUtils.release(dataBuffer);
-					}
-					catch (IOException ex) {
-						throw new UncheckedIOException(ex);
-					}
-				})
-				.verifyComplete();
-	}
-
 }

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

@@ -18,21 +18,19 @@ package org.springframework.http.codec.xml;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
-import java.util.Collections;
 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 reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+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.MediaType;
 import org.springframework.http.codec.Pojo;
@@ -44,10 +42,11 @@ import static org.xmlunit.matchers.CompareMatcher.isSimilarTo;
  * @author Sebastien Deleuze
  * @author Arjen Poutsma
  */
-public class Jaxb2XmlEncoderTests extends AbstractDataBufferAllocatingTestCase {
-
-	private final Jaxb2XmlEncoder encoder = new Jaxb2XmlEncoder();
+public class Jaxb2XmlEncoderTests extends AbstractEncoderTestCase<Object, Jaxb2XmlEncoder> {
 
+	public Jaxb2XmlEncoderTests() {
+		super(new Jaxb2XmlEncoder(), Pojo.class);
+	}
 
 	@Test
 	public void canEncode() {
@@ -69,60 +68,22 @@ public class Jaxb2XmlEncoderTests extends AbstractDataBufferAllocatingTestCase {
 		assertFalse(this.encoder.canEncode(ResolvableType.NONE, null));
 	}
 
-	@Test
-	public void encode() {
-		Mono<Pojo> source = Mono.just(new Pojo("foofoo", "barbar"));
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClass(Pojo.class),
-				MediaType.APPLICATION_XML, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> {
-					try {
-						String s = DataBufferTestUtils
-								.dumpString(dataBuffer, StandardCharsets.UTF_8);
-						assertThat(s, isSimilarTo("<?xml version='1.0' encoding='UTF-8' standalone='yes'?>" +
-								"<pojo><bar>barbar</bar><foo>foofoo</foo></pojo>"));
-					}
-					finally {
-						DataBufferUtils.release(dataBuffer);
-					}
-				})
-				.verifyComplete();
+	@Override
+	protected Flux<Object> input() {
+		return Flux.just(new Container());
 	}
 
-	@Test
-	public void encodeError() {
-		Flux<Pojo> source = Flux.error(RuntimeException::new);
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClass(Pojo.class),
-				MediaType.APPLICATION_XML, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.expectError(RuntimeException.class)
-				.verify();
-	}
-
-	@Test
-	public void encodeElementsWithCommonType() {
-		Mono<Container> source = Mono.just(new Container());
-		Flux<DataBuffer> output = this.encoder.encode(source, this.bufferFactory,
-				ResolvableType.forClass(Pojo.class),
-				MediaType.APPLICATION_XML, Collections.emptyMap());
-
-		StepVerifier.create(output)
-				.consumeNextWith(dataBuffer -> {
-					try {
-						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>"));
-					}
-					finally {
-						DataBufferUtils.release(dataBuffer);
-					}
+	@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>"));
 				})
-				.verifyComplete();
+				.build();
 	}