SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleanup 'handleAndReply' logic in RSocketListenerFunction

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-03-05 09:51:59 +01:00
parent 157949df93
commit c8a0cebd1f
1 changed files with 3 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java
View File

@@ -97,16 +97,9 @@ class RSocketListenerFunction implements Function<Message<Flux<Object>>, 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);