Filter empty buffers PayloadMethodArgumentResolver

An empty buffer for RSocket is an empty message and while some codecs
such as StringDecoder may be able turn that into something such as an
empty String, for other formats such as JSON it is invalid input.

Closes gh-26344

Author: rstoyanchev

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -38,6 +38,7 @@
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -232,6 +233,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)
 							.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
 							.onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
@@ -245,6 +247,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)
 							.map(buffer -> decoder.decode(buffer, elementType, mimeType, hints))
 							.onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex)));
 					if (isContentRequired) {
@@ -262,6 +265,14 @@ private Mono<Object> decodeContent(MethodParameter parameter, Message<?> message
 				message, parameter, "Cannot decode to [" + targetType + "]" + message));
 	}
 
+	private boolean nonEmptyDataBuffer(DataBuffer buffer) {
+		if (buffer.readableByteCount() > 0) {
+			return true;
+		}
+		DataBufferUtils.release(buffer);
+		return false;
+	}
+
 	private Throwable handleReadError(MethodParameter parameter, Message<?> message, Throwable ex) {
 		return ex instanceof DecodingException ?
 				new MethodArgumentResolutionException(message, parameter, "Failed to read HTTP message", ex) : ex;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -19,7 +19,6 @@
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import io.rsocket.Payload;
 import io.rsocket.RSocket;
 import io.rsocket.SocketAcceptor;
 import io.rsocket.core.RSocketServer;
@@ -43,6 +42,7 @@
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
 import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.rsocket.annotation.ConnectMapping;
 import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
 import org.springframework.stereotype.Controller;
@@ -164,6 +164,12 @@ public void echoChannel() {
 				.verify(Duration.ofSeconds(5));
 	}
 
+	@Test // gh-26344
+	public void echoChannelWithEmptyInput() {
+		Flux<String> result = requester.route("echo-channel-empty").data(Flux.empty()).retrieveFlux(String.class);
+		StepVerifier.create(result).verifyComplete();
+	}
+
 	@Test
 	public void metadataPush() {
 		Flux.just("bar", "baz")
@@ -254,6 +260,11 @@ Flux<String> echoChannel(Flux<String> payloads) {
 			return payloads.delayElements(Duration.ofMillis(10)).map(payload -> payload + " async");
 		}
 
+		@MessageMapping("echo-channel-empty")
+		Flux<String> echoChannelEmpty(@Payload(required = false) Flux<String> payloads) {
+			return payloads.map(payload -> payload + " echoed");
+		}
+
 		@MessageMapping("thrown-exception")
 		Mono<String> handleAndThrow(String payload) {
 			throw new IllegalArgumentException("Invalid input error");
@@ -338,29 +349,29 @@ public RSocket apply(RSocket rsocket) {
 		}
 
 		@Override
-		public Mono<Void> fireAndForget(Payload payload) {
+		public Mono<Void> fireAndForget(io.rsocket.Payload payload) {
 			return this.delegate.fireAndForget(payload)
 					.doOnSuccess(aVoid -> this.fireAndForgetCount.incrementAndGet());
 		}
 
 		@Override
-		public Mono<Void> metadataPush(Payload payload) {
+		public Mono<Void> metadataPush(io.rsocket.Payload payload) {
 			return this.delegate.metadataPush(payload)
 					.doOnSuccess(aVoid -> this.metadataPushCount.incrementAndGet());
 		}
 
 		@Override
-		public Mono<Payload> requestResponse(Payload payload) {
+		public Mono<io.rsocket.Payload> requestResponse(io.rsocket.Payload payload) {
 			return this.delegate.requestResponse(payload);
 		}
 
 		@Override
-		public Flux<Payload> requestStream(Payload payload) {
+		public Flux<io.rsocket.Payload> requestStream(io.rsocket.Payload payload) {
 			return this.delegate.requestStream(payload);
 		}
 
 		@Override
-		public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
+		public Flux<io.rsocket.Payload> requestChannel(Publisher<io.rsocket.Payload> payloads) {
 			return this.delegate.requestChannel(payloads);
 		}
 	}