MessageMappingProcessorActor extends AbstractGraphActor<MessageMappingProcessorActor.OutboundSignalWithId, OutboundSignal> { private static Optional<AuthorizationContext> getAuthorizationContext(final ExternalMessage externalMessage) { final Either<RuntimeException, AuthorizationContext> result = getAuthorizationContextAsEither(externalMessage); if (result.isRight()) { return Optional.of(result.right().get()); } else { return Optional.empty(); } } @SuppressWarnings("unused") private  MessageMappingProcessorActor(final ActorRef proxyActor,
            final ActorRef clientActor,
            final MessageMappingProcessor messageMappingProcessor,
            final Connection connection,
            final ActorRef connectionActor,
            final int processorPoolSize); static Props props(final ActorRef proxyActor,
            final ActorRef clientActor,
            final MessageMappingProcessor processor,
            final Connection connection,
            final ActorRef connectionActor,
            final int processorPoolSize); static final String ACTOR_NAME; }
@Test public void testSignalEnrichment() { final TestProbe proxyActorProbe = TestProbe.apply("mockProxyActorProbe", actorSystem); setUpProxyActor(proxyActorProbe.ref()); new TestKit(actorSystem) {{ final ActorRef underTest = createMessageMappingProcessorActor(this); final JsonFieldSelector extraFields = JsonFieldSelector.newInstance("attributes/x", "attributes/y"); final AuthorizationSubject targetAuthSubject = AuthorizationSubject.newInstance("target:auth-subject"); final AuthorizationSubject targetAuthSubjectWithoutIssuer = AuthorizationSubject.newInstance("auth-subject"); final Target targetWithEnrichment = ConnectivityModelFactory.newTargetBuilder() .address("target/address") .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED, targetAuthSubject)) .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS) .withExtraFields(extraFields) .build()) .build(); final Target targetWithoutEnrichment = ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment) .topics(Topic.TWIN_EVENTS) .build(); final Signal<?> signal = TestConstants.thingModified(Collections.emptyList()); final OutboundSignal outboundSignal = OutboundSignalFactory.newOutboundSignal(signal, Arrays.asList(targetWithEnrichment, targetWithoutEnrichment)); underTest.tell(outboundSignal, getRef()); final RetrieveThing retrieveThing = proxyActorProbe.expectMsgClass(RetrieveThing.class); assertThat(retrieveThing.getSelectedFields()).contains(extraFields); assertThat(retrieveThing.getDittoHeaders().getAuthorizationContext()).containsExactly(targetAuthSubject, targetAuthSubjectWithoutIssuer); final JsonObject extra = JsonObject.newBuilder().set("/attributes/x", 5).build(); proxyActorProbe.reply( RetrieveThingResponse.of(retrieveThing.getEntityId(), extra, retrieveThing.getDittoHeaders())); final PublishMappedMessage publishMappedMessage = expectMsgClass(PublishMappedMessage.class); int i = 0; expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichment); expectPublishedMappedMessage(publishMappedMessage, i, signal, targetWithEnrichment, mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra)); }}; }
@Test public void testSignalEnrichmentWithPayloadMappedTargets() { resetActorSystemWithCachingSignalEnrichmentProvider(); final TestProbe proxyActorProbe = TestProbe.apply("mockConciergeForwarder", actorSystem); setUpProxyActor(proxyActorProbe.ref()); new TestKit(actorSystem) {{ final ActorRef underTest = createMessageMappingProcessorActor(this); final JsonFieldSelector extraFields = JsonFactory.newFieldSelector("attributes/x,attributes/y", JsonParseOptions.newBuilder().withoutUrlDecoding().build()); final AuthorizationSubject targetAuthSubject = AuthorizationSubject.newInstance("target:auth-subject"); final AuthorizationSubject targetAuthSubjectWithoutIssuer = AuthorizationSubject.newInstance("auth-subject"); final Target targetWithEnrichment = ConnectivityModelFactory.newTargetBuilder() .address("target/address") .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED, targetAuthSubject)) .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS) .withExtraFields(extraFields) .build()) .build(); final Target targetWithEnrichmentAnd1PayloadMapper = ConnectivityModelFactory.newTargetBuilder() .address("target/address/mapped/1") .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED, targetAuthSubject)) .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS) .withExtraFields(extraFields) .build()) .payloadMapping(ConnectivityModelFactory.newPayloadMapping(ADD_HEADER_MAPPER)) .build(); final Target targetWithEnrichmentAnd2PayloadMappers = ConnectivityModelFactory.newTargetBuilder() .address("target/address/mapped/2") .authorizationContext(AuthorizationContext.newInstance(DittoAuthorizationContextType.UNSPECIFIED, targetAuthSubject)) .topics(ConnectivityModelFactory.newFilteredTopicBuilder(Topic.TWIN_EVENTS) .withExtraFields(extraFields) .build()) .payloadMapping(ConnectivityModelFactory.newPayloadMapping(DUPLICATING_MAPPER, ADD_HEADER_MAPPER)) .build(); final Target targetWithoutEnrichment = ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment) .address("target/address/without/enrichment") .topics(Topic.TWIN_EVENTS) .build(); final Target targetWithoutEnrichmentAnd1PayloadMapper = ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment) .address("target/address/without/enrichment/with/1/payloadmapper") .topics(Topic.TWIN_EVENTS) .payloadMapping(ConnectivityModelFactory.newPayloadMapping(ADD_HEADER_MAPPER)) .build(); final Target targetWithoutEnrichmentAnd2PayloadMappers = ConnectivityModelFactory.newTargetBuilder(targetWithEnrichment) .address("target/address/without/enrichment/with/2/payloadmappers") .topics(Topic.TWIN_EVENTS) .payloadMapping( ConnectivityModelFactory.newPayloadMapping(ADD_HEADER_MAPPER, DUPLICATING_MAPPER)) .build(); final Signal<?> signal = TestConstants.thingModified(Collections.emptyList()) .setRevision(8L); final OutboundSignal outboundSignal = OutboundSignalFactory.newOutboundSignal(signal, Arrays.asList( targetWithEnrichment, targetWithoutEnrichment, targetWithEnrichmentAnd1PayloadMapper, targetWithoutEnrichmentAnd1PayloadMapper, targetWithEnrichmentAnd2PayloadMappers, targetWithoutEnrichmentAnd2PayloadMappers) ); underTest.tell(outboundSignal, getRef()); final RetrieveThing retrieveThing = proxyActorProbe.expectMsgClass(RetrieveThing.class); final JsonFieldSelector extraFieldsWithAdditionalCachingSelectedOnes = JsonFactory.newFieldSelectorBuilder() .addPointers(extraFields) .addFieldDefinition(Thing.JsonFields.REVISION) .build(); assertThat(retrieveThing.getSelectedFields()).contains(extraFieldsWithAdditionalCachingSelectedOnes); assertThat(retrieveThing.getDittoHeaders().getAuthorizationContext()).containsExactly(targetAuthSubject, targetAuthSubjectWithoutIssuer); final JsonObject extra = JsonObject.newBuilder() .set("/attributes/x", 5) .build(); final JsonObject extraForCachingFacade = JsonObject.newBuilder() .set("_revision", 8) .setAll(extra) .build(); proxyActorProbe.reply(RetrieveThingResponse.of(retrieveThing.getEntityId(), extraForCachingFacade, retrieveThing.getDittoHeaders())); final PublishMappedMessage publishMappedMessage = expectMsgClass(PublishMappedMessage.class); int i = 0; expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichment, mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type") ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichmentAnd1PayloadMapper, mapped -> assertThat(mapped.getExternalMessage().getHeaders()) .contains(AddHeaderMessageMapper.OUTBOUND_HEADER) ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichmentAnd2PayloadMappers, mapped -> assertThat(mapped.getExternalMessage().getHeaders()) .contains(AddHeaderMessageMapper.OUTBOUND_HEADER) ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichmentAnd2PayloadMappers, mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type") ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithoutEnrichmentAnd2PayloadMappers, mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type") ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithEnrichment, mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra), mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type") ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithEnrichmentAnd1PayloadMapper, mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra), mapped -> assertThat(mapped.getExternalMessage().getHeaders()) .contains(AddHeaderMessageMapper.OUTBOUND_HEADER) ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithEnrichmentAnd2PayloadMappers, mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra), mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type") ); expectPublishedMappedMessage(publishMappedMessage, i++, signal, targetWithEnrichmentAnd2PayloadMappers, mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra), mapped -> assertThat(mapped.getExternalMessage().getHeaders()).containsOnlyKeys("content-type") ); expectPublishedMappedMessage(publishMappedMessage, i, signal, targetWithEnrichmentAnd2PayloadMappers, mapped -> assertThat(mapped.getAdaptable().getPayload().getExtra()).contains(extra), mapped -> assertThat(mapped.getExternalMessage().getHeaders()) .contains(AddHeaderMessageMapper.OUTBOUND_HEADER) ); }}; }
@Test public void testReplacementOfPlaceholders() { final String correlationId = UUID.randomUUID().toString(); final AuthorizationContext contextWithPlaceholders = AuthorizationModelFactory.newAuthContext(DittoAuthorizationContextType.UNSPECIFIED, AuthorizationModelFactory.newAuthSubject( "integration:{{header:correlation-id}}:hub-{{ header:content-type }}"), AuthorizationModelFactory.newAuthSubject( "integration:{{header:content-type}}:hub-{{ header:correlation-id }}")); final AuthorizationContext expectedAuthContext = TestConstants.Authorization.withUnprefixedSubjects( AuthorizationModelFactory.newAuthContext( DittoAuthorizationContextType.UNSPECIFIED, AuthorizationModelFactory.newAuthSubject( "integration:" + correlationId + ":hub-application/json"), AuthorizationModelFactory.newAuthSubject("integration:application/json:hub-" + correlationId))); testMessageMapping(correlationId, contextWithPlaceholders, ModifyAttribute.class, modifyAttribute -> { assertThat(modifyAttribute.getType()).isEqualTo(ModifyAttribute.TYPE); assertThat(modifyAttribute.getDittoHeaders().getCorrelationId()).contains(correlationId); assertThat(modifyAttribute.getDittoHeaders().getAuthorizationContext().getAuthorizationSubjects()) .isEqualTo(expectedAuthContext.getAuthorizationSubjects()); assertThat(modifyAttribute.getDittoHeaders().get("source")) .contains("integration:" + correlationId + ":hub-application/json"); }); }
@Test public void testMessageWithoutCorrelationId() { final AuthorizationContext connectionAuthContext = AuthorizationModelFactory.newAuthContext( DittoAuthorizationContextType.UNSPECIFIED, AuthorizationModelFactory.newAuthSubject("integration:application/json:hub"), AuthorizationModelFactory.newAuthSubject("integration:hub-application/json")); final AuthorizationContext expectedMessageAuthContext = TestConstants.Authorization.withUnprefixedSubjects(connectionAuthContext); testMessageMappingWithoutCorrelationId(connectionAuthContext, ModifyAttribute.class, modifyAttribute -> { assertThat(modifyAttribute.getType()).isEqualTo(ModifyAttribute.TYPE); assertThat(modifyAttribute.getDittoHeaders().getAuthorizationContext()).isEqualTo( expectedMessageAuthContext); }); }