Skip to content

Commit 73364d2

Browse files
authored
Watch test 284 (#1117)
1 parent cda5212 commit 73364d2

File tree

3 files changed

+18
-21
lines changed

3 files changed

+18
-21
lines changed

src/main/java/io/nats/client/KeyValue.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -168,8 +168,8 @@ public interface KeyValue {
168168
void purge(String key, long expectedRevision) throws IOException, JetStreamApiException;
169169

170170
/**
171-
* Watch updates for a specific key or keys.
172-
* @param key the key or a comma delimited list of keys.
171+
* Watch updates for a specific key.
172+
* @param key the key.
173173
* @param watcher the watcher the implementation to receive changes
174174
* @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
175175
* @return The KeyValueWatchSubscription
@@ -181,8 +181,8 @@ public interface KeyValue {
181181
NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException;
182182

183183
/**
184-
* Watch updates for a specific key or keys, starting at a specific revision.
185-
* @param key the key or a comma delimited list of keys.
184+
* Watch updates for a specific key, starting at a specific revision.
185+
* @param key the key.
186186
* @param watcher the watcher the implementation to receive changes
187187
* @param fromRevision the revision to start from
188188
* @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.

src/main/java/io/nats/client/impl/NatsKeyValue.java

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import java.nio.charset.StandardCharsets;
2626
import java.time.ZonedDateTime;
2727
import java.util.ArrayList;
28-
import java.util.Arrays;
2928
import java.util.Collections;
3029
import java.util.List;
3130

@@ -239,28 +238,27 @@ private PublishAck _write(String key, byte[] data, Headers h) throws IOException
239238

240239
@Override
241240
public NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
242-
if (key.contains(",")) {
243-
return watch(Arrays.asList(key.split(",")), watcher, -1, watchOptions);
244-
}
245-
return watch(Collections.singletonList(key), watcher, -1, watchOptions);
241+
validateKvKeyWildcardAllowedRequired(key);
242+
validateNotNull(watcher, "Watcher is required");
243+
return new NatsKeyValueWatchSubscription(this, Collections.singletonList(key), watcher, -1, watchOptions);
246244
}
247245

248246
@Override
249247
public NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
250-
if (key.contains(",")) {
251-
return watch(Arrays.asList(key.split(",")), watcher, fromRevision, watchOptions);
252-
}
253-
return watch(Collections.singletonList(key), watcher, fromRevision, watchOptions);
248+
validateKvKeyWildcardAllowedRequired(key);
249+
validateNotNull(watcher, "Watcher is required");
250+
return new NatsKeyValueWatchSubscription(this, Collections.singletonList(key), watcher, fromRevision, watchOptions);
254251
}
255252

256253
@Override
257254
public NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
258-
return watch(keys, watcher, -1, watchOptions);
255+
validateKvKeysWildcardAllowedRequired(keys);
256+
validateNotNull(watcher, "Watcher is required");
257+
return new NatsKeyValueWatchSubscription(this, keys, watcher, -1, watchOptions);
259258
}
260259

261260
@Override
262261
public NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) throws IOException, JetStreamApiException, InterruptedException {
263-
// all watch methods (watch, watchAll) delegate to here
264262
validateKvKeysWildcardAllowedRequired(keys);
265263
validateNotNull(watcher, "Watcher is required");
266264
return new NatsKeyValueWatchSubscription(this, keys, watcher, fromRevision, watchOptions);

src/test/java/io/nats/client/impl/KeyValueTests.java

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -914,8 +914,6 @@ public void testWatch() throws Exception {
914914
TestKeyValueWatcher gtMetaWatcher = new TestKeyValueWatcher("gtMetaWatcher", true, META_ONLY);
915915
TestKeyValueWatcher multipleFullWatcher = new TestKeyValueWatcher("multipleFullWatcher", true);
916916
TestKeyValueWatcher multipleMetaWatcher = new TestKeyValueWatcher("multipleMetaWatcher", true, META_ONLY);
917-
TestKeyValueWatcher multipleFullWatcher2 = new TestKeyValueWatcher("multipleFullWatcher2", true);
918-
TestKeyValueWatcher multipleMetaWatcher2 = new TestKeyValueWatcher("multipleMetaWatcher2", true, META_ONLY);
919917
TestKeyValueWatcher key1AfterWatcher = new TestKeyValueWatcher("key1AfterWatcher", false, META_ONLY);
920918
TestKeyValueWatcher key1AfterIgDelWatcher = new TestKeyValueWatcher("key1AfterIgDelWatcher", false, META_ONLY, IGNORE_DELETE);
921919
TestKeyValueWatcher key1AfterStartNewWatcher = new TestKeyValueWatcher("key1AfterStartNewWatcher", false, META_ONLY, UPDATES_ONLY);
@@ -943,10 +941,6 @@ public void testWatch() throws Exception {
943941
_testWatch(nc, starMetaWatcher, allExpecteds, -1, kv -> kv.watch("key.*", starMetaWatcher, starMetaWatcher.watchOptions));
944942
_testWatch(nc, gtFullWatcher, allExpecteds, -1, kv -> kv.watch("key.>", gtFullWatcher, gtFullWatcher.watchOptions));
945943
_testWatch(nc, gtMetaWatcher, allExpecteds, -1, kv -> kv.watch("key.>", gtMetaWatcher, gtMetaWatcher.watchOptions));
946-
_testWatch(nc, multipleFullWatcher, allExpecteds, -1, kv -> kv.watch(allKeys, multipleFullWatcher, multipleFullWatcher.watchOptions));
947-
_testWatch(nc, multipleMetaWatcher, allExpecteds, -1, kv -> kv.watch(allKeys, multipleMetaWatcher, multipleMetaWatcher.watchOptions));
948-
_testWatch(nc, multipleFullWatcher2, allExpecteds, -1, kv -> kv.watch(String.join(",", allKeys), multipleFullWatcher2, multipleFullWatcher.watchOptions));
949-
_testWatch(nc, multipleMetaWatcher2, allExpecteds, -1, kv -> kv.watch(String.join(",", allKeys), multipleMetaWatcher2, multipleMetaWatcher.watchOptions));
950944
_testWatch(nc, key1AfterWatcher, purgeOnlyExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1AfterWatcher, key1AfterWatcher.watchOptions));
951945
_testWatch(nc, key1AfterIgDelWatcher, noExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1AfterIgDelWatcher, key1AfterIgDelWatcher.watchOptions));
952946
_testWatch(nc, key1AfterStartNewWatcher, noExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_1, key1AfterStartNewWatcher, key1AfterStartNewWatcher.watchOptions));
@@ -956,6 +950,11 @@ public void testWatch() throws Exception {
956950
_testWatch(nc, key2AfterStartFirstWatcher, key2AllExpecteds, -1, kv -> kv.watch(TEST_WATCH_KEY_2, key2AfterStartFirstWatcher, key2AfterStartFirstWatcher.watchOptions));
957951
_testWatch(nc, key1FromRevisionAfterWatcher, key1FromRevisionExpecteds, 2, kv -> kv.watch(TEST_WATCH_KEY_1, key1FromRevisionAfterWatcher, 2, key1FromRevisionAfterWatcher.watchOptions));
958952
_testWatch(nc, allFromRevisionAfterWatcher, allFromRevisionExpecteds, 2, kv -> kv.watchAll(allFromRevisionAfterWatcher, 2, allFromRevisionAfterWatcher.watchOptions));
953+
954+
if (atLeast2_10()) {
955+
_testWatch(nc, multipleFullWatcher, allExpecteds, -1, kv -> kv.watch(allKeys, multipleFullWatcher, multipleFullWatcher.watchOptions));
956+
_testWatch(nc, multipleMetaWatcher, allExpecteds, -1, kv -> kv.watch(allKeys, multipleMetaWatcher, multipleMetaWatcher.watchOptions));
957+
}
959958
});
960959
}
961960

0 commit comments

Comments
 (0)