From c8a0cebd1f2d9c632f1d669e38de2a1b3d09b4f9 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 5 Mar 2021 09:51:59 +0100 Subject: [PATCH] Cleanup 'handleAndReply' logic in RSocketListenerFunction --- .../function/rsocket/RSocketListenerFunction.java | 13 +++---------- 1 file changed, 3 insertions(+), 10 deletions(-) diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java index e53806ba8..09c0af0df 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java @@ -97,16 +97,9 @@ class RSocketListenerFunction implements Function>, Publish Flux dataFlux = messageToProcess.getPayload() .map((payload) -> { - if (payload instanceof Message) { - return MessageBuilder.fromMessage((Message) payload).copyHeadersIfAbsent(messageToProcess.getHeaders()).build(); - } - else { - return MessageBuilder.withPayload(payload).copyHeadersIfAbsent(messageToProcess.getHeaders()).build(); - } -// if (!(payload instanceof Message)) { -// payload = MessageBuilder.createMessage(payload, messageToProcess.getHeaders()); -// } -// return payload; + return payload instanceof Message + ? (Message) payload + : MessageBuilder.withPayload(payload).copyHeadersIfAbsent(messageToProcess.getHeaders()).build(); }); if (this.targetFunction.getInputType() != null && FunctionTypeUtils.isPublisher(this.targetFunction.getInputType())) { dataFlux = dataFlux.transform((Function) this.targetFunction);