Fix header propagation for rsocket message

This commit is contained in:
Oleg Zhurakousky
2021-03-10 15:25:50 +01:00
parent 485d6967bd
commit 2dea991c3e

View File

@@ -98,7 +98,7 @@ class RSocketListenerFunction implements Function<Message<Flux<Object>>, Publish
messageToProcess.getPayload()
.map((payload) -> {
return payload instanceof Message
? (Message) payload
? MessageBuilder.fromMessage((Message<?>) payload).copyHeadersIfAbsent(messageToProcess.getHeaders()).build()
: MessageBuilder.withPayload(payload).copyHeadersIfAbsent(messageToProcess.getHeaders()).build();
});
if (this.targetFunction.getInputType() != null && FunctionTypeUtils.isPublisher(this.targetFunction.getInputType())) {