Added allocator property to DataBuffer

spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DataBuffer.java

@@ -27,6 +27,12 @@ import java.nio.ByteBuffer;
  */
 public interface DataBuffer {
 
+	/**
+	 * Returns the {@link DataBufferAllocator} that created this buffer.
+	 * @return the creating buffer allocator
+	 */
+	DataBufferAllocator allocator();
+
 	/**
 	 * Gets the byte at the specified index.
 	 * @param index the index

spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBuffer.java

@@ -36,6 +36,8 @@ import org.springframework.util.ObjectUtils;
  */
 public class DefaultDataBuffer implements DataBuffer {
 
+	private final DefaultDataBufferAllocator allocator;
+
 	private ByteBuffer byteBuffer;
 
 	private int readPosition;
@@ -48,20 +50,27 @@ public class DefaultDataBuffer implements DataBuffer {
 	 * ByteBuffer#position() position} of the given buffer.
 	 * @param byteBuffer the buffer to base this buffer on
 	 */
-	DefaultDataBuffer(ByteBuffer byteBuffer) {
-		this(byteBuffer, byteBuffer.position(), byteBuffer.position());
+	DefaultDataBuffer(ByteBuffer byteBuffer, DefaultDataBufferAllocator allocator) {
+		this(byteBuffer, byteBuffer.position(), byteBuffer.position(), allocator);
 	}
 
-	DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition) {
+	DefaultDataBuffer(ByteBuffer byteBuffer, int readPosition, int writePosition, DefaultDataBufferAllocator allocator) {
 		Assert.notNull(byteBuffer, "'byteBuffer' must not be null");
 		Assert.isTrue(readPosition >= 0, "'readPosition' must be 0 or higher");
 		Assert.isTrue(writePosition >= 0, "'writePosition' must be 0 or higher");
 		Assert.isTrue(readPosition <= writePosition,
 				"'readPosition' must be smaller than or equal to 'writePosition'");
+		Assert.notNull(allocator, "'allocator' must not be null");
 
 		this.byteBuffer = byteBuffer;
 		this.readPosition = readPosition;
 		this.writePosition = writePosition;
+		this.allocator = allocator;
+	}
+
+	@Override
+	public DefaultDataBufferAllocator allocator() {
+		return this.allocator;
 	}
 
 	/**

spring-web-reactive/src/main/java/org/springframework/core/io/buffer/DefaultDataBufferAllocator.java

@@ -80,14 +80,14 @@ public class DefaultDataBufferAllocator implements DataBufferAllocator {
 	@Override
 	public DefaultDataBuffer allocateBuffer(int initialCapacity) {
 		return this.preferDirect ?
-				new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity)) :
-				new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity));
+				new DefaultDataBuffer(ByteBuffer.allocateDirect(initialCapacity), this) :
+				new DefaultDataBuffer(ByteBuffer.allocate(initialCapacity), this);
 	}
 
 	@Override
 	public DefaultDataBuffer wrap(ByteBuffer byteBuffer) {
 		ByteBuffer sliced = byteBuffer.slice();
-		return new DefaultDataBuffer(sliced, 0, byteBuffer.remaining());
+		return new DefaultDataBuffer(sliced, 0, byteBuffer.remaining(), this);
 	}
 
 	@Override

spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBuffer.java

@@ -38,16 +38,25 @@ import org.springframework.util.ObjectUtils;
  */
 public class NettyDataBuffer implements DataBuffer {
 
+	private final NettyDataBufferAllocator allocator;
+
 	private ByteBuf byteBuf;
 
 	/**
 	 * Creates a new {@code NettyDataBuffer} based on the given {@code ByteBuff}.
 	 * @param byteBuf the buffer to base this buffer on
 	 */
-	public NettyDataBuffer(ByteBuf byteBuf) {
+	NettyDataBuffer(ByteBuf byteBuf, NettyDataBufferAllocator allocator) {
 		Assert.notNull(byteBuf, "'byteBuf' must not be null");
+		Assert.notNull(allocator, "'allocator' must not be null");
 
 		this.byteBuf = byteBuf;
+		this.allocator = allocator;
+	}
+
+	@Override
+	public NettyDataBufferAllocator allocator() {
+		return allocator;
 	}
 
 	/**

spring-web-reactive/src/main/java/org/springframework/core/io/buffer/NettyDataBufferAllocator.java

@@ -51,19 +51,19 @@ public class NettyDataBufferAllocator implements DataBufferAllocator {
 	@Override
 	public NettyDataBuffer allocateBuffer() {
 		ByteBuf byteBuf = this.byteBufAllocator.buffer();
-		return new NettyDataBuffer(byteBuf);
+		return new NettyDataBuffer(byteBuf, this);
 	}
 
 	@Override
 	public NettyDataBuffer allocateBuffer(int initialCapacity) {
 		ByteBuf byteBuf = this.byteBufAllocator.buffer(initialCapacity);
-		return new NettyDataBuffer(byteBuf);
+		return new NettyDataBuffer(byteBuf, this);
 	}
 
 	@Override
 	public NettyDataBuffer wrap(ByteBuffer byteBuffer) {
 		ByteBuf byteBuf = Unpooled.wrappedBuffer(byteBuffer);
-		return new NettyDataBuffer(byteBuf);
+		return new NettyDataBuffer(byteBuf, this);
 	}
 
 	/**
@@ -72,7 +72,7 @@ public class NettyDataBufferAllocator implements DataBufferAllocator {
 	 * @return the wrapped buffer
 	 */
 	public NettyDataBuffer wrap(ByteBuf byteBuf) {
-		return new NettyDataBuffer(byteBuf);
+		return new NettyDataBuffer(byteBuf, this);
 	}
 
 	@Override