MessageMappingProcessorActor extends AbstractGraphActor<MessageMappingProcessorActor.OutboundSignalWithId, OutboundSignal> { static Signal<?> filterAcknowledgements(final Signal<?> signal, final @Nullable String filter) { if (filter != null) { final String requestedAcks = DittoHeaderDefinition.REQUESTED_ACKS.getKey(); final boolean headerDefined = signal.getDittoHeaders().containsKey(requestedAcks); final String fullFilter = "header:" + requestedAcks + "|fn:default('[]')|" + filter; final ExpressionResolver resolver = Resolvers.forSignal(signal); final Optional<String> resolverResult = resolver.resolveAsPipelineElement(fullFilter).toOptional(); if (resolverResult.isEmpty()) { return signal.setDittoHeaders(DittoHeaders.newBuilder(signal.getDittoHeaders()) .acknowledgementRequests(Collections.emptySet()) .build()); } else if (headerDefined) { return signal.setDittoHeaders(DittoHeaders.newBuilder(signal.getDittoHeaders()) .putHeader(requestedAcks, resolverResult.orElseThrow()) .build()); } else { final Optional<String> unsetFilterResult = resolver.resolveAsPipelineElement(filter).toOptional(); return unsetFilterResult.<Signal<?>>map(newAckRequests -> signal.setDittoHeaders(DittoHeaders.newBuilder(signal.getDittoHeaders()) .putHeader(requestedAcks, newAckRequests) .build())) .orElse(signal); } } return signal; } @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 testRequestedAcknowledgementFilter() { final String requestedAcks = DittoHeaderDefinition.REQUESTED_ACKS.getKey(); final AcknowledgementRequest twinPersistedAckRequest = AcknowledgementRequest.of(DittoAcknowledgementLabel.TWIN_PERSISTED); final Signal<?> signal = DeleteThing.of(ThingId.of("thing:id"), DittoHeaders.empty()); final Signal<?> signalWithRequestedAcks = DeleteThing.of(ThingId.of("thing:id"), DittoHeaders.newBuilder() .acknowledgementRequest(twinPersistedAckRequest) .build()); final Signal<?> notFilteredSignal = filterAcknowledgements(signal, "fn:filter('2+2','ne','5')"); assertThat(notFilteredSignal.getDittoHeaders()).doesNotContainKey(requestedAcks); final Signal<?> filteredSignal = filterAcknowledgements(signal, "fn:filter('2+2','eq','5')"); assertThat(filteredSignal.getDittoHeaders()).contains(Map.entry(requestedAcks, "[]")); assertThatExceptionOfType(PlaceholderFunctionSignatureInvalidException.class).isThrownBy(() -> filterAcknowledgements(signal, "fn:filter('2','+','2','eq','5')") ); final Signal<?> defaultValueSetSignal = filterAcknowledgements(signal, "fn:default('[\"twin-persisted\"]')"); assertThat(defaultValueSetSignal.getDittoHeaders().getAcknowledgementRequests()) .containsExactly(twinPersistedAckRequest); final Signal<?> transformedSignal = filterAcknowledgements(signalWithRequestedAcks, "fn:filter('2+2','eq','5')|fn:default('[\"custom\"]')"); assertThat(transformedSignal.getDittoHeaders().getAcknowledgementRequests()) .containsExactly(AcknowledgementRequest.parseAcknowledgementRequest("custom")); }