Support decoding empty DataBuffers for Decoders that support it

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

@@ -44,6 +44,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 		return (elementType.resolve() == byte[].class && super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public byte[] decode(DataBuffer dataBuffer, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -47,6 +47,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 				super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public ByteBuffer decode(DataBuffer dataBuffer, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -56,6 +56,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 				super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public Flux<DataBuffer> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -53,6 +53,17 @@ public interface Decoder<T> {
 	 */
 	boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType);
 
+	/**
+	 * Whether the decoder supports decoding an Object of its target type from an
+	 * empty message. When it is true, the decoder will always return a non-null
+	 * value from its {@code decode} method when an empty message is decoded.
+	 * @return {@code true} if supported, {@code false} otherwise
+	 * @since 6.0.5
+	 */
+	default boolean canDecodeEmptyMessage()  {
+		return false;
+	}
+
 	/**
 	 * Decode a {@link DataBuffer} input stream into a Flux of {@code T}.
 	 * @param inputStream the {@code DataBuffer} input stream to decode

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

@@ -48,6 +48,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 				super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public Buffer decode(DataBuffer dataBuffer, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -48,6 +48,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 				super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public ByteBuf decode(DataBuffer dataBuffer, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -56,6 +56,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 				super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public Flux<Resource> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -105,6 +105,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 		return (elementType.resolve() == String.class && super.canDecode(elementType, mimeType));
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public Flux<String> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

spring-core/src/testFixtures/java/org/springframework/core/testfixture/codec/AbstractDecoderTests.java

@@ -36,6 +36,7 @@
 import org.springframework.util.MimeType;
 
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.springframework.core.io.buffer.DataBufferUtils.release;
 
 /**
  * Abstract base class for {@link Decoder} unit tests. Subclasses need to implement
@@ -114,6 +115,7 @@ protected <T> void testDecodeAll(Publisher<DataBuffer> input, Class<? extends T>
 	 *     <li>{@link #testDecodeError(Publisher, ResolvableType, MimeType, Map)}</li>
 	 *     <li>{@link #testDecodeCancel(Publisher, ResolvableType, MimeType, Map)}</li>
 	 *     <li>{@link #testDecodeEmpty(ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeEmptyMessage(ResolvableType, MimeType, Map)}</li>
 	 * </ul>
 	 *
 	 * @param input the input to be provided to the decoder
@@ -131,6 +133,7 @@ protected <T> void testDecodeAll(Publisher<DataBuffer> input, ResolvableType out
 		testDecodeError(input, outputType, mimeType, hints);
 		testDecodeCancel(input, outputType, mimeType, hints);
 		testDecodeEmpty(outputType, mimeType, hints);
+		testDecodeEmptyMessage(outputType, mimeType, hints);
 	}
 
 	/**
@@ -258,6 +261,25 @@ protected void testDecodeEmpty(ResolvableType outputType, @Nullable MimeType mim
 		StepVerifier.create(result).verifyComplete();
 	}
 
+	/**
+	 * Test a {@link Decoder#decode decode} scenario where the input stream is an empty buffer.
+	 * The output is expected to be filled when the decoder supports it.
+	 *
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 */
+	protected void testDecodeEmptyMessage(ResolvableType outputType, MimeType mimeType, Map<String, Object> hints) {
+		if (!this.decoder.canDecodeEmptyMessage()) {
+			return;
+		}
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(0);
+		Object result = this.decoder.decode(buffer, outputType, mimeType, hints);
+		releaseDataBufferIfIdentical(buffer, result);
+		Assert.notNull(result, "result expected to be non null");
+		Assert.isAssignable(outputType.toClass(), result.getClass(), "result not of specified type");
+	}
+
 	// Mono
 
 	/**
@@ -289,6 +311,7 @@ protected <T> void testDecodeToMonoAll(Publisher<DataBuffer> input,
 	 *     <li>{@link #testDecodeToMonoError(Publisher, ResolvableType, MimeType, Map)}</li>
 	 *     <li>{@link #testDecodeToMonoCancel(Publisher, ResolvableType, MimeType, Map)}</li>
 	 *     <li>{@link #testDecodeToMonoEmpty(ResolvableType, MimeType, Map)}</li>
+	 *     <li>{@link #testDecodeToMonoEmptyMessage(ResolvableType, MimeType, Map)}</li>
 	 * </ul>
 	 *
 	 * @param input the input to be provided to the decoder
@@ -306,6 +329,7 @@ protected <T> void testDecodeToMonoAll(Publisher<DataBuffer> input, ResolvableTy
 		testDecodeToMonoError(input, outputType, mimeType, hints);
 		testDecodeToMonoCancel(input, outputType, mimeType, hints);
 		testDecodeToMonoEmpty(outputType, mimeType, hints);
+		testDecodeToMonoEmptyMessage(outputType, mimeType, hints);
 	}
 
 	/**
@@ -419,6 +443,32 @@ protected void testDecodeToMonoEmpty(ResolvableType outputType, @Nullable MimeTy
 		StepVerifier.create(result).verifyComplete();
 	}
 
+	/**
+	 * Test a {@link Decoder#decodeToMono decode} scenario where the input stream is an empty buffer.
+	 * The output is expected to be filled when the decoder supports it.
+	 *
+	 * @param outputType the desired output type
+	 * @param mimeType the mime type to use for decoding. May be {@code null}.
+	 * @param hints the hints used for decoding. May be {@code null}.
+	 */
+	protected void testDecodeToMonoEmptyMessage(ResolvableType outputType, @Nullable MimeType mimeType,
+			@Nullable Map<String, Object> hints) {
+
+		if (!this.decoder.canDecodeEmptyMessage()) {
+			return;
+		}
+
+		Flux<DataBuffer> source = Flux.range(0, 2)
+				.map(i -> this.bufferFactory.allocateBuffer(0));
+
+		Mono<?> result = this.decoder.decodeToMono(source, outputType, mimeType, hints)
+						.doOnNext(this::releaseIfDataBuffer);
+
+		StepVerifier.create(result)
+				.expectNextMatches(next -> outputType.toClass().isInstance(next))
+				.verifyComplete();
+	}
+
 	/**
 	 * Creates a deferred {@link DataBuffer} containing the given bytes.
 	 * @param bytes the bytes that are to be stored in the buffer
@@ -432,6 +482,27 @@ protected Mono<DataBuffer> dataBuffer(byte[] bytes) {
 		});
 	}
 
+	/**
+	 * If {@code value} is referentially identical to {@code buffer}, release it.
+	 * @param buffer the {@link DataBuffer} that is compared
+	 * @param value  the {@link Object} that is compared
+	 */
+	private void releaseDataBufferIfIdentical(DataBuffer buffer, Object value) {
+		if (buffer == value) {
+			release(buffer);
+		}
+	}
+
+	/**
+	 * If {@code value} is a {@link DataBuffer}, release it.
+	 * @param value the {@link Object} that is checked
+	 */
+	private void releaseIfDataBuffer(Object value) {
+		if (value instanceof DataBuffer) {
+			release((DataBuffer) value);
+		}
+	}
+
 	/**
 	 * Exception used in {@link #testDecodeError} and {@link #testDecodeToMonoError}
 	 */

spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/reactive/PayloadMethodArgumentResolver.java

@@ -231,7 +231,7 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 			if (decoder.canDecode(elementType, mimeType)) {
 				if (adapter != null && adapter.isMultiValue()) {
 					Flux<?> flux = content
-							.filter(this::nonEmptyDataBuffer)
+							.filter(dataBuffer -> nonEmptyDataBuffer(dataBuffer, decoder))
 							.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
 							.onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
@@ -245,7 +245,7 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 				else {
 					// Single-value (with or without reactive type wrapper)
 					Mono<?> mono = content.next()
-							.filter(this::nonEmptyDataBuffer)
+							.filter(dataBuffer -> nonEmptyDataBuffer(dataBuffer, decoder))
 							.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
 							.onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
@@ -263,7 +263,10 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 				message, parameter, "Cannot decode to [" + targetType + "]" + message));
 	}
 
-	private boolean nonEmptyDataBuffer(DataBuffer buffer) {
+	private boolean nonEmptyDataBuffer(DataBuffer buffer, Decoder<?> decoder) {
+		if (decoder.canDecodeEmptyMessage()) {
+			return true;
+		}
 		if (buffer.readableByteCount() > 0) {
 			return true;
 		}

spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java

@@ -167,7 +167,7 @@ public void echoChannel() {
 	@Test // gh-26344
 	public void echoChannelWithEmptyInput() {
 		Flux<String> result = requester.route("echo-channel-empty").data(Flux.empty()).retrieveFlux(String.class);
-		StepVerifier.create(result).verifyComplete();
+		StepVerifier.create(result).expectNext(" echoed").verifyComplete();
 	}
 
 	@Test

spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java

@@ -125,6 +125,11 @@ public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType
 		return Message.class.isAssignableFrom(elementType.toClass()) && supportsMimeType(mimeType);
 	}
 
+	@Override
+	public boolean canDecodeEmptyMessage() {
+		return true;
+	}
+
 	@Override
 	public Flux<Message> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

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

@@ -221,6 +221,11 @@ public void exceedMaxSize() {
 		testDecode(input, Msg.class, step -> step.verifyError(DecodingException.class));
 	}
 
+	@Test
+	public void decodeEmpty() {
+		testDecodeEmptyMessage(ResolvableType.forClass(Msg.class), null, null);
+	}
+
 	private Mono<DataBuffer> dataBuffer(Msg msg) {
 		return Mono.fromCallable(() -> {
 			byte[] bytes = msg.toByteArray();