GH-3207: RSocket inbound: decode each flux item

Fixes #3207

Previously an incoming RSocket Publisher has been decoded as a single unit
leading to extra work on the client side, e.g. a delimiter has to be provided
to treat each payload item as independent

* To have a consistency with Spring Messaging and its `PayloadMethodArgumentResolver`
change an `RSocketInboundGateway` to process inbound payloads as `Flux` and decode
each item independently.
* Change `RSocketDslTests` to remove delimiters and make it consistent with the regular
`RSocketRequester` client

Author: artembilan

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -219,14 +219,17 @@ private Mono<Message<?>> decodeRequestMessage(Message<?> requestMessage) {
 	@SuppressWarnings("unchecked")
 	@Nullable
 	private Object decodePayload(Message<?> requestMessage) {
-		ResolvableType elementType = this.requestElementType;
+		ResolvableType elementType;
 		MimeType mimeType = requestMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class);
-		if (elementType == null) {
+		if (this.requestElementType == null) {
 			elementType =
 					mimeType != null && "text".equals(mimeType.getType())
 							? ResolvableType.forClass(String.class)
 							: ResolvableType.forClass(byte[].class);
 		}
+		else {
+			elementType = this.requestElementType;
+		}
 
 		Object payload = requestMessage.getPayload();
 
@@ -236,7 +239,14 @@ private Object decodePayload(Message<?> requestMessage) {
 			return decoder.decode((DataBuffer) payload, elementType, mimeType, null);
 		}
 		else {
-			return decoder.decode((Publisher<DataBuffer>) payload, elementType, mimeType, null);
+			return Flux.from((Publisher<DataBuffer>) payload)
+					.handle((buffer, synchronousSink) -> {
+						Object value = decoder.decode(buffer, elementType, mimeType, null);
+						if (value == null) {
+							value = buffer;
+						}
+						synchronousSink.next(value);
+					});
 		}
 	}
 

spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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,7 +51,7 @@ public class RSocketDslTests {
 
 	@Test
 	void testRsocketUpperCaseFlows() {
-		Flux<String> result = this.rsocketUpperCaseFlowFunction.apply(Flux.just("a\n", "b\n", "c\n"));
+		Flux<String> result = this.rsocketUpperCaseFlowFunction.apply(Flux.just("a", "b", "c"));
 
 		StepVerifier.create(result)
 				.expectNext("A", "B", "C")