Add DataBuffer.retainedSlice

Add method retainedSlice to the DataBuffer, defaulting to using
DataBufferUtils for retain, but allowing for ByteBuf specific override.

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

@@ -289,6 +289,23 @@ public interface DataBuffer {
 	 */
 	DataBuffer slice(int index, int length);
 
+	/**
+	 * Create a new {@code DataBuffer} whose contents is a shared, retained subsequence of this
+	 * data buffer's content.  Data between this data buffer and the returned buffer is
+	 * shared; though changes in the returned buffer's position will not be reflected
+	 * in the reading nor writing position of this data buffer.
+	 * <p><strong>Note</strong> that unlike {@link #slice(int, int)}, this method
+	 * <strong>will</strong> call {@link DataBufferUtils#retain(DataBuffer)} (or equivalent) on the
+	 * resulting slice.
+	 * @param index the index at which to start the slice
+	 * @param length the length of the slice
+	 * @return the specified, retained slice of this data buffer
+	 * @since 5.2
+	 */
+	default DataBuffer retainedSlice(int index, int length) {
+		return DataBufferUtils.retain(slice(index, length));
+	}
+
 	/**
 	 * Expose this buffer's bytes as a {@link ByteBuffer}. Data between this
 	 * {@code DataBuffer} and the returned {@code ByteBuffer} is shared; though

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

@@ -261,6 +261,12 @@ public class NettyDataBuffer implements PooledDataBuffer {
 		return new NettyDataBuffer(slice, this.dataBufferFactory);
 	}
 
+	@Override
+	public NettyDataBuffer retainedSlice(int index, int length) {
+		ByteBuf slice = this.byteBuf.retainedSlice(index, length);
+		return new NettyDataBuffer(slice, this.dataBufferFactory);
+	}
+
 	@Override
 	public ByteBuffer asByteBuffer() {
 		return this.byteBuf.nioBuffer();

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

@@ -578,6 +578,37 @@ public class DataBufferTests extends AbstractDataBufferAllocatingTestCase {
 		release(buffer);
 	}
 
+	@Test
+	public void retainedSlice() {
+		DataBuffer buffer = createDataBuffer(3);
+		buffer.write(new byte[]{'a', 'b'});
+
+		DataBuffer slice = buffer.retainedSlice(1, 2);
+		assertEquals(2, slice.readableByteCount());
+		try {
+			slice.write((byte) 0);
+			fail("Exception expected");
+		}
+		catch (Exception ignored) {
+		}
+		buffer.write((byte) 'c');
+
+		assertEquals(3, buffer.readableByteCount());
+		byte[] result = new byte[3];
+		buffer.read(result);
+
+		assertArrayEquals(new byte[]{'a', 'b', 'c'}, result);
+
+		assertEquals(2, slice.readableByteCount());
+		result = new byte[2];
+		slice.read(result);
+
+		assertArrayEquals(new byte[]{'b', 'c'}, result);
+
+
+		release(buffer, slice);
+	}
+
 	@Test
 	public void spr16351() {
 		DataBuffer buffer = createDataBuffer(6);