Resolved DataBuffer leaks in codec package

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

@@ -23,11 +23,13 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
 /**
  * @author Sebastien Deleuze
+ * @author Arjen Poutsma
  */
 public class ByteBufferDecoder extends AbstractDecoder<ByteBuffer> {
 
@@ -46,7 +48,13 @@ public class ByteBufferDecoder extends AbstractDecoder<ByteBuffer> {
 	@Override
 	public Flux<ByteBuffer> decode(Publisher<DataBuffer> inputStream, ResolvableType type,
 			MimeType mimeType, Object... hints) {
-		return Flux.from(inputStream).map(DataBuffer::asByteBuffer);
+		return Flux.from(inputStream).map((dataBuffer) -> {
+			ByteBuffer copy = ByteBuffer.allocate(dataBuffer.readableByteCount());
+			copy.put(dataBuffer.asByteBuffer());
+			copy.flip();
+			DataBufferUtils.release(dataBuffer);
+			return copy;
+		});
 	}
 
 }
\ No newline at end of file

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

@@ -66,6 +66,7 @@ public class ResourceDecoder extends AbstractDecoder<Resource> {
 			return Flux.from(singleBuffer.map(buffer -> {
 				byte[] bytes = new byte[buffer.readableByteCount()];
 				buffer.read(bytes);
+				DataBufferUtils.release(buffer);
 				return new ByteArrayResource(bytes);
 			}));
 		}

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

@@ -25,6 +25,7 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
@@ -84,8 +85,9 @@ public class StringDecoder extends AbstractDecoder<String> {
 			inputFlux = Flux.from(inputFlux.reduce(DataBuffer::write));
 		}
 		Charset charset = getCharset(mimeType);
-		return inputFlux.map(content -> {
-			CharBuffer charBuffer = charset.decode(content.asByteBuffer());
+		return inputFlux.map(dataBuffer -> {
+			CharBuffer charBuffer = charset.decode(dataBuffer.asByteBuffer());
+			DataBufferUtils.release(dataBuffer);
 			return charBuffer.toString();
 		});
 	}

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

@@ -21,26 +21,31 @@ import java.nio.ByteBuffer;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
+import reactor.core.test.TestSubscriber;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static org.junit.Assert.*;
-import reactor.core.test.TestSubscriber;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author Sebastien Deleuze
  */
-public class ByteBufferDecoderTests extends AbstractAllocatingTestCase {
+public class ByteBufferDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private final ByteBufferDecoder decoder = new ByteBufferDecoder();
 
 	@Test
 	public void canDecode() {
-		assertTrue(decoder.canDecode(ResolvableType.forClass(ByteBuffer.class), MediaType.TEXT_PLAIN));
-		assertFalse(decoder.canDecode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(ByteBuffer.class), MediaType.APPLICATION_JSON));
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(ByteBuffer.class),
+				MediaType.TEXT_PLAIN));
+		assertFalse(this.decoder
+				.canDecode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(ByteBuffer.class),
+				MediaType.APPLICATION_JSON));
 	}
 
 	@Test
@@ -48,20 +53,12 @@ public class ByteBufferDecoderTests extends AbstractAllocatingTestCase {
 		DataBuffer fooBuffer = stringBuffer("foo");
 		DataBuffer barBuffer = stringBuffer("bar");
 		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
-		Flux<ByteBuffer> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null);
+		Flux<ByteBuffer> output = this.decoder.decode(source,
+				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
+				null);
 		TestSubscriber<ByteBuffer> testSubscriber = new TestSubscriber<>();
-		testSubscriber.bindTo(output)
-				.assertValuesWith(b -> assertBufferEquals(fooBuffer, b), b -> assertBufferEquals(barBuffer, b));
+		testSubscriber.bindTo(output).assertNoError().assertComplete()
+				.assertValues(ByteBuffer.wrap("foo".getBytes()),
+						ByteBuffer.wrap("bar".getBytes()));
 	}
-
-	public void assertBufferEquals(DataBuffer expected, ByteBuffer actual) {
-		byte[] byteBufferBytes = new byte[actual.remaining()];
-		actual.get(byteBufferBytes);
-
-		byte[] dataBufferBytes = new byte[expected.readableByteCount()];
-		expected.read(dataBufferBytes);
-
-		assertArrayEquals(dataBufferBytes, byteBufferBytes);
-	}
-
 }

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

@@ -26,6 +26,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
@@ -34,20 +35,23 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class ByteBufferEncoderTests extends AbstractAllocatingTestCase {
+public class ByteBufferEncoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private ByteBufferEncoder encoder;
 
 	@Before
 	public void createEncoder() {
-		encoder = new ByteBufferEncoder();
+		this.encoder = new ByteBufferEncoder();
 	}
 
 	@Test
 	public void canEncode() {
-		assertTrue(encoder.canEncode(ResolvableType.forClass(ByteBuffer.class), MediaType.TEXT_PLAIN));
-		assertFalse(encoder.canEncode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
-		assertTrue(encoder.canEncode(ResolvableType.forClass(ByteBuffer.class), MediaType.APPLICATION_JSON));
+		assertTrue(this.encoder.canEncode(ResolvableType.forClass(ByteBuffer.class),
+				MediaType.TEXT_PLAIN));
+		assertFalse(this.encoder
+				.canEncode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
+		assertTrue(this.encoder.canEncode(ResolvableType.forClass(ByteBuffer.class),
+				MediaType.APPLICATION_JSON));
 	}
 
 	@Test
@@ -57,7 +61,7 @@ public class ByteBufferEncoderTests extends AbstractAllocatingTestCase {
 		Flux<ByteBuffer> source =
 				Flux.just(ByteBuffer.wrap(fooBytes), ByteBuffer.wrap(barBytes));
 
-		Flux<DataBuffer> output = encoder.encode(source, allocator,
+		Flux<DataBuffer> output = this.encoder.encode(source, this.allocator,
 				ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class),
 				null);
 		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();

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

@@ -18,32 +18,35 @@ package org.springframework.core.codec.support;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.test.TestSubscriber;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
-import static org.junit.Assert.*;
-import reactor.core.test.TestSubscriber;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author Sebastien Deleuze
  */
-public class JacksonJsonDecoderTests extends AbstractAllocatingTestCase {
+public class JacksonJsonDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private final JacksonJsonDecoder decoder = new JacksonJsonDecoder();
 
 	@Test
 	public void canDecode() {
-		assertTrue(decoder.canDecode(null, MediaType.APPLICATION_JSON));
-		assertFalse(decoder.canDecode(null, MediaType.APPLICATION_XML));
+		assertTrue(this.decoder.canDecode(null, MediaType.APPLICATION_JSON));
+		assertFalse(this.decoder.canDecode(null, MediaType.APPLICATION_XML));
 	}
 
 	@Test
 	public void decode() {
 		Flux<DataBuffer> source =
 				Flux.just(stringBuffer("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}"));
-		Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
+		Flux<Object> output =
+				this.decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
 		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(output).assertValues(new Pojo("foofoo", "barbar"));
 	}

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

@@ -16,13 +16,13 @@
 
 package org.springframework.core.codec.support;
 
-import java.nio.charset.StandardCharsets;
-
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
 import static org.junit.Assert.assertFalse;
@@ -31,33 +31,36 @@ import static org.junit.Assert.assertTrue;
 /**
  * @author Sebastien Deleuze
  */
-public class JacksonJsonEncoderTests extends AbstractAllocatingTestCase {
+public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private JacksonJsonEncoder encoder;
 
 	@Before
 	public void createEncoder() {
-		encoder = new JacksonJsonEncoder();
+		this.encoder = new JacksonJsonEncoder();
 	}
 
 	@Test
 	public void canWrite() {
-		assertTrue(encoder.canEncode(null, MediaType.APPLICATION_JSON));
-		assertFalse(encoder.canEncode(null, MediaType.APPLICATION_XML));
+		assertTrue(this.encoder.canEncode(null, MediaType.APPLICATION_JSON));
+		assertFalse(this.encoder.canEncode(null, MediaType.APPLICATION_XML));
 	}
 
 	@Test
 	public void write() {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
-		Flux<String> output = encoder.encode(source, allocator, null, null).map(chunk -> {
-			byte[] b = new byte[chunk.readableByteCount()];
-			chunk.read(b);
-			return new String(b, StandardCharsets.UTF_8);
-		});
-		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
-		testSubscriber.bindTo(output)
-				.assertValues("[", "{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", ",",
-						"{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}", "]");
+
+		Flux<DataBuffer> output = this.encoder.encode(source, this.allocator, null, null);
+
+		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output).
+				assertComplete().
+				assertNoError().
+				assertValuesWith(stringConsumer("["),
+						stringConsumer("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}"),
+						stringConsumer(","),
+						stringConsumer("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}"),
+						stringConsumer("]"));
 	}
 
 }

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

@@ -31,6 +31,7 @@ import org.springframework.core.codec.support.jaxb.XmlRootElementWithNameAndName
 import org.springframework.core.codec.support.jaxb.XmlType;
 import org.springframework.core.codec.support.jaxb.XmlTypeWithName;
 import org.springframework.core.codec.support.jaxb.XmlTypeWithNameAndNamespace;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
@@ -39,7 +40,7 @@ import static org.junit.Assert.*;
 /**
  * @author Sebastien Deleuze
  */
-public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
+public class Jaxb2DecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private static final String POJO_ROOT = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
 			"<pojo>" +
@@ -67,25 +68,25 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void canDecode() {
-		assertTrue(decoder.canDecode(ResolvableType.forClass(Pojo.class),
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(Pojo.class),
 				MediaType.APPLICATION_XML));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(Pojo.class),
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(Pojo.class),
 				MediaType.TEXT_XML));
-		assertFalse(decoder.canDecode(ResolvableType.forClass(Pojo.class),
+		assertFalse(this.decoder.canDecode(ResolvableType.forClass(Pojo.class),
 				MediaType.APPLICATION_JSON));
 
-		assertTrue(decoder.canDecode(ResolvableType.forClass(TypePojo.class),
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(TypePojo.class),
 				MediaType.APPLICATION_XML));
 
-		assertFalse(decoder.canDecode(ResolvableType.forClass(getClass()),
+		assertFalse(this.decoder.canDecode(ResolvableType.forClass(getClass()),
 				MediaType.APPLICATION_XML));
 	}
 
 	@Test
 	public void splitOneBranches() {
-		Flux<XMLEvent> xmlEvents =
-				xmlEventDecoder.decode(Flux.just(stringBuffer(POJO_ROOT)), null, null);
-		Flux<List<XMLEvent>> result = decoder.split(xmlEvents, new QName("pojo"));
+		Flux<XMLEvent> xmlEvents = this.xmlEventDecoder
+				.decode(Flux.just(stringBuffer(POJO_ROOT)), null, null);
+		Flux<List<XMLEvent>> result = this.decoder.split(xmlEvents, new QName("pojo"));
 
 		TestSubscriber<List<XMLEvent>> resultSubscriber = new TestSubscriber<>();
 		resultSubscriber.bindTo(result).
@@ -108,9 +109,9 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void splitMultipleBranches() {
-		Flux<XMLEvent> xmlEvents =
-				xmlEventDecoder.decode(Flux.just(stringBuffer(POJO_CHILD)), null, null);
-		Flux<List<XMLEvent>> result = decoder.split(xmlEvents, new QName("pojo"));
+		Flux<XMLEvent> xmlEvents = this.xmlEventDecoder
+				.decode(Flux.just(stringBuffer(POJO_CHILD)), null, null);
+		Flux<List<XMLEvent>> result = this.decoder.split(xmlEvents, new QName("pojo"));
 
 		TestSubscriber<List<XMLEvent>> resultSubscriber = new TestSubscriber<>();
 		resultSubscriber.bindTo(result).
@@ -158,7 +159,7 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 	public void decodeSingleXmlRootElement() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_ROOT));
 		Flux<Object> output =
-				decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
+				this.decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
 
 		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
 
@@ -173,8 +174,8 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 	@Test
 	public void decodeSingleXmlTypeElement() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_ROOT));
-		Flux<Object> output =
-				decoder.decode(source, ResolvableType.forClass(TypePojo.class), null);
+		Flux<Object> output = this.decoder
+				.decode(source, ResolvableType.forClass(TypePojo.class), null);
 
 		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
 
@@ -190,7 +191,7 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 	public void decodeMultipleXmlRootElement() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_CHILD));
 		Flux<Object> output =
-				decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
+				this.decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
 
 		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
 
@@ -205,8 +206,8 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 	@Test
 	public void decodeMultipleXmlTypeElement() throws Exception {
 		Flux<DataBuffer> source = Flux.just(stringBuffer(POJO_CHILD));
-		Flux<Object> output =
-				decoder.decode(source, ResolvableType.forClass(TypePojo.class), null);
+		Flux<Object> output = this.decoder
+				.decode(source, ResolvableType.forClass(TypePojo.class), null);
 
 		TestSubscriber<Object> testSubscriber = new TestSubscriber<>();
 
@@ -220,21 +221,22 @@ public class Jaxb2DecoderTests extends AbstractAllocatingTestCase {
 
 	@Test
 	public void toExpectedQName() {
-		assertEquals(new QName("pojo"), decoder.toQName(Pojo.class));
-		assertEquals(new QName("pojo"), decoder.toQName(TypePojo.class));
+		assertEquals(new QName("pojo"), this.decoder.toQName(Pojo.class));
+		assertEquals(new QName("pojo"), this.decoder.toQName(TypePojo.class));
 
 		assertEquals(new QName("namespace", "name"),
-				decoder.toQName(XmlRootElementWithNameAndNamespace.class));
+				this.decoder.toQName(XmlRootElementWithNameAndNamespace.class));
 		assertEquals(new QName("namespace", "name"),
-				decoder.toQName(XmlRootElementWithName.class));
+				this.decoder.toQName(XmlRootElementWithName.class));
 		assertEquals(new QName("namespace", "xmlRootElement"),
-				decoder.toQName(XmlRootElement.class));
+				this.decoder.toQName(XmlRootElement.class));
 
 		assertEquals(new QName("namespace", "name"),
-				decoder.toQName(XmlTypeWithNameAndNamespace.class));
+				this.decoder.toQName(XmlTypeWithNameAndNamespace.class));
 		assertEquals(new QName("namespace", "name"),
-				decoder.toQName(XmlTypeWithName.class));
-		assertEquals(new QName("namespace", "xmlType"), decoder.toQName(XmlType.class));
+				this.decoder.toQName(XmlTypeWithName.class));
+		assertEquals(new QName("namespace", "xmlType"),
+				this.decoder.toQName(XmlType.class));
 
 	}
 

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

@@ -26,7 +26,10 @@ import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
 import org.springframework.http.MediaType;
 
 import static org.custommonkey.xmlunit.XMLAssert.assertXMLEqual;
@@ -38,47 +41,51 @@ import static org.junit.Assert.assertTrue;
  * @author Sebastien Deleuze
  * @author Arjen Poutsma
  */
-public class Jaxb2EncoderTests extends AbstractAllocatingTestCase {
+public class Jaxb2EncoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private Jaxb2Encoder encoder;
 
 	@Before
 	public void createEncoder() {
-		encoder = new Jaxb2Encoder();
+		this.encoder = new Jaxb2Encoder();
 	}
 
 	@Test
 	public void canEncode() {
-		assertTrue(encoder.canEncode(ResolvableType.forClass(Pojo.class),
+		assertTrue(this.encoder.canEncode(ResolvableType.forClass(Pojo.class),
 				MediaType.APPLICATION_XML));
-		assertTrue(encoder.canEncode(ResolvableType.forClass(Pojo.class),
+		assertTrue(this.encoder.canEncode(ResolvableType.forClass(Pojo.class),
 				MediaType.TEXT_XML));
-		assertFalse(encoder.canEncode(ResolvableType.forClass(Pojo.class),
+		assertFalse(this.encoder.canEncode(ResolvableType.forClass(Pojo.class),
 				MediaType.APPLICATION_JSON));
 
-		assertTrue(encoder.canEncode(
+		assertTrue(this.encoder.canEncode(
 				ResolvableType.forClass(Jaxb2DecoderTests.TypePojo.class),
 				MediaType.APPLICATION_XML));
 
-		assertFalse(encoder.canEncode(ResolvableType.forClass(getClass()),
+		assertFalse(this.encoder.canEncode(ResolvableType.forClass(getClass()),
 				MediaType.APPLICATION_XML));
 	}
 
 	@Test
 	public void encode() {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
-		Flux<String> output =
-				encoder.encode(source, allocator, ResolvableType.forClass(Pojo.class),
-						MediaType.APPLICATION_XML).map(chunk -> DataBufferTestUtils
-						.dumpString(chunk, StandardCharsets.UTF_8));
-		TestSubscriber<String> testSubscriber = new TestSubscriber<>();
-		testSubscriber.bindTo(output).assertValuesWith(s -> {
+		Flux<DataBuffer> output = this.encoder
+				.encode(source, this.allocator, ResolvableType.forClass(Pojo.class),
+						MediaType.APPLICATION_XML);
+		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
+		testSubscriber.bindTo(output).assertValuesWith(dataBuffer -> {
 			try {
+				String s = DataBufferTestUtils
+						.dumpString(dataBuffer, StandardCharsets.UTF_8);
 				assertXMLEqual("<pojo><bar>barbar</bar><foo>foofoo</foo></pojo>", s);
 			}
 			catch (SAXException | IOException e) {
 				fail(e.getMessage());
 			}
+			finally {
+				DataBufferUtils.release(dataBuffer);
+			}
 		});
 	}
 

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

@@ -22,12 +22,13 @@ import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 
 /**
  * @author Sebastien Deleuze
  */
-public class JsonObjectDecoderTests extends AbstractAllocatingTestCase {
+public class JsonObjectDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 
 	@Test

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

@@ -26,6 +26,7 @@ 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.http.MediaType;
 import org.springframework.util.StreamUtils;
@@ -35,19 +36,22 @@ import static org.junit.Assert.*;
 /**
  * @author Arjen Poutsma
  */
-public class ResourceDecoderTests extends AbstractAllocatingTestCase {
+public class ResourceDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private final ResourceDecoder decoder = new ResourceDecoder();
 
 	@Test
 	public void canDecode() throws Exception {
-		assertTrue(decoder.canDecode(ResolvableType.forClass(InputStreamResource.class),
+		assertTrue(
+				this.decoder.canDecode(ResolvableType.forClass(InputStreamResource.class),
 				MediaType.TEXT_PLAIN));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(ByteArrayResource.class),
+		assertTrue(
+				this.decoder.canDecode(ResolvableType.forClass(ByteArrayResource.class),
 				MediaType.TEXT_PLAIN));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(Resource.class),
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(Resource.class),
 				MediaType.TEXT_PLAIN));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(InputStreamResource.class),
+		assertTrue(
+				this.decoder.canDecode(ResolvableType.forClass(InputStreamResource.class),
 				MediaType.APPLICATION_JSON));
 	}
 
@@ -57,8 +61,8 @@ public class ResourceDecoderTests extends AbstractAllocatingTestCase {
 		DataBuffer barBuffer = stringBuffer("bar");
 		Flux<DataBuffer> source = Flux.just(fooBuffer, barBuffer);
 
-		Flux<Resource> result =
-				decoder.decode(source, ResolvableType.forClass(Resource.class), null);
+		Flux<Resource> result = this.decoder
+				.decode(source, ResolvableType.forClass(Resource.class), null);
 
 		TestSubscriber<Resource> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(result).

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

@@ -27,6 +27,7 @@ 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.http.MediaType;
 
@@ -35,19 +36,22 @@ import static org.junit.Assert.assertTrue;
 /**
  * @author Arjen Poutsma
  */
-public class ResourceEncoderTests extends AbstractAllocatingTestCase {
+public class ResourceEncoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private final ResourceEncoder encoder = new ResourceEncoder();
 
 	@Test
 	public void canEncode() throws Exception {
-		assertTrue(encoder.canEncode(ResolvableType.forClass(InputStreamResource.class),
+		assertTrue(
+				this.encoder.canEncode(ResolvableType.forClass(InputStreamResource.class),
 				MediaType.TEXT_PLAIN));
-		assertTrue(encoder.canEncode(ResolvableType.forClass(ByteArrayResource.class),
+		assertTrue(
+				this.encoder.canEncode(ResolvableType.forClass(ByteArrayResource.class),
 				MediaType.TEXT_PLAIN));
-		assertTrue(encoder.canEncode(ResolvableType.forClass(Resource.class),
+		assertTrue(this.encoder.canEncode(ResolvableType.forClass(Resource.class),
 				MediaType.TEXT_PLAIN));
-		assertTrue(encoder.canEncode(ResolvableType.forClass(InputStreamResource.class),
+		assertTrue(
+				this.encoder.canEncode(ResolvableType.forClass(InputStreamResource.class),
 				MediaType.APPLICATION_JSON));
 	}
 
@@ -58,13 +62,13 @@ public class ResourceEncoderTests extends AbstractAllocatingTestCase {
 
 		Mono<Resource> source = Mono.just(resource);
 
-		Flux<DataBuffer> output =
-				encoder.encode(source, allocator, ResolvableType.forClass(Resource.class),
+		Flux<DataBuffer> output = this.encoder
+				.encode(source, this.allocator, ResolvableType.forClass(Resource.class),
 						null);
 
 		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(output).assertNoError().assertComplete()
-				.assertValues(stringBuffer(s));
+				.assertValuesWith(stringConsumer(s));
 
 	}
 

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

@@ -25,6 +25,7 @@ import reactor.core.test.TestSubscriber;
 import rx.Single;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
 
@@ -35,23 +36,28 @@ import static org.junit.Assert.*;
  * @author Brian Clozel
  * @author Mark Paluch
  */
-public class StringDecoderTests extends AbstractAllocatingTestCase {
+public class StringDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private StringDecoder decoder;
 
 	@Before
 	public void createEncoder() {
-		decoder = new StringDecoder();
+		this.decoder = new StringDecoder();
 	}
 
 
 	@Test
 	public void canDecode() {
-		assertTrue(decoder.canDecode(ResolvableType.forClass(String.class), MediaType.TEXT_PLAIN));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(String.class), MediaType.TEXT_HTML));
-		assertTrue(decoder.canDecode(ResolvableType.forClass(String.class), MediaType.APPLICATION_JSON));
-		assertFalse(decoder.canDecode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
-		assertFalse(decoder.canDecode(ResolvableType.forClass(Pojo.class), MediaType.APPLICATION_JSON));
+		assertTrue(this.decoder
+				.canDecode(ResolvableType.forClass(String.class), MediaType.TEXT_PLAIN));
+		assertTrue(this.decoder
+				.canDecode(ResolvableType.forClass(String.class), MediaType.TEXT_HTML));
+		assertTrue(this.decoder.canDecode(ResolvableType.forClass(String.class),
+				MediaType.APPLICATION_JSON));
+		assertFalse(this.decoder
+				.canDecode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
+		assertFalse(this.decoder.canDecode(ResolvableType.forClass(Pojo.class),
+				MediaType.APPLICATION_JSON));
 	}
 
 	@Test

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

@@ -26,6 +26,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.http.MediaType;
 
 import static org.junit.Assert.assertFalse;
@@ -35,26 +36,29 @@ import static org.junit.Assert.assertTrue;
  * @author Sebastien Deleuze
  */
 @RunWith(Parameterized.class)
-public class StringEncoderTests extends AbstractAllocatingTestCase {
+public class StringEncoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private StringEncoder encoder;
 
 	@Before
 	public void createEncoder() {
-		encoder = new StringEncoder();
+		this.encoder = new StringEncoder();
 	}
 
 	@Test
 	public void canWrite() {
-		assertTrue(encoder.canEncode(ResolvableType.forClass(String.class), MediaType.TEXT_PLAIN));
-		assertFalse(encoder.canEncode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
-		assertFalse(encoder.canEncode(ResolvableType.forClass(String.class), MediaType.APPLICATION_JSON));
+		assertTrue(this.encoder
+				.canEncode(ResolvableType.forClass(String.class), MediaType.TEXT_PLAIN));
+		assertFalse(this.encoder
+				.canEncode(ResolvableType.forClass(Integer.class), MediaType.TEXT_PLAIN));
+		assertFalse(this.encoder.canEncode(ResolvableType.forClass(String.class),
+				MediaType.APPLICATION_JSON));
 	}
 
 	@Test
 	public void write() throws InterruptedException {
-		Flux<String> output =
-				Flux.from(encoder.encode(Flux.just("foo"), allocator, null, null))
+		Flux<String> output = Flux.from(
+				this.encoder.encode(Flux.just("foo"), this.allocator, null, null))
 						.map(chunk -> {
 			byte[] b = new byte[chunk.readableByteCount()];
 			chunk.read(b);

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

@@ -22,13 +22,15 @@ import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 /**
  * @author Arjen Poutsma
  */
-public class XmlEventDecoderTests extends AbstractAllocatingTestCase {
+public class XmlEventDecoderTests extends AbstractDataBufferAllocatingTestCase {
 
 	private static final String XML = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
 			"<pojo>" +
@@ -41,7 +43,8 @@ public class XmlEventDecoderTests extends AbstractAllocatingTestCase {
 	@Test
 	public void toXMLEvents() {
 
-		Flux<XMLEvent> events = decoder.decode(Flux.just(stringBuffer(XML)), null, null);
+		Flux<XMLEvent> events =
+				this.decoder.decode(Flux.just(stringBuffer(XML)), null, null);
 
 		TestSubscriber<XMLEvent> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(events).

spring-web-reactive/src/test/java/org/springframework/core/codec/support/AbstractAllocatingTestCase.java → spring-web-reactive/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java

@@ -14,25 +14,27 @@
  * limitations under the License.
  */
 
-package org.springframework.core.codec.support;
+package org.springframework.core.io.buffer;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.function.Consumer;
 
 import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferAllocator;
-import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
-import org.springframework.core.io.buffer.NettyDataBufferAllocator;
+import org.springframework.core.io.buffer.support.DataBufferTestUtils;
+import org.springframework.core.io.buffer.support.DataBufferUtils;
+
+import static org.junit.Assert.assertEquals;
 
 /**
  * @author Arjen Poutsma
  */
 @RunWith(Parameterized.class)
-public abstract class AbstractAllocatingTestCase {
+public abstract class AbstractDataBufferAllocatingTestCase {
 
 	@Parameterized.Parameter
 	public DataBufferAllocator allocator;
@@ -50,10 +52,28 @@ public abstract class AbstractAllocatingTestCase {
 		};
 	}
 
+	protected DataBuffer createDataBuffer(int capacity) {
+		return this.allocator.allocateBuffer(capacity);
+	}
+
 	protected DataBuffer stringBuffer(String value) {
 		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
-		DataBuffer buffer = allocator.allocateBuffer(bytes.length);
+		DataBuffer buffer = this.allocator.allocateBuffer(bytes.length);
 		buffer.write(bytes);
 		return buffer;
 	}
+
+	protected void release(DataBuffer... buffers) {
+		Arrays.stream(buffers).forEach(DataBufferUtils::release);
+	}
+
+	protected Consumer<DataBuffer> stringConsumer(String expected) {
+		return dataBuffer -> {
+			String value =
+					DataBufferTestUtils.dumpString(dataBuffer, StandardCharsets.UTF_8);
+			assertEquals(expected, value);
+			DataBufferUtils.release(dataBuffer);
+		};
+	}
+
 }

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

@@ -22,13 +22,7 @@ import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.UnpooledByteBufAllocator;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-import org.springframework.core.io.buffer.support.DataBufferUtils;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
@@ -36,31 +30,7 @@ import static org.junit.Assert.assertEquals;
 /**
  * @author Arjen Poutsma
  */
-@RunWith(Parameterized.class)
-public class DataBufferTests {
-
-	@Parameterized.Parameter
-	public DataBufferAllocator allocator;
-
-	@Parameterized.Parameters(name = "{0}")
-	public static Object[][] buffers() {
-
-		return new Object[][]{
-				{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(true))},
-				{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false))},
-				{new NettyDataBufferAllocator(new PooledByteBufAllocator(true))},
-				{new NettyDataBufferAllocator(new PooledByteBufAllocator(false))},
-				{new DefaultDataBufferAllocator(true)},
-				{new DefaultDataBufferAllocator(false)}};
-	}
-
-	private DataBuffer createDataBuffer(int capacity) {
-		return allocator.allocateBuffer(capacity);
-	}
-
-	private void release(DataBuffer... buffers) {
-		Arrays.stream(buffers).forEach(DataBufferUtils::release);
-	}
+public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void writeAndRead() {

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

@@ -18,16 +18,10 @@ package org.springframework.core.io.buffer.support;
 
 import java.nio.charset.StandardCharsets;
 
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.UnpooledByteBufAllocator;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
 
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferAllocator;
-import org.springframework.core.io.buffer.DefaultDataBufferAllocator;
-import org.springframework.core.io.buffer.NettyDataBufferAllocator;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
@@ -35,45 +29,32 @@ import static org.junit.Assert.assertEquals;
 /**
  * @author Arjen Poutsma
  */
-@RunWith(Parameterized.class)
-public class DataBufferTestUtilsTests {
-
-	@Parameterized.Parameter
-	public DataBufferAllocator allocator;
-
-	@Parameterized.Parameters(name = "{0}")
-	public static Object[][] buffers() {
-
-		return new Object[][]{
-				{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(true))},
-				{new NettyDataBufferAllocator(new UnpooledByteBufAllocator(false))},
-				{new NettyDataBufferAllocator(new PooledByteBufAllocator(true))},
-				{new NettyDataBufferAllocator(new PooledByteBufAllocator(false))},
-				{new DefaultDataBufferAllocator(true)},
-				{new DefaultDataBufferAllocator(false)}};
-	}
+public class DataBufferTestUtilsTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void dumpBytes() {
-		DataBuffer buffer = allocator.allocateBuffer(4);
+		DataBuffer buffer = this.allocator.allocateBuffer(4);
 		byte[] source = {'a', 'b', 'c', 'd'};
 		buffer.write(source);
 
 		byte[] result = DataBufferTestUtils.dumpBytes(buffer);
 
 		assertArrayEquals(source, result);
+
+		release(buffer);
 	}
 
 	@Test
 	public void dumpString() {
-		DataBuffer buffer = allocator.allocateBuffer(4);
+		DataBuffer buffer = this.allocator.allocateBuffer(4);
 		String source = "abcd";
 		buffer.write(source.getBytes(StandardCharsets.UTF_8));
 
 		String result = DataBufferTestUtils.dumpString(buffer, StandardCharsets.UTF_8);
 
-
 		assertEquals(source, result);
+
+		release(buffer);
 	}
 
 }
\ No newline at end of file

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

@@ -26,7 +26,7 @@ import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
-import org.springframework.core.codec.support.AbstractAllocatingTestCase;
+import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 
 import static org.junit.Assert.assertFalse;
@@ -34,7 +34,7 @@ import static org.junit.Assert.assertFalse;
 /**
  * @author Arjen Poutsma
  */
-public class DataBufferUtilsTests extends AbstractAllocatingTestCase {
+public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
 
 	@Test
 	public void readChannel() throws Exception {
@@ -42,14 +42,14 @@ public class DataBufferUtilsTests extends AbstractAllocatingTestCase {
 				.toURI();
 		FileChannel channel = FileChannel.open(Paths.get(uri), StandardOpenOption.READ);
 
-		Flux<DataBuffer> flux = DataBufferUtils.read(channel, allocator, 4);
+		Flux<DataBuffer> flux = DataBufferUtils.read(channel, this.allocator, 4);
 
 		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(flux).
 				assertNoError().
 				assertComplete().
-				assertValues(stringBuffer("foo\n"), stringBuffer("bar\n"),
-						stringBuffer("baz\n"), stringBuffer("qux\n"));
+				assertValuesWith(stringConsumer("foo\n"), stringConsumer("bar\n"),
+						stringConsumer("baz\n"), stringConsumer("qux\n"));
 
 		assertFalse(channel.isOpen());
 	}
@@ -60,15 +60,15 @@ public class DataBufferUtilsTests extends AbstractAllocatingTestCase {
 				.toURI();
 		FileChannel channel = FileChannel.open(Paths.get(uri), StandardOpenOption.READ);
 
-		Flux<DataBuffer> flux = DataBufferUtils.read(channel, allocator, 3);
+		Flux<DataBuffer> flux = DataBufferUtils.read(channel, this.allocator, 3);
 
 		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(flux).
 				assertNoError().
 				assertComplete().
-				assertValues(stringBuffer("foo"), stringBuffer("\nba"),
-						stringBuffer("r\nb"), stringBuffer("az\n"), stringBuffer("qux"),
-						stringBuffer("\n"));
+				assertValuesWith(stringConsumer("foo"), stringConsumer("\nba"),
+						stringConsumer("r\nb"), stringConsumer("az\n"),
+						stringConsumer("qux"), stringConsumer("\n"));
 
 		assertFalse(channel.isOpen());
 	}
@@ -78,37 +78,22 @@ public class DataBufferUtilsTests extends AbstractAllocatingTestCase {
 		InputStream is = DataBufferUtilsTests.class
 				.getResourceAsStream("DataBufferUtilsTests.txt");
 
-		Flux<DataBuffer> flux = DataBufferUtils.read(is, allocator, 4);
+		Flux<DataBuffer> flux = DataBufferUtils.read(is, this.allocator, 4);
 
 		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
 		testSubscriber.bindTo(flux).
 				assertNoError().
 				assertComplete().
-				assertValues(stringBuffer("foo\n"), stringBuffer("bar\n"),
-						stringBuffer("baz\n"), stringBuffer("qux\n"));
+				assertValuesWith(stringConsumer("foo\n"), stringConsumer("bar\n"),
+						stringConsumer("baz\n"), stringConsumer("qux\n"));
 	}
 
-	@Test
-	public void readUnalignedInputStream() throws Exception {
-		InputStream is = DataBufferUtilsTests.class
-				.getResourceAsStream("DataBufferUtilsTests.txt");
-
-		Flux<DataBuffer> flux = DataBufferUtils.read(is, allocator, 3);
-
-		TestSubscriber<DataBuffer> testSubscriber = new TestSubscriber<>();
-		testSubscriber.bindTo(flux).
-				assertNoError().
-				assertComplete().
-				assertValues(stringBuffer("foo"), stringBuffer("\nba"),
-						stringBuffer("r\nb"), stringBuffer("az\n"), stringBuffer("qux"),
-						stringBuffer("\n"));
-	}
-
-
 	@Test
 	public void takeUntilByteCount() {
-		Flux<DataBuffer> flux =
-				Flux.just(stringBuffer("foo"), stringBuffer("bar"), stringBuffer("baz"));
+		DataBuffer foo = stringBuffer("foo");
+		DataBuffer bar = stringBuffer("bar");
+		DataBuffer baz = stringBuffer("baz");
+		Flux<DataBuffer> flux = Flux.just(foo, bar, baz);
 
 		Flux<DataBuffer> result = DataBufferUtils.takeUntilByteCount(flux, 5L);
 
@@ -116,7 +101,9 @@ public class DataBufferUtilsTests extends AbstractAllocatingTestCase {
 		testSubscriber.bindTo(result).
 				assertNoError().
 				assertComplete().
-				assertValues(stringBuffer("foo"), stringBuffer("ba"));
+				assertValuesWith(stringConsumer("foo"), stringConsumer("ba"));
+
+		release(bar, baz);
 	}