Client: match subscriptions by context, not path only

client/src/main/java/io/opencmw/client/OpenCmwDataSource.java

@@ -96,7 +96,10 @@ public List<DnsResolver> getRegisteredDnsResolver() {
     private final ZMonitor socketMonitor;
     private final Map<String, URI> subscriptions = new HashMap<>(); // NOPMD: not accessed concurrently
     private final URI serverUri;
-    private final BiPredicate<URI, URI> subscriptionMatcher = new SubscriptionMatcher(); // <notify topic, subscribe topic>
+    // <notify topic, subscribe topic>
+    // Context matching differentiates subscriptions to the same path by the context keys whose Filter
+    // implementations are registered in FilterRegistry; keys without one are ignored.
+    private final BiPredicate<URI, URI> subscriptionMatcher = SubscriptionMatcher.pathAndContext();
     private final long heartbeatInterval;
     private Future<URI> dnsWorkerResult;
     private long nextReconnectAttemptTimeStamp;

core/src/main/java/io/opencmw/filter/SubscriptionMatcher.java

@@ -28,6 +28,21 @@ public SubscriptionMatcher(final Class<? extends Filter>... filterConfig) {
         FilterRegistry.registerNewFilter(filterConfig);
     }
 
+    private SubscriptionMatcher(final boolean pathOnly) {
+        isPathOnly = pathOnly;
+    }
+
+    /**
+     * Matcher that compares path and context, discriminating on whichever filters are registered in
+     * the global {@link FilterRegistry} when a match is evaluated. Use this when the filters are
+     * registered elsewhere, rather than naming them here.
+     *
+     * @return context-aware matcher that registers no filters of its own
+     */
+    public static SubscriptionMatcher pathAndContext() {
+        return new SubscriptionMatcher(false);
+    }
+
     @Override
     public boolean test(final @NotNull URI notified, final @NotNull URI subscriber) {
         return isPathOnly ? testPathOnly(notified, subscriber) : testPathAndContext(notified, subscriber);

core/src/test/java/io/opencmw/filter/SubscriptionMatcherTest.java

@@ -75,6 +75,25 @@ void testMatcherTimingContentContext() {
         test(matcher, false, "property?ctx=FAIR.SELECTOR.ALL", "property?ctx=FAIR.SELECTOR.C=2&contentType=text/html");
     }
 
+    @Test
+    void testMatcherPathAndContextFactory() {
+        // filters registered elsewhere, none named at construction
+        FilterRegistry.registerNewFilter(TestFilter1.class);
+
+        final BiPredicate<URI, URI> matcher = SubscriptionMatcher.pathAndContext();
+
+        // subscriptions to the same path differing only by a registered context key must not cross-match
+        test(matcher, false, "property/A?testKey1=41", "property/A?testKey1=42");
+        test(matcher, false, "property/A?testKey1=42", "property/A?testKey1=41");
+        test(matcher, true, "property/A?testKey1=42", "property/A?testKey1=42");
+
+        // unregistered context keys keep their existing 'ignore' semantics on either side
+        test(matcher, true, "property/A?testKey1=42&unknownKey=x", "property/A?testKey1=42");
+        test(matcher, true, "property/A?testKey1=42", "property/A?testKey1=42&unknownKey=x");
+        test(matcher, true, "property/A?testKey1=42", "property/A*");
+        test(matcher, false, "property/A?testKey1=42", "property/B?testKey1=42");
+    }
+
     private static void test(final BiPredicate<URI, URI> matcher, final boolean expected, final String notify, final String subscription) throws AssertionFailedError {
         final boolean actual = matcher.test(URI.create(notify), URI.create(subscription));
         if (actual != expected) {