Fix header propagation for rsocket message

This commit is contained in:
Oleg Zhurakousky
2021-03-10 15:25:50 +01:00
parent 73164beb68
commit 3255ae7455

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())) {