forward "live" CommandResponses received in InboundDispatchingSink to…

… edge commandForwarder

* let headers by Signal take precedence from inbound payload mapping

Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch.io>

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/InboundDispatchingSink.java

@@ -306,7 +306,10 @@ public Optional<Signal<?>> onMapped(final String mapperId,
 
         final DittoHeaders mappedHeaders =
                 applyInboundHeaderMapping(signal, incomingMessage, authorizationContext,
-                        mappedInboundMessage.getTopicPath(), incomingMessage.getInternalHeaders());
+                        mappedInboundMessage.getTopicPath(), incomingMessage.getInternalHeaders())
+                        .toBuilder()
+                        .putHeaders(signal.getDittoHeaders()) // headers from signal take precedence
+                        .build();
 
         logger.withCorrelationId(mappedHeaders).info("onMapped mappedHeaders {}", mappedHeaders);
 
@@ -501,9 +504,10 @@ private PartialFunction<Signal<?>, Stream<IncomingSignal>> dispatchResponsesAndS
                                 forwardAcknowledgement(ack, declaredAckLabels, outcomes))
                         .match(Acknowledgements.class, acks ->
                                 forwardAcknowledgements(acks, declaredAckLabels, outcomes))
-                        .match(CommandResponse.class, ProtocolAdapter::isLiveSignal, liveResponse ->
-                                forwardToClientActor(liveResponse, ActorRef.noSender())
-                        )
+                        .match(CommandResponse.class, ProtocolAdapter::isLiveSignal, liveResponse -> {
+                                proxyActor.tell(liveResponse, ActorRef.noSender());
+                                return Stream.empty();
+                        })
                         .match(CreateSubscription.class, cmd -> forwardToConnectionActor(cmd, sender))
                         .match(WithSubscriptionId.class, cmd -> forwardToClientActor(cmd, sender))
                         .matchAny(baseSignal -> ackregatorStarter.preprocess(baseSignal,
@@ -638,7 +642,6 @@ private void handleErrorDuringStartingOfAckregator(final DittoRuntimeException e
      * Only special Signals must be forwarded to the {@code ClientActor}:
      * <ul>
      * <li>{@code Acknowledgement}s which were received via an incoming connection source</li>
-     * <li>live {@code CommandResponse}s which were received via an incoming connection source</li>
      * <li>{@code SearchCommand}s which were received via an incoming connection source</li>
      * </ul>
      *