replace defined placeholders in authorization context for inbound mes…

…sages and in targets for outbound messages, add new exception for unresolved placeholders

Signed-off-by: Dominik Guggemos <dominik.guggemos@bosch-si.com>

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/ConnectivityModelFactory.java

@@ -339,6 +339,17 @@ public static Target newTarget(final String address, final Topic requiredTopic,
         return newTarget(address, AuthorizationModelFactory.emptyAuthContext(), requiredTopic, additionalTopics);
     }
 
+    /**
+     * Creates a new {@link Target} from existing target but different address.
+     *
+     * @param target the target
+     * @param address the address where the signals will be published
+     * @return the created {@link Target}
+     */
+    public static Target newTarget(final Target target, final String address) {
+        return newTarget(address, target.getTopics(), target.getAuthorizationContext());
+    }
+
     /**
      * Creates a new {@link Target}.
      *

model/connectivity/src/main/java/org/eclipse/ditto/model/connectivity/UnresolvedPlaceholderException.java

@@ -0,0 +1,134 @@
+/*
+ * Copyright (c) 2017 Bosch Software Innovations GmbH.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * which accompanies this distribution, and is available at
+ * https://www.eclipse.org/org/documents/epl-2.0/index.php
+ * Contributors:
+ *    Bosch Software Innovations GmbH - initial contribution
+ *
+ */
+package org.eclipse.ditto.model.connectivity;
+
+import java.net.URI;
+import java.text.MessageFormat;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+import javax.annotation.concurrent.NotThreadSafe;
+
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.model.base.common.HttpStatusCode;
+import org.eclipse.ditto.model.base.exceptions.DittoRuntimeException;
+import org.eclipse.ditto.model.base.exceptions.DittoRuntimeExceptionBuilder;
+import org.eclipse.ditto.model.base.headers.DittoHeaders;
+
+/**
+ * Thrown if a placeholder in the connection configuration could be resolved.
+ */
+@Immutable
+public final class UnresolvedPlaceholderException extends DittoRuntimeException
+        implements ConnectivityException {
+
+    /**
+     * Error code of this exception.
+     */
+    public static final String ERROR_CODE = ERROR_CODE_PREFIX + "connection.placeholder.unresolved";
+
+    private static final String MESSAGE_TEMPLATE = "The placeholder ''{0}'' could not be resolved.";
+
+    private static final String DEFAULT_DESCRIPTION = "Some placeholders could not be resolved. "
+            + "Check the spelling of the placeholder and make sure all required headers are set.";
+
+    private static final long serialVersionUID = 6272495302389903822L;
+
+    private UnresolvedPlaceholderException(final DittoHeaders dittoHeaders,
+            @Nullable final String message,
+            @Nullable final String description,
+            @Nullable final Throwable cause,
+            @Nullable final URI href) {
+
+        super(ERROR_CODE, HttpStatusCode.BAD_REQUEST, dittoHeaders, message, description, cause, href);
+    }
+
+    /**
+     * A mutable builder for a {@code {@link UnresolvedPlaceholderException}}.
+     *
+     * @return the builder.
+     */
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    /**
+     * A mutable builder for a {@code {@link UnresolvedPlaceholderException}}.
+     *
+     * @param unresolvedPlaceholder the unresolved placeholder.
+     * @return the builder.
+     */
+    public static Builder newBuilder(final String unresolvedPlaceholder) {
+        return new Builder(unresolvedPlaceholder);
+    }
+
+    /**
+     * Constructs a new {@code UnresolvedPlaceholderException} object with given message.
+     *
+     * @param message detail message. This message can be later retrieved by the {@link #getMessage()} method.
+     * @param dittoHeaders the headers of the command which resulted in this exception.
+     * @return the new UnresolvedPlaceholderException.
+     */
+    public static UnresolvedPlaceholderException fromMessage(final String message,
+            final DittoHeaders dittoHeaders) {
+
+        return new Builder()
+                .dittoHeaders(dittoHeaders)
+                .message(message)
+                .build();
+    }
+
+    /**
+     * Constructs a new {@code UnresolvedPlaceholderException} object with the exception message extracted from
+     * the given
+     * JSON object.
+     *
+     * @param jsonObject the JSON to read the {@link JsonFields#MESSAGE} field from.
+     * @param dittoHeaders the headers of the command which resulted in this exception.
+     * @return the new UnresolvedPlaceholderException.
+     * @throws org.eclipse.ditto.json.JsonMissingFieldException if the {@code jsonObject} does not have the {@link
+     * JsonFields#MESSAGE} field.
+     */
+    public static UnresolvedPlaceholderException fromJson(final JsonObject jsonObject,
+            final DittoHeaders dittoHeaders) {
+
+        return fromMessage(readMessage(jsonObject), dittoHeaders);
+    }
+
+    /**
+     * A mutable builder with a fluent API for a {@link UnresolvedPlaceholderException}.
+     */
+    @NotThreadSafe
+    public static final class Builder extends DittoRuntimeExceptionBuilder<UnresolvedPlaceholderException> {
+
+        private Builder() {
+            description(DEFAULT_DESCRIPTION);
+        }
+
+        private Builder(final String unresolvedPlaceholder) {
+            this();
+            message(MessageFormat.format(MESSAGE_TEMPLATE, unresolvedPlaceholder));
+        }
+
+        @Override
+        protected UnresolvedPlaceholderException doBuild(final DittoHeaders dittoHeaders,
+                @Nullable final String message,
+                @Nullable final String description,
+                @Nullable final Throwable cause,
+                @Nullable final URI href) {
+
+            return new UnresolvedPlaceholderException(dittoHeaders, message, description, cause, href);
+        }
+
+    }
+
+}

model/connectivity/src/test/java/org/eclipse/ditto/model/connectivity/ImmutableExternalMessageTest.java

@@ -18,6 +18,7 @@
 
 import java.nio.ByteBuffer;
 
+import org.eclipse.ditto.model.base.auth.AuthorizationContext;
 import org.junit.Test;
 
 import nl.jqno.equalsverifier.EqualsVerifier;
@@ -31,7 +32,7 @@ public final class ImmutableExternalMessageTest {
     public void assertImmutability() {
         assertInstancesOf(ImmutableExternalMessage.class, areImmutable(),
                 assumingFields("bytePayload").areNotModifiedAndDoNotEscape(),
-                provided(ByteBuffer.class).areAlsoImmutable());
+                provided(ByteBuffer.class, AuthorizationContext.class).areAlsoImmutable());
     }
 
     @Test

services/connectivity/mapping/src/main/java/org/eclipse/ditto/services/connectivity/mapping/DittoMessageMapper.java

@@ -31,6 +31,7 @@
 import org.eclipse.ditto.protocoladapter.Adaptable;
 import org.eclipse.ditto.protocoladapter.JsonifiableAdaptable;
 import org.eclipse.ditto.protocoladapter.ProtocolFactory;
+import org.eclipse.ditto.protocoladapter.TopicPath;
 
 import com.typesafe.config.Config;
 
@@ -77,9 +78,13 @@ public Optional<ExternalMessage> map(final Adaptable adaptable) {
 
         final String jsonString = ProtocolFactory.wrapAsJsonifiableAdaptable(adaptable).toJsonString();
 
+        final boolean isError = TopicPath.Criterion.ERRORS.equals(adaptable.getTopicPath().getCriterion());
+        final boolean isResponse = adaptable.getPayload().getStatus().isPresent();
         return Optional.of(
                 ConnectivityModelFactory.newExternalMessageBuilder(headers)
-                        .withText(jsonString).asResponse(adaptable.getPayload().getStatus().isPresent())
+                        .withText(jsonString)
+                        .asResponse(isResponse)
+                        .asError(isError)
                 .build());
     }
 

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/ConnectionActor.java

@@ -127,6 +127,7 @@ final class ConnectionActor extends AbstractPersistentActor {
 
     private long lastSnapshotSequenceNr = -1L;
     private boolean snapshotInProgress = false;
+    private final PlaceholderFilter placeholdersFilter = new PlaceholderFilter();
 
     private Set<Topic> uniqueTopicPaths = Collections.emptySet();
 
@@ -311,8 +312,10 @@ private void handleSignal(final Signal<?> signal) {
         final Set<Target> subscribedAndAuthorizedTargets = SignalFilter.filter(connection, signal);
         // forward to client actor if topic was subscribed and there are targets that are authorized to read
         if (!subscribedAndAuthorizedTargets.isEmpty()) {
-            log.debug("Forwarding signal <{}> to client actor.", signal.getType());
-            final OutboundSignal outbound = new UnmappedOutboundSignal(signal, subscribedAndAuthorizedTargets);
+            final Set<Target> filteredTargets =
+                    placeholdersFilter.filterTargets(subscribedAndAuthorizedTargets, signal.getId());
+            log.debug("Forwarding signal <{}> to client actor with targets: {}.", signal.getType(), filteredTargets);
+            final OutboundSignal outbound = new UnmappedOutboundSignal(signal, filteredTargets);
             clientActor.tell(outbound, getSelf());
         }
     }

services/connectivity/messaging/src/main/java/org/eclipse/ditto/services/connectivity/messaging/MessageMappingProcessorActor.java

@@ -74,6 +74,7 @@ public final class MessageMappingProcessorActor extends AbstractActor {
     private final MessageMappingProcessor processor;
     private final String connectionId;
     private final ActorRef conciergeForwarder;
+    private final PlaceholderFilter placeholderFilter;
 
     private MessageMappingProcessorActor(final ActorRef publisherActor,
             final ActorRef conciergeForwarder,
@@ -85,6 +86,7 @@ private MessageMappingProcessorActor(final ActorRef publisherActor,
         this.processor = processor;
         this.headerFilter = headerFilter;
         this.connectionId = connectionId;
+        this.placeholderFilter = new PlaceholderFilter();
         final Caffeine caffeine = Caffeine.newBuilder()
                 .expireAfterWrite(5, TimeUnit.MINUTES);
         traces = CaffeineCache.of(caffeine);
@@ -163,7 +165,9 @@ private void handle(final ExternalMessage externalMessage) {
 
         try {
             final AuthorizationContext authorizationContext = getAuthorizationContextFromMessage(externalMessage);
-            final String authSubjectsArray = mapAuthorizationContextToSubjectsArray(authorizationContext);
+            final AuthorizationContext filteredContext =
+                    placeholderFilter.filterAuthorizationContext(authorizationContext, externalMessage.getHeaders());
+            final String authSubjectsArray = mapAuthorizationContextToSubjectsArray(filteredContext);
             final ExternalMessage messageWithAuthSubject =
                     externalMessage.withHeader(DittoHeaderDefinition.AUTHORIZATION_SUBJECTS.getKey(),
                             authSubjectsArray);
@@ -173,7 +177,7 @@ private void handle(final ExternalMessage externalMessage) {
             signalOpt.ifPresent(signal -> {
                 enhanceLogUtil(signal);
                 final DittoHeadersBuilder adjustedHeadersBuilder = signal.getDittoHeaders().toBuilder()
-                        .authorizationContext(authorizationContext);
+                        .authorizationContext(filteredContext);
 
                 if (!signal.getDittoHeaders().getOrigin().isPresent()) {
                     adjustedHeadersBuilder.origin(connectionId);