Merge pull request #1323 from bosch-io/bugfix/response-receiver-cache

thjaeckle · web-flow · commit ad0ef9f75dc2 · 2022-03-11T13:35:37.000+01:00
Invalidate ResponseReceiver cache after a response is received.
diff --git a/concierge/service/src/main/java/org/eclipse/ditto/concierge/service/enforcement/LiveSignalEnforcement.java b/concierge/service/src/main/java/org/eclipse/ditto/concierge/service/enforcement/LiveSignalEnforcement.java
@@ -256,6 +256,7 @@ private CompletionStage<Contextual<WithDittoHeaders>> returnCommandResponseConte
                         final var receiver = responseReceiverEntry.get();
                         log().info("Scheduling CommandResponse <{}> to original sender <{}>", liveResponse, receiver);
                         commandResponseContextual = withMessageToReceiver(liveResponse, receiver);
+                        responseReceiverCache.invalidate(correlationId);
                     } else {
                         log().info("Got <{}> with unknown correlation ID: <{}>", liveResponse.getType(), correlationId);
                         commandResponseContextual = withMessageToReceiver(null, null);
@@ -396,13 +397,16 @@ private <T extends Signal<?>, S extends T> CompletionStage<Contextual<WithDittoH
             final DistributedPub<T> pub) {
 
         // using pub/sub to publish the command to any interested parties (e.g. a Websocket):
-        log(signal).debug("Publish message to pub-sub: <{}>", signal);
 
         if (enforcementConfig.shouldDispatchGlobally(signal)) {
             return responseReceiverCache.insertResponseReceiverConflictFree(signal,
                     newSignal -> sender(),
-                    (newSignal, receiver) -> publishSignal(newSignal, ackExtractor, pub));
+                    (newSignal, receiver) -> {
+                        log(newSignal).debug("Publish message to pub-sub: <{}>", newSignal);
+                        return publishSignal(newSignal, ackExtractor, pub);
+                    });
         } else {
+            log(signal).debug("Publish message to pub-sub: <{}>", signal);
             return CompletableFuture.completedStage(publishSignal(signal, ackExtractor, pub));
         }
     }
diff --git a/concierge/service/src/main/java/org/eclipse/ditto/concierge/service/enforcement/ResponseReceiverCache.java b/concierge/service/src/main/java/org/eclipse/ditto/concierge/service/enforcement/ResponseReceiverCache.java
@@ -158,6 +158,22 @@ public CompletableFuture<Optional<ActorRef>> get(final CharSequence correlationI
         return cache.get(CorrelationIdKey.forCacheRetrieval(correlationIdString));
     }
 
+    /**
+     * Invalidates the cached response receiver for the specified correlation ID argument.
+     *
+     * @param correlationId the correlation ID to invalidate the cached response receiver for.
+     * @throws NullPointerException if {@code correlationId} is {@code null}.
+     * @throws IllegalArgumentException if {@code correlationId} is empty or blank.
+     */
+    public void invalidate(final CharSequence correlationId) {
+        final var correlationIdString = checkNotNull(correlationId, "correlationId").toString();
+        ConditionChecker.checkArgument(correlationIdString,
+                Predicate.not(String::isBlank),
+                () -> "The correlationId must not be blank.");
+
+        cache.invalidate(CorrelationIdKey.forCacheRetrieval(correlationIdString));
+    }
+
     /**
      * Insert a response receiver for a live or message command.
      *
diff --git a/concierge/service/src/test/java/org/eclipse/ditto/concierge/service/enforcement/LiveSignalEnforcementTest.java b/concierge/service/src/test/java/org/eclipse/ditto/concierge/service/enforcement/LiveSignalEnforcementTest.java
@@ -26,7 +26,9 @@
 import org.eclipse.ditto.base.model.auth.AuthorizationContext;
 import org.eclipse.ditto.base.model.auth.AuthorizationSubject;
 import org.eclipse.ditto.base.model.auth.DittoAuthorizationContextType;
+import org.eclipse.ditto.base.model.entity.id.WithEntityId;
 import org.eclipse.ditto.base.model.headers.DittoHeaders;
+import org.eclipse.ditto.base.model.headers.WithDittoHeaders;
 import org.eclipse.ditto.base.model.json.FieldType;
 import org.eclipse.ditto.base.model.json.JsonSchemaVersion;
 import org.eclipse.ditto.base.model.signals.events.Event;
@@ -119,7 +121,7 @@ public void rejectMessageCommandByPolicy() {
             mockEntitiesActorInstance.setReply(TestSetup.POLICY_SUDO, sudoRetrievePolicyResponse);
 
             final ActorRef underTest = newEnforcerActor(getRef());
-            underTest.tell(thingMessageCommand(), getRef());
+            underTest.tell(thingMessageCommand("abc"), getRef());
             TestSetup.fishForMsgClass(this, MessageSendNotAllowedException.class);
         }};
     }
@@ -253,6 +255,118 @@ public void retrieveLiveThingCommandAndResponseByPolicy() {
         }};
     }
 
+    @Test
+    public void correlationIdSameAfterResponseSuccessful() {
+        final PolicyId policyId = PolicyId.of("policy", "id");
+        final JsonObject thingWithPolicy = newThingWithPolicyId(policyId);
+        final JsonObject policy = PoliciesModelFactory.newPolicyBuilder(policyId)
+                .setRevision(1L)
+                .forLabel("authorize-self")
+                .setSubject(GOOGLE, SUBJECT_ID)
+                .setGrantedPermissions(PoliciesResourceType.thingResource("/"),
+                        Permissions.newInstance(Permission.READ, Permission.WRITE))
+                .setRevokedPermissions(PoliciesResourceType.thingResource("/features/x/properties/key2"),
+                        Permissions.newInstance(Permission.READ))
+                .build()
+                .toJson(FieldType.all());
+        final SudoRetrieveThingResponse sudoRetrieveThingResponse =
+                SudoRetrieveThingResponse.of(thingWithPolicy, DittoHeaders.empty());
+        final SudoRetrievePolicyResponse sudoRetrievePolicyResponse =
+                SudoRetrievePolicyResponse.of(policyId, policy, DittoHeaders.empty());
+
+        new TestKit(system) {{
+            mockEntitiesActorInstance.setReply(TestSetup.THING_SUDO, sudoRetrieveThingResponse);
+            mockEntitiesActorInstance.setReply(TestSetup.POLICY_SUDO, sudoRetrievePolicyResponse);
+
+            final ActorRef underTest = newEnforcerActor(getRef());
+
+            final DittoHeaders headers = headers();
+            final ThingCommand<?> read = getRetrieveThingCommand(headers);
+
+            underTest.tell(read, getRef());
+
+            final var responseHeaders = headers.toBuilder()
+                    .authorizationContext(AuthorizationContext.newInstance(
+                            DittoAuthorizationContextType.PRE_AUTHENTICATED_CONNECTION,
+                            AuthorizationSubject.newInstance("myIssuer:mySubject")))
+                    .build();
+
+            final ThingCommandResponse<?> readResponse = getRetrieveThingResponse(responseHeaders);
+
+            // Second message right after the response for the first was sent, should have the same correlation-id (Not suffixed).
+            underTest.tell(readResponse, getRef());
+            final RetrieveThingResponse retrieveThingResponse =
+                    TestSetup.fishForMsgClass(this, RetrieveThingResponse.class);
+            assertThat(retrieveThingResponse.getDittoHeaders().getCorrelationId()).isEqualTo(
+                    read.getDittoHeaders().getCorrelationId());
+
+            underTest.tell(read, getRef());
+
+            underTest.tell(readResponse, getRef());
+            final RetrieveThingResponse retrieveThingResponse2 =
+                    TestSetup.fishForMsgClass(this, RetrieveThingResponse.class);
+            assertThat(retrieveThingResponse2.getDittoHeaders().getCorrelationId()).isEqualTo(
+                    read.getDittoHeaders().getCorrelationId());
+        }};
+    }
+
+    @Test
+    public void correlationIdDifferentInCaseOfConflict() {
+        final PolicyId policyId = PolicyId.of("policy:id");
+        final JsonObject thingWithPolicy = newThingWithPolicyId(policyId);
+        final JsonObject policy = PoliciesModelFactory.newPolicyBuilder(policyId)
+                .setRevision(1L)
+                .forLabel("authorize-self")
+                .setSubject(GOOGLE, SUBJECT_ID)
+                .setGrantedPermissions(PoliciesResourceType.messageResource(JsonPointer.empty()),
+                        Permissions.newInstance(Permission.READ, Permission.WRITE))
+                .build()
+                .toJson(FieldType.all());
+        final SudoRetrieveThingResponse sudoRetrieveThingResponse =
+                SudoRetrieveThingResponse.of(thingWithPolicy, DittoHeaders.empty());
+        final SudoRetrievePolicyResponse sudoRetrievePolicyResponse =
+                SudoRetrievePolicyResponse.of(policyId, policy, DittoHeaders.empty());
+
+        new TestKit(system) {{
+            mockEntitiesActorInstance.setReply(TestSetup.THING_SUDO, sudoRetrieveThingResponse);
+            mockEntitiesActorInstance.setReply(TestSetup.POLICY_SUDO, sudoRetrievePolicyResponse);
+
+            final ActorRef underTest = newEnforcerActor(getRef());
+
+            final MessageCommand<?, ?> message = thingMessageCommand("abc");
+
+            underTest.tell(message, getRef());
+            final DistributedPubSubMediator.Publish firstPublishRead =
+                    pubSubMediatorProbe.expectMsgClass(DistributedPubSubMediator.Publish.class);
+            assertThat(firstPublishRead.topic()).isEqualTo(StreamingType.MESSAGES.getDistributedPubSubTopic());
+            assertThat(firstPublishRead.msg()).isInstanceOf(MessageCommand.class);
+            assertThat((CharSequence) ((WithEntityId) firstPublishRead.msg()).getEntityId()).isEqualTo(
+                    message.getEntityId());
+            assertThat((CharSequence) ((WithDittoHeaders) firstPublishRead.msg()).getDittoHeaders()
+                    .getCorrelationId()
+                    .orElseThrow()).isEqualTo(
+                    message.getDittoHeaders().getCorrelationId().orElseThrow());
+
+            underTest.tell(message, getRef());
+            final DistributedPubSubMediator.Publish secondPublishRead =
+                    pubSubMediatorProbe.expectMsgClass(DistributedPubSubMediator.Publish.class);
+            assertThat(secondPublishRead.topic()).isEqualTo(StreamingType.MESSAGES.getDistributedPubSubTopic());
+            assertThat(secondPublishRead.msg()).isInstanceOf(MessageCommand.class);
+            assertThat((CharSequence) ((WithEntityId) secondPublishRead.msg()).getEntityId()).isEqualTo(
+                    message.getEntityId());
+            // Assure second command has suffixed correlation-id, because of conflict with first command.
+            assertThat((CharSequence) ((WithDittoHeaders) secondPublishRead.msg()).getDittoHeaders()
+                    .getCorrelationId()
+                    .orElseThrow()).startsWith(
+                    message.getDittoHeaders().getCorrelationId().orElseThrow());
+            assertThat((CharSequence) ((WithDittoHeaders) secondPublishRead.msg()).getDittoHeaders()
+                    .getCorrelationId()
+                    .orElseThrow()).isNotEqualTo(
+                    message.getDittoHeaders().getCorrelationId().orElseThrow());
+
+        }};
+    }
+
     @Test
     public void acceptMessageCommandByPolicy() {
         final PolicyId policyId = PolicyId.of("policy:id");
@@ -276,7 +390,7 @@ public void acceptMessageCommandByPolicy() {
 
             final ActorRef underTest = newEnforcerActor(getRef());
 
-            final MessageCommand<?, ?> msgCommand = thingMessageCommand();
+            final MessageCommand<?, ?> msgCommand = thingMessageCommand("abc");
             mockEntitiesActorInstance.setReply(msgCommand);
             underTest.tell(msgCommand, getRef());
             final DistributedPubSubMediator.Publish publish =
@@ -419,11 +533,12 @@ private static ThingCommand<?> getModifyFeatureCommand(final DittoHeaders header
         return ModifyFeature.of(TestSetup.THING_ID, TestSetup.FEATURE, headers);
     }
 
-    private static MessageCommand<?, ?> thingMessageCommand() {
+    private static MessageCommand<?, ?> thingMessageCommand(final String correlationId) {
         final Message<Object> message = Message.newBuilder(
-                        MessageBuilder.newHeadersBuilder(MessageDirection.TO, TestSetup.THING_ID, "my-subject")
-                                .contentType("text/plain")
-                                .build())
+                MessageBuilder.newHeadersBuilder(MessageDirection.TO, TestSetup.THING_ID, "my-subject")
+                        .contentType("text/plain")
+                        .correlationId(correlationId)
+                        .build())
                 .payload("Hello you!")
                 .build();
         return SendThingMessage.of(TestSetup.THING_ID, message, headers());
@@ -441,10 +556,10 @@ private static ThingEvent<?> liveEventRevoked() {
 
     private static MessageCommand<?, ?> featureMessageCommand() {
         final Message<?> message = Message.newBuilder(
-                        MessageBuilder.newHeadersBuilder(MessageDirection.TO, TestSetup.THING_ID, "my-subject")
-                                .contentType("text/plain")
-                                .featureId("foo")
-                                .build())
+                MessageBuilder.newHeadersBuilder(MessageDirection.TO, TestSetup.THING_ID, "my-subject")
+                        .contentType("text/plain")
+                        .featureId("foo")
+                        .build())
                 .payload("Hello you!")
                 .build();
         return SendFeatureMessage.of(TestSetup.THING_ID, "foo", message, headers());
diff --git a/connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessor.java b/connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessor.java
@@ -26,6 +26,7 @@
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.base.model.headers.DittoHeaderDefinition;
 import org.eclipse.ditto.base.model.headers.DittoHeaders;
+import org.eclipse.ditto.base.model.headers.WithDittoHeaders;
 import org.eclipse.ditto.base.model.signals.Signal;
 import org.eclipse.ditto.connectivity.api.ExternalMessage;
 import org.eclipse.ditto.connectivity.api.ExternalMessageFactory;
@@ -184,17 +185,33 @@ private List<MappingOutcome<OutboundSignal.Mapped>> processMappableSignals(final
                 .map(extra -> ProtocolFactory.setExtra(adaptableWithoutExtra, extra))
                 .orElse(adaptableWithoutExtra);
 
+        final var adaptableWithInternalCorrelationId = mappableSignals.stream()
+                .findFirst()
+                .map(signal -> setInternalCorrelationIdToAdaptable(adaptable, signal.getSource()))
+                .orElse(adaptable);
+
         return timer.overall(() -> mappableSignals.stream()
                 .flatMap(mappableSignal -> {
                     final Signal<?> source = mappableSignal.getSource();
                     final List<Target> targets = mappableSignal.getTargets();
                     final List<MessageMapper> mappers = getMappers(mappableSignal.getPayloadMapping());
-                    logger.withCorrelationId(adaptable)
+                    logger.withCorrelationId(adaptableWithInternalCorrelationId)
                             .debug("Resolved mappers for message {} to targets {}: {}", source, targets, mappers);
                     // convert messages in the order of payload mapping and forward to result handler
-                    return mappers.stream().flatMap(mapper -> runMapper(mappableSignal, adaptable, mapper, timer));
+                    return mappers.stream()
+                            .flatMap(mapper -> runMapper(mappableSignal, adaptableWithInternalCorrelationId, mapper,
+                                    timer));
                 })
-                .collect(Collectors.toList()));
+                .toList());
+    }
+
+    private static Adaptable setInternalCorrelationIdToAdaptable(final Adaptable adaptable,
+            final WithDittoHeaders internalSignal) {
+        final var optionalCorrelationId = internalSignal.getDittoHeaders().getCorrelationId();
+        final Adaptable result;
+        result = optionalCorrelationId.map(s -> adaptable.setDittoHeaders(
+                adaptable.getDittoHeaders().toBuilder().correlationId(s).build())).orElse(adaptable);
+        return result;
     }
 
     private Stream<MappingOutcome<OutboundSignal.Mapped>> runMapper(final OutboundSignal.Mappable outboundSignal,
diff --git a/connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessorTest.java b/connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessorTest.java
diff --git a/connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/httppush/HttpPushClientActorTest.java b/connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/httppush/HttpPushClientActorTest.java
