support fn: pipeline expressions in connection target topic filters

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

@@ -842,8 +842,9 @@ public static Target targetFromJson(final JsonObject jsonObject) {
     }
 
     /**
-     * Creates a new {@code FilteredTopic} from the passed {@code topicString} which consists of a {@code Topic} and an
-     * optional filter string supplied with {@code ?filter=...}.
+     * Creates a new {@code FilteredTopic} from the passed {@code topicString} which consists of a {@code Topic} and
+     * optional filter strings supplied with {@code ?filter=...}. The {@code filter} query parameter may be repeated
+     * ({@code ?filter=...&filter=...}); all given filters must match for a signal to be processed (AND semantics).
      *
      * @param topicString the {@code FilteredTopic} String representation
      * @return the created FilteredTopic

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

@@ -18,8 +18,10 @@
 import org.eclipse.ditto.json.JsonFieldSelector;
 
 /**
- * A FilteredTopic wraps a {@link Topic} and an optional {@code filter} String which additionally restricts which
- * kind of Signals should be processed/filtered based on an {@code RQL} query.
+ * A FilteredTopic wraps a {@link Topic} and optional {@code filter} Strings which additionally restrict which
+ * kind of Signals should be processed/filtered. Each filter is either an {@code RQL} query or a placeholder
+ * pipeline expression starting with {@code fn:}; all filters of one topic must match for a signal to be
+ * processed (AND semantics).
  */
 public interface FilteredTopic extends CharSequence {
 
@@ -34,10 +36,22 @@ public interface FilteredTopic extends CharSequence {
     List<String> getNamespaces();
 
     /**
-     * @return the optional filter string as RQL query
+     * @return the first filter string of this FilteredTopic, or an empty Optional if no filter is set.
+     * @deprecated as of 3.10.0 a FilteredTopic may carry multiple filters; use {@link #getFilters()} instead.
      */
+    @Deprecated
     Optional<String> getFilter();
 
+    /**
+     * Returns the filter strings of this FilteredTopic in insertion order. All filters of one topic must match
+     * for a signal to be processed (AND semantics). At most one entry may be an RQL expression; any number of
+     * entries may be placeholder pipeline expressions starting with {@code fn:}.
+     *
+     * @return the filter strings, or an empty list if no filter is set.
+     * @since 3.10.0
+     */
+    List<String> getFilters();
+
     /**
      * Returns the selector for the extra fields and their values to enrich outgoing signals with.
      *

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

@@ -35,13 +35,28 @@ public interface FilteredTopicBuilder {
     FilteredTopicBuilder withNamespaces(@Nullable Collection<String> namespaces);
 
     /**
-     * Sets the given filter to this builder.
+     * Sets the given filter to this builder, replacing all previously set filters.
      *
-     * @param filter the optional RQL filter of the topic to be built.
+     * @param filter the optional filter of the topic to be built.
      * @return this builder instance to allow method chaining.
+     * @deprecated as of 3.10.0 a FilteredTopic may carry multiple filters; use {@link #withFilters(Collection)}
+     * instead.
      */
+    @Deprecated
     FilteredTopicBuilder withFilter(@Nullable CharSequence filter);
 
+    /**
+     * Sets the given filters to this builder, replacing all previously set filters. The insertion order is
+     * preserved and determines the serialization order of the {@code filter} query parameters; two topics with
+     * the same filters in different order are not equal.
+     *
+     * @param filters the filters of the topic to be built - each entry is either an RQL expression or a
+     * placeholder pipeline expression starting with {@code fn:}.
+     * @return this builder instance to allow method chaining.
+     * @since 3.10.0
+     */
+    FilteredTopicBuilder withFilters(@Nullable Collection<? extends CharSequence> filters);
+
     /**
      * Sets the selector for the extra fields and their values to enrich outgoing signals of the topic to be built with.
      *

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

@@ -21,6 +21,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -51,7 +52,7 @@ final class ImmutableFilteredTopic implements FilteredTopic {
 
     private final Topic topic;
     private final List<String> namespaces;
-    @Nullable private final String filterString;
+    private final List<String> filters;
     @Nullable private final ThingFieldSelector extraFields;
 
     private ImmutableFilteredTopic(final ImmutableFilteredTopicBuilder builder) {
@@ -60,7 +61,10 @@ private ImmutableFilteredTopic(final ImmutableFilteredTopicBuilder builder) {
         namespaces = null != namespacesFromBuilder
                 ? Collections.unmodifiableList(new ArrayList<>(namespacesFromBuilder))
                 : Collections.emptyList();
-        filterString = Objects.toString(builder.filter, null);
+        final Collection<String> filtersFromBuilder = builder.filters;
+        filters = null != filtersFromBuilder
+                ? Collections.unmodifiableList(new ArrayList<>(filtersFromBuilder))
+                : Collections.emptyList();
         extraFields = builder.extraFields;
     }
 
@@ -102,7 +106,12 @@ public List<String> getNamespaces() {
 
     @Override
     public Optional<String> getFilter() {
-        return Optional.ofNullable(filterString);
+        return filters.isEmpty() ? Optional.empty() : Optional.of(filters.get(0));
+    }
+
+    @Override
+    public List<String> getFilters() {
+        return filters;
     }
 
     @Override
@@ -131,9 +140,13 @@ public String toString() {
     }
 
     private String getQueryParametersAsString() {
-        return join(QUERY_ARG_DELIMITER, getQueryParameterString(NAMESPACES_ARG, String.join(",", namespaces)),
-                getQueryParameterString(FILTER_ARG, filterString),
-                getQueryParameterString(EXTRA_FIELDS_ARG, extraFields));
+        final List<String> queryParameterStrings = new ArrayList<>(2 + filters.size());
+        queryParameterStrings.add(getQueryParameterString(NAMESPACES_ARG, String.join(",", namespaces)));
+        for (final String filter : filters) {
+            queryParameterStrings.add(getQueryParameterString(FILTER_ARG, filter));
+        }
+        queryParameterStrings.add(getQueryParameterString(EXTRA_FIELDS_ARG, extraFields));
+        return join(QUERY_ARG_DELIMITER, queryParameterStrings.toArray(new String[0]));
     }
 
     private static String getQueryParameterString(final String parameterName, @Nullable final Object parameterValue) {
@@ -169,13 +182,13 @@ public boolean equals(@Nullable final Object o) {
         final ImmutableFilteredTopic that = (ImmutableFilteredTopic) o;
         return topic == that.topic &&
                 namespaces.equals(that.namespaces) &&
-                Objects.equals(filterString, that.filterString) &&
+                filters.equals(that.filters) &&
                 Objects.equals(extraFields, that.extraFields);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(topic, namespaces, filterString, extraFields);
+        return Objects.hash(topic, namespaces, filters, extraFields);
     }
 
     /**
@@ -186,13 +199,13 @@ static final class ImmutableFilteredTopicBuilder implements FilteredTopicBuilder
 
         private final Topic topic;
         @Nullable private Collection<String> namespaces;
-        @Nullable private CharSequence filter;
+        @Nullable private List<String> filters;
         @Nullable private ThingFieldSelector extraFields;
 
         private ImmutableFilteredTopicBuilder(final Topic topic) {
             this.topic = checkNotNull(topic, "topic");
             namespaces = null;
-            filter = null;
+            filters = null;
             extraFields = null;
         }
 
@@ -206,8 +219,15 @@ public ImmutableFilteredTopicBuilder withNamespaces(@Nullable final Collection<S
 
         @Override
         public ImmutableFilteredTopicBuilder withFilter(@Nullable final CharSequence filter) {
+            return withFilters(null != filter ? Collections.singletonList(filter) : null);
+        }
+
+        @Override
+        public ImmutableFilteredTopicBuilder withFilters(@Nullable final Collection<? extends CharSequence> filters) {
             if (supportsFilters()) {
-                this.filter = filter;
+                this.filters = null != filters
+                        ? filters.stream().map(CharSequence::toString).collect(Collectors.toList())
+                        : null;
             }
             return this;
         }
@@ -239,13 +259,15 @@ private boolean supportsExtraFields() {
 
     }
 
-    @Immutable
+    @NotThreadSafe
     private static final class FilteredTopicStringParser {
 
         private final String filteredTopicString;
+        private final List<String> filterValues;
 
         private FilteredTopicStringParser(final String filteredTopicString) {
             this.filteredTopicString = filteredTopicString;
+            filterValues = new ArrayList<>(1);
         }
 
         ImmutableFilteredTopic parse() {
@@ -260,7 +282,7 @@ ImmutableFilteredTopic parse() {
 
             return getBuilder(parseTopic(topicName))
                     .withNamespaces(parseNamespaces(queryParameters.get(NAMESPACES_ARG)))
-                    .withFilter(queryParameters.get(FILTER_ARG))
+                    .withFilters(filterValues.isEmpty() ? null : filterValues)
                     .withExtraFields(parseExtraFields(queryParameters.get(EXTRA_FIELDS_ARG)))
                     .build();
         }
@@ -271,14 +293,26 @@ private Topic parseTopic(final String topicName) {
                             "Unknown topic: " + topicName).build());
         }
 
-        private static Map<String, String> parseQueryParameters(@Nullable final String queryParamsString) {
+        private Map<String, String> parseQueryParameters(@Nullable final String queryParamsString) {
             if (null == queryParamsString || queryParamsString.isEmpty()) {
                 return Collections.emptyMap();
             }
-            return Arrays.stream(queryParamsString.split(QUERY_ARG_DELIMITER))
-                    .map(paramString -> paramString.split(QUERY_ARG_VALUE_DELIMITER, 2))
-                    .filter(queryParamPair -> 2 == queryParamPair.length)
-                    .collect(Collectors.toMap(queryParamPair -> urlDecode(queryParamPair[0]), av -> urlDecode(av[1])));
+            final Map<String, String> queryParameters = new HashMap<>(4);
+            for (final String paramString : queryParamsString.split(QUERY_ARG_DELIMITER)) {
+                final String[] queryParamPair = paramString.split(QUERY_ARG_VALUE_DELIMITER, 2);
+                if (2 != queryParamPair.length) {
+                    continue;
+                }
+                final String name = urlDecode(queryParamPair[0]);
+                final String value = urlDecode(queryParamPair[1]);
+                if (FILTER_ARG.equals(name)) {
+                    // the filter parameter is the only repeatable one: values are collected in insertion order
+                    filterValues.add(value);
+                } else if (null != queryParameters.putIfAbsent(name, value)) {
+                    throw new IllegalStateException("Duplicate key " + name);
+                }
+            }
+            return queryParameters;
         }
 
         private static String urlDecode(final String value) {

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

@@ -15,6 +15,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.text.MessageFormat;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -247,4 +248,119 @@ public void fromStringParsesAsExpectedWithOnlyExtraFields() {
         assertThat(actual).isEqualTo(filteredTopic);
     }
 
+    @Test
+    public void fromStringParsesAsExpectedWithNamespacesExtraFieldsAndRqlAndPipelineFilters() {
+        final String rqlFilter = "gt(attributes/counter,42)";
+        final String pipelineFilter = "fn:filter(header:ditto-originator,'ne','some:subject')";
+        final ImmutableFilteredTopic filteredTopic = ImmutableFilteredTopic.getBuilder(Topic.TWIN_EVENTS)
+                .withNamespaces(Lists.list("ns1", "ns2"))
+                .withExtraFields(ThingFieldSelector.fromString("attributes"))
+                .withFilters(Arrays.asList(rqlFilter, pipelineFilter))
+                .build();
+        final String filteredTopicString = filteredTopic.toString();
+
+        final ImmutableFilteredTopic actual = ImmutableFilteredTopic.fromString(filteredTopicString);
+
+        assertThat(filteredTopicString).contains("filter=" + rqlFilter + "&filter=" + pipelineFilter);
+        assertThat(actual.getFilters()).containsExactly(rqlFilter, pipelineFilter);
+        assertThat(actual.toString()).isEqualTo(filteredTopicString);
+        assertThat(actual).isEqualTo(filteredTopic);
+    }
+
+    @Test
+    public void getFiltersReturnsAllInInsertionOrder() {
+        final ImmutableFilteredTopic underTest = ImmutableFilteredTopic.getBuilder(Topic.TWIN_EVENTS)
+                .withFilters(Arrays.asList("fn:filter(header:b,'exists')", FILTER_EXAMPLE, "fn:filter(header:a,'exists')"))
+                .build();
+
+        assertThat(underTest.getFilters())
+                .containsExactly("fn:filter(header:b,'exists')", FILTER_EXAMPLE, "fn:filter(header:a,'exists')");
+    }
+
+    @Test
+    public void getFilterReturnsFirstOfMultipleFilters() {
+        // contract of the deprecated single-filter accessor: the FIRST filter in insertion order
+        final ImmutableFilteredTopic underTest = ImmutableFilteredTopic.getBuilder(Topic.TWIN_EVENTS)
+                .withFilters(Arrays.asList(FILTER_EXAMPLE, "fn:filter(header:a,'exists')"))
+                .build();
+
+        assertThat(underTest.getFilter()).contains(FILTER_EXAMPLE);
+    }
+
+    @Test
+    public void withFilterReplacesPreviouslySetFilters() {
+        final ImmutableFilteredTopic underTest = ImmutableFilteredTopic.getBuilder(Topic.TWIN_EVENTS)
+                .withFilters(Arrays.asList("fn:filter(header:a,'exists')", "fn:filter(header:b,'exists')"))
+                .withFilter(FILTER_EXAMPLE)
+                .build();
+
+        assertThat(underTest.getFilters()).containsExactly(FILTER_EXAMPLE);
+    }
+
+    @Test
+    public void withFiltersNullResetsFilters() {
+        final ImmutableFilteredTopic underTest = ImmutableFilteredTopic.getBuilder(Topic.TWIN_EVENTS)
+                .withFilters(Collections.singletonList(FILTER_EXAMPLE))
+                .withFilters(null)
+                .build();
+
+        assertThat(underTest.getFilters()).isEmpty();
+        assertThat(underTest.getFilter()).isEmpty();
+    }
+
+    @Test
+    public void toStringEmitsOneFilterParamPerEntryInOrder() {
+        final ImmutableFilteredTopic underTest = ImmutableFilteredTopic.getBuilder(Topic.TWIN_EVENTS)
+                .withNamespaces(NAMESPACES)
+                .withFilters(Arrays.asList(FILTER_EXAMPLE, "fn:filter(header:a,'exists')"))
+                .withExtraFields(EXTRA_FIELDS)
+                .build();
+
+        assertThat(underTest.toString()).isEqualTo(
+                "_/_/things/twin/events?namespaces=" + String.join(",", NAMESPACES)
+                        + "&filter=" + FILTER_EXAMPLE
+                        + "&filter=fn:filter(header:a,'exists')"
+                        + "&extraFields=" + EXTRA_FIELDS);
+    }
+
+    @Test
+    public void fromStringCollectsRepeatedFilterParamsInOrder() {
+        final ImmutableFilteredTopic actual = ImmutableFilteredTopic.fromString(
+                "_/_/things/twin/events?filter=fn:filter(header:a,'exists')&filter=" + FILTER_EXAMPLE);
+
+        assertThat(actual.getFilters()).containsExactly("fn:filter(header:a,'exists')", FILTER_EXAMPLE);
+    }
+
+    @Test
+    public void fromStringToStringRoundTripsWithMultipleFilters() {
+        final ImmutableFilteredTopic filteredTopic = ImmutableFilteredTopic.getBuilder(Topic.LIVE_MESSAGES)
+                .withFilters(Arrays.asList("fn:filter(header:ditto-originator,'ne','some:subject')",
+                        "fn:filter(header:ditto-origin,'ne','some-connection')"))
+                .build();
+
+        final ImmutableFilteredTopic actual = ImmutableFilteredTopic.fromString(filteredTopic.toString());
+
+        assertThat(actual).isEqualTo(filteredTopic);
+        assertThat(actual.toString()).isEqualTo(filteredTopic.toString());
+    }
+
+    @Test
+    public void fromStringDuplicateNamespacesParamStillThrows() {
+        // freezes the (out-of-scope) pre-existing behavior: only the "filter" query parameter is repeatable,
+        // any other duplicated parameter keeps failing like the previous Collectors.toMap-based parsing did
+        org.assertj.core.api.Assertions.assertThatExceptionOfType(IllegalStateException.class)
+                .isThrownBy(() -> ImmutableFilteredTopic.fromString(
+                        "_/_/things/twin/events?namespaces=ns1&namespaces=ns2"))
+                .withMessageContaining("Duplicate key");
+    }
+
+    @Test
+    public void announcementTopicsDropFiltersSetViaWithFilters() {
+        final ImmutableFilteredTopic underTest = ImmutableFilteredTopic.getBuilder(Topic.POLICY_ANNOUNCEMENTS)
+                .withFilters(Arrays.asList(FILTER_EXAMPLE, "fn:filter(header:a,'exists')"))
+                .build();
+
+        assertThat(underTest.getFilters()).isEmpty();
+    }
+
 }