Introduce support for Netty 5 Buffer

This commit introduces support for Netty 5's Buffer, in the form of
Netty5DataBuffer. Because of the new API offered by Buffer, several
changes have been made to the DataBuffer API:

- CloseableDataBuffer is a simpler alternative to PooledDataBuffer, and
  implemented by Netty5DataBuffer. DataBufferUtils::release can now
  handle CloseableDataBuffer as well as PooledDataBuffer.
- PooledDataBuffer::touch has been moved into a separate interface:
  TouchableDataBuffer, which is implemented by Netty5DataBuffer.
- The capacity of DataBuffers can no longer be reduced, they can only
  grow larger. As a consequence, DataBuffer::capacity(int) has been
  deprecated, but ensureWritable (formally ensureCapacity) still exists.
- DataBuffer::slice and retainedSlice have been deprecated in favor of
  split, a new method that ensures that memory regions do not overlap.
- DataBuffer::asByteBuffer has been deprecated in favor of toByteBuffer,
  a new method that returns a copy, instead of shared data.
- DataBufferFactory::allocateBuffer has been deprecated in favor of
  allocateBuffer(int).

Closes gh-28874

Author: poutsma

spring-core/spring-core.gradle

@@ -72,6 +72,7 @@ dependencies {
 	optional("io.reactivex.rxjava3:rxjava")
 	optional("io.smallrye.reactive:mutiny")
 	optional("io.netty:netty-buffer")
+	optional("io.netty:netty5-buffer:5.0.0.Alpha4")
 	testImplementation("jakarta.annotation:jakarta.annotation-api")
 	testImplementation("jakarta.xml.bind:jakarta.xml.bind-api")
 	testImplementation("com.google.code.findbugs:jsr305")

spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -25,7 +25,6 @@
 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.PooledDataBuffer;
 import org.springframework.lang.Nullable;
 import org.springframework.util.MimeType;
 
@@ -52,7 +51,7 @@ public final Flux<DataBuffer> encode(Publisher<? extends T> inputStream, DataBuf
 		return Flux.from(inputStream)
 				.take(1)
 				.concatMap(value -> encode(value, bufferFactory, elementType, mimeType, hints))
-				.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
+				.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -51,15 +51,12 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 	public ByteBuffer decode(DataBuffer dataBuffer, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		int byteCount = dataBuffer.readableByteCount();
-		ByteBuffer copy = ByteBuffer.allocate(byteCount);
-		copy.put(dataBuffer.asByteBuffer());
-		copy.flip();
-		DataBufferUtils.release(dataBuffer);
+		ByteBuffer result = dataBuffer.toByteBuffer();
 		if (logger.isDebugEnabled()) {
-			logger.debug(Hints.getLogPrefix(hints) + "Read " + byteCount + " bytes");
+			logger.debug(Hints.getLogPrefix(hints) + "Read " + dataBuffer.readableByteCount() + " bytes");
 		}
-		return copy;
+		DataBufferUtils.release(dataBuffer);
+		return result;
 	}
 
 }

spring-core/src/main/java/org/springframework/core/codec/DataBufferDecoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -35,7 +35,7 @@
  * after they have been consumed. In addition, if using {@code Flux} or
  * {@code Mono} operators such as flatMap, reduce, and others that prefetch,
  * cache, and skip or filter out data items internally, please add
- * {@code doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release)} to the
+ * {@code doOnDiscard(DataBuffer.class, DataBufferUtils::release)} to the
  * composition chain to ensure cached data buffers are released prior to an
  * error or cancellation signal.
  *

spring-core/src/main/java/org/springframework/core/codec/Netty5BufferDecoder.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2002-2022 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
+ *
+ *      https://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 io.netty5.buffer.api.Buffer;
+import io.netty5.buffer.api.DefaultBufferAllocators;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.Netty5DataBuffer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * Decoder for {@link Buffer Buffers}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+public class Netty5BufferDecoder extends AbstractDataBufferDecoder<Buffer> {
+
+	public Netty5BufferDecoder() {
+		super(MimeTypeUtils.ALL);
+	}
+
+
+	@Override
+	public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType) {
+		return (Buffer.class.isAssignableFrom(elementType.toClass()) &&
+				super.canDecode(elementType, mimeType));
+	}
+
+	@Override
+	public Buffer decode(DataBuffer dataBuffer, ResolvableType elementType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		if (logger.isDebugEnabled()) {
+			logger.debug(Hints.getLogPrefix(hints) + "Read " + dataBuffer.readableByteCount() + " bytes");
+		}
+		if (dataBuffer instanceof Netty5DataBuffer netty5DataBuffer) {
+			return netty5DataBuffer.getNativeBuffer();
+		}
+		byte[] bytes = new byte[dataBuffer.readableByteCount()];
+		dataBuffer.read(bytes);
+		Buffer buffer = DefaultBufferAllocators.preferredAllocator().copyOf(bytes);
+		DataBufferUtils.release(dataBuffer);
+		return buffer;
+	}
+
+}

spring-core/src/main/java/org/springframework/core/codec/Netty5BufferEncoder.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2002-2022 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
+ *
+ *      https://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 io.netty5.buffer.api.Buffer;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+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.Netty5DataBufferFactory;
+import org.springframework.lang.Nullable;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * Encoder for {@link Buffer Buffers}.
+ *
+ * @author Violeta Georgieva
+ * @since 6.0
+ */
+public class Netty5BufferEncoder extends AbstractEncoder<Buffer> {
+
+	public Netty5BufferEncoder() {
+		super(MimeTypeUtils.ALL);
+	}
+
+
+	@Override
+	public boolean canEncode(ResolvableType type, @Nullable MimeType mimeType) {
+		Class<?> clazz = type.toClass();
+		return super.canEncode(type, mimeType) && Buffer.class.isAssignableFrom(clazz);
+	}
+
+	@Override
+	public Flux<DataBuffer> encode(Publisher<? extends Buffer> inputStream,
+			DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType,
+			@Nullable Map<String, Object> hints) {
+
+		return Flux.from(inputStream).map(byteBuffer ->
+				encodeValue(byteBuffer, bufferFactory, elementType, mimeType, hints));
+	}
+
+	@Override
+	public DataBuffer encodeValue(Buffer buffer, DataBufferFactory bufferFactory, ResolvableType valueType,
+			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
+
+		if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
+			String logPrefix = Hints.getLogPrefix(hints);
+			logger.debug(logPrefix + "Writing " + buffer.readableBytes() + " bytes");
+		}
+		if (bufferFactory instanceof Netty5DataBufferFactory netty5DataBufferFactory) {
+			return netty5DataBufferFactory.wrap(buffer);
+		}
+		byte[] bytes = new byte[buffer.readableBytes()];
+		buffer.readBytes(bytes, 0, bytes.length);
+		buffer.close();
+		return bufferFactory.wrap(bytes);
+	}
+}

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

@@ -35,7 +35,6 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.LimitedDataBufferList;
-import org.springframework.core.io.buffer.PooledDataBuffer;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -127,7 +126,7 @@ public Flux<String> decode(Publisher<DataBuffer> input, ResolvableType elementTy
 					return Mono.just(lastBuffer);
 				}))
 				.doOnTerminate(chunks::releaseAndClear)
-				.doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release)
+				.doOnDiscard(DataBuffer.class, DataBufferUtils::release)
 				.map(buffer -> decode(buffer, elementType, mimeType, hints));
 	}
 
@@ -153,26 +152,26 @@ private Collection<DataBuffer> processDataBuffer(
 					DataBufferUtils.retain(buffer); // retain after add (may raise DataBufferLimitException)
 					break;
 				}
-				int startIndex = buffer.readPosition();
-				int length = (endIndex - startIndex + 1);
-				DataBuffer slice = buffer.retainedSlice(startIndex, length);
-				result = (result != null ? result : new ArrayList<>());
+				DataBuffer split = buffer.split(endIndex + 1);
+				if (result == null) {
+					result = new ArrayList<>();
+				}
+				int delimiterLength = matcher.delimiter().length;
 				if (chunks.isEmpty()) {
 					if (this.stripDelimiter) {
-						slice.writePosition(slice.writePosition() - matcher.delimiter().length);
+						split.writePosition(split.writePosition() - delimiterLength);
 					}
-					result.add(slice);
+					result.add(split);
 				}
 				else {
-					chunks.add(slice);
+					chunks.add(split);
 					DataBuffer joined = buffer.factory().join(chunks);
 					if (this.stripDelimiter) {
-						joined.writePosition(joined.writePosition() - matcher.delimiter().length);
+						joined.writePosition(joined.writePosition() - delimiterLength);
 					}
 					result.add(joined);
 					chunks.clear();
 				}
-				buffer.readPosition(endIndex + 1);
 			}
 			while (buffer.readableByteCount() > 0);
 			return (result != null ? result : Collections.emptyList());
@@ -187,7 +186,7 @@ public String decode(DataBuffer dataBuffer, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
 		Charset charset = getCharset(mimeType);
-		CharBuffer charBuffer = charset.decode(dataBuffer.asByteBuffer());
+		CharBuffer charBuffer = charset.decode(dataBuffer.toByteBuffer());
 		DataBufferUtils.release(dataBuffer);
 		String value = charBuffer.toString();
 		LogFormatUtils.traceDebug(logger, traceOn -> {

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2022 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
+ *
+ *      https://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.io.buffer;
+
+/**
+ * Extension of {@link DataBuffer} that allows for buffers that can be used
+ * in a {@code try}-with-resources statement.
+
+ * @author Arjen Poutsma
+ * @since 6.0
+ */
+public interface CloseableDataBuffer extends DataBuffer, AutoCloseable {
+
+	/**
+	 * Closes this data buffer, freeing any resources.
+	 * @throws IllegalStateException if this buffer has already been closed
+	 */
+	@Override
+	void close();
+
+}