KV Watch Multiple Filters (#1113)

scottf · web-flow · commit 9661dfc2a3ff · 2024-04-10T10:41:06.000-04:00
diff --git a/src/main/java/io/nats/client/KeyValue.java b/src/main/java/io/nats/client/KeyValue.java
@@ -169,7 +169,7 @@ public interface KeyValue {
 
     /**
      * Watch updates for a specific key.
-     * @param key the key
+     * @param key the key. Also accepts a comma delimited list.
      * @param watcher the watcher the implementation to receive changes
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
      * @return The KeyValueWatchSubscription
@@ -182,7 +182,7 @@ public interface KeyValue {
 
     /**
      * Watch updates for a specific key, starting at a specific revision.
-     * @param key the key
+     * @param key the key. Also accepts a comma delimited list.
      * @param watcher the watcher the implementation to receive changes
      * @param fromRevision the revision to start from
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
@@ -194,6 +194,33 @@ public interface KeyValue {
      */
     NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException;
 
+    /**
+     * Watch updates for specific keys.
+     * @param keys the keys
+     * @param watcher the watcher the implementation to receive changes
+     * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
+     * @return The KeyValueWatchSubscription
+     * @throws IOException covers various communication issues with the NATS
+     *         server such as timeout or interruption
+     * @throws JetStreamApiException the request had an error related to the data
+     * @throws InterruptedException if the thread is interrupted
+     */
+    NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException;
+
+    /**
+     * Watch updates for specific keys, starting at a specific revision.
+     * @param keys the keys
+     * @param watcher the watcher the implementation to receive changes
+     * @param fromRevision the revision to start from
+     * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
+     * @return The KeyValueWatchSubscription
+     * @throws IOException covers various communication issues with the NATS
+     *         server such as timeout or interruption
+     * @throws JetStreamApiException the request had an error related to the data
+     * @throws InterruptedException if the thread is interrupted
+     */
+    NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException;
+
     /**
      * Watch updates for all keys.
      * @param watcher the watcher the implementation to receive changes
diff --git a/src/main/java/io/nats/client/impl/NatsKeyValue.java b/src/main/java/io/nats/client/impl/NatsKeyValue.java
@@ -25,6 +25,8 @@
 import java.nio.charset.StandardCharsets;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 import static io.nats.client.support.NatsConstants.DOT;
@@ -237,28 +239,41 @@ private PublishAck _write(String key, byte[] data, Headers h) throws IOException
 
     @Override
     public NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
-        validateKvKeyWildcardAllowedRequired(key);
-        validateNotNull(watcher, "Watcher is required");
-        return new NatsKeyValueWatchSubscription(this, key, watcher, -1, watchOptions);
+        if (key.contains(",")) {
+            return watch(Arrays.asList(key.split(",")), watcher, -1, watchOptions);
+        }
+        return watch(Collections.singletonList(key), watcher, -1, watchOptions);
     }
 
     @Override
     public NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
-        validateKvKeyWildcardAllowedRequired(key);
+        if (key.contains(",")) {
+            return watch(Arrays.asList(key.split(",")), watcher, fromRevision, watchOptions);
+        }
+        return watch(Collections.singletonList(key), watcher, fromRevision, watchOptions);
+    }
+
+    @Override
+    public NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
+        return watch(keys, watcher, -1, watchOptions);
+    }
+
+    @Override
+    public NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
+        // all watch methods (watch, watch all) delegate to here
+        validateKvKeyWildcardAllowedRequired(keys);
         validateNotNull(watcher, "Watcher is required");
-        return new NatsKeyValueWatchSubscription(this, key, watcher, fromRevision, watchOptions);
+        return new NatsKeyValueWatchSubscription(this, keys, watcher, fromRevision, watchOptions);
     }
 
     @Override
     public NatsKeyValueWatchSubscription watchAll(KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
-        validateNotNull(watcher, "Watcher is required");
-        return new NatsKeyValueWatchSubscription(this, ">", watcher, -1, watchOptions);
+        return new NatsKeyValueWatchSubscription(this, Collections.singletonList(">"), watcher, -1, watchOptions);
     }
 
     @Override
     public NatsKeyValueWatchSubscription watchAll(KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
-        validateNotNull(watcher, "Watcher is required");
-        return new NatsKeyValueWatchSubscription(this, ">", watcher, fromRevision, watchOptions);
+        return new NatsKeyValueWatchSubscription(this, Collections.singletonList(">"), watcher, fromRevision, watchOptions);
     }
 
     /**
diff --git a/src/main/java/io/nats/client/impl/NatsKeyValueWatchSubscription.java b/src/main/java/io/nats/client/impl/NatsKeyValueWatchSubscription.java
@@ -18,12 +18,22 @@
 import io.nats.client.api.*;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 public class NatsKeyValueWatchSubscription extends NatsWatchSubscription<KeyValueEntry> {
 
     public NatsKeyValueWatchSubscription(NatsKeyValue kv, String keyPattern, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException {
+        this(kv, Collections.singletonList(keyPattern), watcher, fromRevision, watchOptions);
+    }
+
+    public NatsKeyValueWatchSubscription(NatsKeyValue kv, List<String> keyPatterns, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException {
         super(kv.js);
+        kwWatchInit(kv, keyPatterns, watcher, fromRevision, watchOptions);
+    }
 
+    private void kwWatchInit(NatsKeyValue kv, List<String> keyPatterns, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption[] watchOptions) throws IOException, JetStreamApiException {
         // figure out the result options
         boolean headersOnly = false;
         boolean ignoreDeletes = false;
@@ -54,6 +64,12 @@ public void onMessage(Message m) throws InterruptedException {
                 }
             };
 
-        finishInit(kv, kv.readSubject(keyPattern), deliverPolicy, headersOnly, fromRevision, handler);
+        // convert each key to a read subject
+        List<String> readSubjects = new ArrayList<>();
+        for (String keyPattern : keyPatterns) {
+            readSubjects.add(kv.readSubject(keyPattern.trim()));
+        }
+
+        finishInit(kv, readSubjects, deliverPolicy, headersOnly, fromRevision, handler);
     }
 }
diff --git a/src/main/java/io/nats/client/impl/NatsObjectStoreWatchSubscription.java b/src/main/java/io/nats/client/impl/NatsObjectStoreWatchSubscription.java
@@ -21,6 +21,7 @@
 import io.nats.client.api.ObjectStoreWatcher;
 
 import java.io.IOException;
+import java.util.Collections;
 
 import static io.nats.client.api.ConsumerConfiguration.ULONG_UNSET;
 
@@ -58,6 +59,6 @@ public void onMessage(Message m) throws InterruptedException {
                 }
             };
 
-        finishInit(os, os.rawAllMetaSubject(), deliverPolicy, headersOnly, ULONG_UNSET, handler);
+        finishInit(os, Collections.singletonList(os.rawAllMetaSubject()), deliverPolicy, headersOnly, ULONG_UNSET, handler);
     }
 }
diff --git a/src/main/java/io/nats/client/impl/NatsWatchSubscription.java b/src/main/java/io/nats/client/impl/NatsWatchSubscription.java
@@ -20,6 +20,7 @@
 import io.nats.client.api.Watcher;
 
 import java.io.IOException;
+import java.util.List;
 
 import static io.nats.client.api.ConsumerConfiguration.ULONG_UNSET;
 
@@ -32,15 +33,15 @@ public NatsWatchSubscription(JetStream js) {
         this.js = js;
     }
 
-    protected void finishInit(NatsFeatureBase fb, String subscribeSubject, DeliverPolicy deliverPolicy, boolean headersOnly, long fromRevision, WatchMessageHandler<T> handler)
+    protected void finishInit(NatsFeatureBase fb, List<String> subscribeSubjects, DeliverPolicy deliverPolicy, boolean headersOnly, long fromRevision, WatchMessageHandler<T> handler)
         throws IOException, JetStreamApiException
     {
         if (fromRevision > ULONG_UNSET) {
             deliverPolicy = DeliverPolicy.ByStartSequence;
         }
         else {
             fromRevision = ULONG_UNSET; // easier on the builder since we aren't starting at a fromRevision
-            if (deliverPolicy == DeliverPolicy.New || fb._getLast(subscribeSubject) == null) {
+            if (deliverPolicy == DeliverPolicy.New) {
                 handler.sendEndOfData();
             }
         }
@@ -53,12 +54,12 @@ protected void finishInit(NatsFeatureBase fb, String subscribeSubject, DeliverPo
                 .deliverPolicy(deliverPolicy)
                 .startSequence(fromRevision)
                 .headersOnly(headersOnly)
-                .filterSubject(subscribeSubject)
+                .filterSubjects(subscribeSubjects)
                 .build())
             .build();
 
         dispatcher = (NatsDispatcher) ((NatsJetStream) js).conn.createDispatcher();
-        sub = js.subscribe(subscribeSubject, dispatcher, handler, false, pso);
+        sub = js.subscribe(null, dispatcher, handler, false, pso);
         if (!handler.endOfDataSent) {
             long pending = sub.getConsumerInfo().getCalculatedPending();
             if (pending == 0) {
diff --git a/src/main/java/io/nats/client/support/Validator.java b/src/main/java/io/nats/client/support/Validator.java
@@ -124,6 +124,14 @@ public static String validatePrefixOrDomain(String s, String label, boolean requ
         });
     }
 
+    public static List<String> validateKvKeyWildcardAllowedRequired(List<String> keys) {
+        required(keys, "Key");
+        for (String key : keys) {
+            validateWildcardKvKey(key, "Key", true);
+        }
+        return keys;
+    }
+
     public static String validateKvKeyWildcardAllowedRequired(String s) {
         return validateWildcardKvKey(s, "Key", true);
     }
diff --git a/src/test/java/io/nats/client/impl/KeyValueTests.java b/src/test/java/io/nats/client/impl/KeyValueTests.java
@@ -912,6 +912,8 @@ public void testWatch() throws Exception {
         TestKeyValueWatcher starMetaWatcher = new TestKeyValueWatcher("starMetaWatcher", true, META_ONLY);
         TestKeyValueWatcher gtFullWatcher = new TestKeyValueWatcher("gtFullWatcher", true);
         TestKeyValueWatcher gtMetaWatcher = new TestKeyValueWatcher("gtMetaWatcher", true, META_ONLY);
+        TestKeyValueWatcher multipleFullWatcher = new TestKeyValueWatcher("multipleFullWatcher", true);
+        TestKeyValueWatcher multipleMetaWatcher = new TestKeyValueWatcher("multipleMetaWatcher", true, META_ONLY);
         TestKeyValueWatcher key1AfterWatcher = new TestKeyValueWatcher("key1AfterWatcher", false, META_ONLY);
         TestKeyValueWatcher key1AfterIgDelWatcher = new TestKeyValueWatcher("key1AfterIgDelWatcher", false, META_ONLY, IGNORE_DELETE);
         TestKeyValueWatcher key1AfterStartNewWatcher = new TestKeyValueWatcher("key1AfterStartNewWatcher", false, META_ONLY, UPDATES_ONLY);
@@ -922,6 +924,8 @@ public void testWatch() throws Exception {
         TestKeyValueWatcher key1FromRevisionAfterWatcher = new TestKeyValueWatcher("key1FromRevisionAfterWatcher", false);
         TestKeyValueWatcher allFromRevisionAfterWatcher = new TestKeyValueWatcher("allFromRevisionAfterWatcher", false);
 
+        List<String> allKeys = Arrays.asList(TEST_WATCH_KEY_1, TEST_WATCH_KEY_2, TEST_WATCH_KEY_NULL);
+
         runInJsServer(nc -> {
             _testWatch(nc, key1FullWatcher, key1AllExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1FullWatcher, key1FullWatcher.watchOptions));
             _testWatch(nc, key1MetaWatcher, key1AllExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1MetaWatcher, key1MetaWatcher.watchOptions));
@@ -937,6 +941,8 @@ public void testWatch() throws Exception {
             _testWatch(nc, starMetaWatcher, allExpecteds, -1, kv -> kv.watch("key.*", starMetaWatcher, starMetaWatcher.watchOptions));
             _testWatch(nc, gtFullWatcher, allExpecteds, -1, kv -> kv.watch("key.>", gtFullWatcher, gtFullWatcher.watchOptions));
             _testWatch(nc, gtMetaWatcher, allExpecteds, -1, kv -> kv.watch("key.>", gtMetaWatcher, gtMetaWatcher.watchOptions));
+            _testWatch(nc, multipleFullWatcher, allExpecteds, -1, kv -> kv.watch(allKeys, multipleFullWatcher, multipleFullWatcher.watchOptions));
+            _testWatch(nc, multipleMetaWatcher, allExpecteds, -1, kv -> kv.watch(allKeys, multipleMetaWatcher, multipleMetaWatcher.watchOptions));
             _testWatch(nc, key1AfterWatcher, purgeOnlyExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1AfterWatcher, key1AfterWatcher.watchOptions));
             _testWatch(nc, key1AfterIgDelWatcher, noExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1AfterIgDelWatcher, key1AfterIgDelWatcher.watchOptions));
             _testWatch(nc, key1AfterStartNewWatcher, noExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1AfterStartNewWatcher, key1AfterStartNewWatcher.watchOptions));
diff --git a/src/test/java/io/nats/client/impl/ObjectStoreTests.java b/src/test/java/io/nats/client/impl/ObjectStoreTests.java
@@ -54,8 +54,8 @@ public void testWorkflow() throws Exception {
                 .build();
 
             ObjectStoreStatus status = osm.create(osc);
-            validateStatus(status, bucket, desc, metadata);
-            validateStatus(osm.getStatus(bucket), bucket, desc, metadata);
+            validateStatus(status, bucket, desc);
+            validateStatus(osm.getStatus(bucket), bucket, desc);
 
             JetStreamManagement jsm = nc.jetStreamManagement();
             assertNotNull(jsm.getStreamInfo("OBJ_" + bucket));
@@ -68,7 +68,7 @@ public void testWorkflow() throws Exception {
             ObjectStore os = nc.objectStore(bucket);
             nc.objectStore(bucket, ObjectStoreOptions.builder(DEFAULT_JS_OPTIONS).build()); // coverage;
 
-            validateStatus(os.getStatus(), bucket, desc, metadata);
+            validateStatus(os.getStatus(), bucket, desc);
 
             // object not found errors
             assertClientError(OsObjectNotFound, () -> os.get("notFound", new ByteArrayOutputStream()));
@@ -159,7 +159,7 @@ public void testWorkflow() throws Exception {
         });
     }
 
-    private static void validateStatus(ObjectStoreStatus status, String bucket, String desc, Map<String, String> metadata) {
+    private static void validateStatus(ObjectStoreStatus status, String bucket, String desc) {
         assertEquals(bucket, status.getBucketName());
         assertEquals(desc, status.getDescription());
         assertFalse(status.isSealed());
@@ -229,7 +229,7 @@ private static Object[] getInput(int size) throws IOException {
         File found = null;
         long foundLen = Long.MAX_VALUE;
         final String classPath = System.getProperty("java.class.path", ".");
-        final String[] classPathElements = classPath.split(System.getProperty("path.separator"));
+        final String[] classPathElements = classPath.split(File.pathSeparator);
         for(final String element : classPathElements){
             File f = new File(element);
             if (f.isFile()) {
diff --git a/src/test/java/io/nats/client/support/ValidatorTests.java b/src/test/java/io/nats/client/support/ValidatorTests.java
@@ -266,7 +266,7 @@ public void testValidateBucketName() {
     }
 
     @Test
-    public void testValidateWildcardKeyRequired() {
+    public void testValidateKvKeyWildcardAllowedRequired() {
         validateKvKeyWildcardAllowedRequired(PLAIN);
         validateKvKeyWildcardAllowedRequired(PLAIN.toUpperCase());
         validateKvKeyWildcardAllowedRequired(HAS_DASH);
@@ -277,14 +277,21 @@ public void testValidateWildcardKeyRequired() {
         validateKvKeyWildcardAllowedRequired(STAR_NOT_SEGMENT);
         validateKvKeyWildcardAllowedRequired(GT_NOT_SEGMENT);
         validateKvKeyWildcardAllowedRequired("numbers9ok");
-        assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(null));
+        String nullKey = null;
+        assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(nullKey));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(HAS_SPACE));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(HAS_DOLLAR));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(HAS_LOW));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(HAS_127));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(HAS_TIC));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired("colon:isbetween9andA"));
         assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(".starts.with.dot.not.allowed"));
+
+        List<String> nullList = null;
+        //noinspection ConstantValue
+        assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(nullList));
+        assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(Collections.singletonList(null)));
+        assertThrows(IllegalArgumentException.class, () -> validateKvKeyWildcardAllowedRequired(Collections.singletonList(HAS_SPACE)));
     }
 
     @Test

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@`
`21`	`21`	`import io.nats.client.api.ObjectStoreWatcher;`
`22`	`22`
`23`	`23`	`import java.io.IOException;`
	`24`	`+import java.util.Collections;`
`24`	`25`
`25`	`26`	`import static io.nats.client.api.ConsumerConfiguration.ULONG_UNSET;`
`26`	`27`
`@@ -58,6 +59,6 @@ public void onMessage(Message m) throws InterruptedException {`
`58`	`59`	`}`
`59`	`60`	`};`
`60`	`61`
`61`		`- finishInit(os, os.rawAllMetaSubject(), deliverPolicy, headersOnly, ULONG_UNSET, handler);`
	`62`	`+ finishInit(os, Collections.singletonList(os.rawAllMetaSubject()), deliverPolicy, headersOnly, ULONG_UNSET, handler);`
`62`	`63`	`}`
`63`	`64`	`}`
Original file line number	Diff line number	Diff line change
`@@ -124,6 +124,14 @@ public static String validatePrefixOrDomain(String s, String label, boolean requ`
`124`	`124`	`});`
`125`	`125`	`}`
`126`	`126`
	`127`	`+ public static List<String> validateKvKeyWildcardAllowedRequired(List<String> keys) {`
	`128`	`+ required(keys, "Key");`
	`129`	`+ for (String key : keys) {`
	`130`	`+ validateWildcardKvKey(key, "Key", true);`
	`131`	`+ }`
	`132`	`+ return keys;`
	`133`	`+ }`
	`134`	`+`
`127`	`135`	`public static String validateKvKeyWildcardAllowedRequired(String s) {`
`128`	`136`	`return validateWildcardKvKey(s, "Key", true);`
`129`	`137`	`}`