kv watch first try

src/main/java/io/nats/vertx/NatsClient.java

@@ -49,15 +49,15 @@ static NatsClient create(final NatsOptions natsOptions) {
      * @param bucketName the bucket name
      * @return Key Value instance
      */
-    Future<NatsKeyValue> keyValue(String bucketName);
+    Future<NatsVertxKeyValue> keyValue(String bucketName);
 
     /**
      * Get interface to Key Value.
      * @param bucketName the bucket name
      * @param options KeyValue options.
      * @return Key Value instance
      */
-    Future<NatsKeyValue> keyValue(String bucketName, KeyValueOptions options);
+    Future<NatsVertxKeyValue> keyValue(String bucketName, KeyValueOptions options);
 
     /**
      * Drain handler.

src/main/java/io/nats/vertx/NatsStream.java

@@ -82,7 +82,7 @@ Future<Void> subscribe(
     /**
      * Subscribe to JetStream stream
      * @param subject The subject of the stream.
-     * @param queue The queue name to share messages accross consumers with the same queue name.
+     * @param queue The queue name to share messages across consumers with the same queue name.
      * @param handler The message handler to listen to messages from the stream.
      * @param autoAck Specify if message handler should auto acknowledge.
      * @param so The PushSubscribeOptions

src/main/java/io/nats/vertx/NatsKeyValue.java → src/main/java/io/nats/vertx/NatsVertxKeyValue.java

@@ -2,6 +2,7 @@
 
 import io.nats.client.api.*;
 import io.nats.client.impl.NatsKeyValueWatchSubscription;
+import io.nats.vertx.impl.NatsImpl;
 import io.vertx.core.Future;
 import io.vertx.core.streams.StreamBase;
 
@@ -10,7 +11,7 @@
 /**
  * Provides a Vert.x WriteStream interface with Futures and Promises.
  */
-public interface NatsKeyValue extends StreamBase {
+public interface NatsVertxKeyValue extends StreamBase {
 
     /**
      * Get the name of the bucket.
@@ -127,7 +128,7 @@ public interface NatsKeyValue extends StreamBase {
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
      * @return The KeyValueWatchSubscription
      */
-    NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions);
+    Future<NatsKeyValueWatchSubscription> watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions);
 
     /**
      * Watch updates for a specific key, starting at a specific revision.
@@ -137,7 +138,7 @@ public interface NatsKeyValue extends StreamBase {
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
      * @return The KeyValueWatchSubscription
      */
-    NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions);
+    Future<NatsKeyValueWatchSubscription> watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions);
 
     /**
      * Watch updates for specific keys.
@@ -146,7 +147,7 @@ public interface NatsKeyValue extends StreamBase {
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
      * @return The KeyValueWatchSubscription
      */
-    NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions);
+    Future<NatsKeyValueWatchSubscription> watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions);
 
     /**
      * Watch updates for specific keys, starting at a specific revision.
@@ -156,15 +157,15 @@ public interface NatsKeyValue extends StreamBase {
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
      * @return The KeyValueWatchSubscription
      */
-    NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions);
+    Future<NatsKeyValueWatchSubscription> watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions);
 
     /**
      * Watch updates for all 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
      */
-    NatsKeyValueWatchSubscription watchAll(KeyValueWatcher watcher, KeyValueWatchOption... watchOptions);
+    Future<NatsKeyValueWatchSubscription> watchAll(KeyValueWatcher watcher, KeyValueWatchOption... watchOptions);
 
     /**
      * Watch updates for all keys starting from a specific revision
@@ -173,7 +174,7 @@ public interface NatsKeyValue extends StreamBase {
      * @param watchOptions the watch options to apply. If multiple conflicting options are supplied, the last options wins.
      * @return The KeyValueWatchSubscription
      */
-    NatsKeyValueWatchSubscription watchAll(KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions);
+    Future<NatsKeyValueWatchSubscription> watchAll(KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions);
 
     /**
      * Get a list of the keys in a bucket.
@@ -221,4 +222,15 @@ public interface NatsKeyValue extends StreamBase {
      * @return the status object
      */
     Future<KeyValueStatus> getStatus();
+
+    /**
+     * Get access to the underlying implementation which contains
+     * the connection, the JetStream context, and other components
+     * @return the NatsImp object
+     */
+    NatsImpl getImpl();
+
+    String readSubject(String key);
+
+    String writeSubject(String key);
 }

src/main/java/io/nats/vertx/impl/NatsClientImpl.java

@@ -4,9 +4,9 @@
 import io.nats.client.impl.Headers;
 import io.nats.client.impl.VertxDispatcherFactory;
 import io.nats.vertx.NatsClient;
-import io.nats.vertx.NatsKeyValue;
 import io.nats.vertx.NatsOptions;
 import io.nats.vertx.NatsStream;
+import io.nats.vertx.NatsVertxKeyValue;
 import io.vertx.core.*;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.streams.WriteStream;
@@ -161,17 +161,17 @@ public Future<NatsStream> jetStream(final JetStreamOptions jso) {
     }
 
     @Override
-    public Future<NatsKeyValue> keyValue(String bucketName) {
+    public Future<NatsVertxKeyValue> keyValue(String bucketName) {
         return keyValue(bucketName, null);
     }
 
     @Override
-    public Future<NatsKeyValue> keyValue(String bucketName, KeyValueOptions kvo) {
-        final Promise<NatsKeyValue> promise = context().promise();
+    public Future<NatsVertxKeyValue> keyValue(String bucketName, KeyValueOptions kvo) {
+        final Promise<NatsVertxKeyValue> promise = context().promise();
         context().executeBlocking(event -> {
             try {
                 promise.complete(
-                    new NatsKeyValueImpl(connection.get(), vertx, exceptionHandler.get(), bucketName, kvo));
+                    new NatsVertxKeyValueImpl(connection.get(), vertx, exceptionHandler.get(), bucketName, kvo));
             } catch (Exception e) {
                 handleException(promise, e);
             }

src/main/java/io/nats/vertx/impl/NatsImpl.java

@@ -22,16 +22,16 @@
  */
 public class NatsImpl {
 
-    protected final Vertx vertx;
+    public final Vertx vertx;
 
     protected final ConcurrentHashMap<String, Dispatcher> dispatcherMap = new ConcurrentHashMap<>();
     protected final ConcurrentHashMap<String, JetStreamSubscription> subscriptionMap = new ConcurrentHashMap<>();
 
-    protected final Connection conn;
-    protected final Duration timeout;
-    protected final JetStreamOptions jso;
-    protected final JetStreamManagement jsm;
-    protected final JetStream js;
+    public final Connection conn;
+    public final Duration timeout;
+    public final JetStreamOptions jso;
+    public final JetStreamManagement jsm;
+    public final JetStream js;
 
     protected final AtomicReference<Handler<Throwable>> exceptionHandler = new AtomicReference<>();
 
@@ -59,8 +59,8 @@ protected NatsImpl(final Connection conn,
 
     }
 
-    protected ContextInternal context() {
-        return (ContextInternal)  vertx.getOrCreateContext();
+    public ContextInternal context() {
+        return (ContextInternal)vertx.getOrCreateContext();
     }
 
     protected void endImpl(Handler<AsyncResult<Void>> handler) {
@@ -69,7 +69,7 @@ protected void endImpl(Handler<AsyncResult<Void>> handler) {
         handler.handle(promise.future());
     }
 
-    protected void handleException(Promise<?> promise, Exception e) {
+    public void handleException(Promise<?> promise, Exception e) {
         promise.fail(e);
         exceptionHandler.get().handle(e);
     }

src/main/java/io/nats/vertx/impl/NatsKeyValueImpl.java → src/main/java/io/nats/vertx/impl/NatsVertxKeyValueImpl.java

@@ -7,7 +7,7 @@
 import io.nats.client.impl.NatsMessage;
 import io.nats.client.support.DateTimeUtils;
 import io.nats.client.support.Validator;
-import io.nats.vertx.NatsKeyValue;
+import io.nats.vertx.NatsVertxKeyValue;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
@@ -23,20 +23,21 @@
 import static io.nats.client.support.NatsConstants.GREATER_THAN;
 import static io.nats.client.support.NatsJetStreamConstants.*;
 import static io.nats.client.support.NatsKeyValueUtil.*;
-import static io.nats.client.support.Validator.validateNonWildcardKvKeyRequired;
+import static io.nats.client.support.Validator.*;
 
 
 /**
  * NATS Client implementation.
  */
-public class NatsKeyValueImpl extends NatsImpl implements NatsKeyValue {
+public class NatsVertxKeyValueImpl extends NatsImpl implements NatsVertxKeyValue {
 
     // JNats KeyValue parallel variables
     private final String bucketName;
     private final String streamName;
     private final String streamSubject;
     private final String readPrefix;
     private final String writePrefix;
+    private final KeyValue kv;
 
     /**
      * Create instance
@@ -47,11 +48,11 @@ public class NatsKeyValueImpl extends NatsImpl implements NatsKeyValue {
      * @param bucketName
      * @param kvo              keyValueOptions also contains jetStreamOptions use to make jetstream/management implementations
      */
-    public NatsKeyValueImpl(final Connection conn,
-                            final Vertx vertx,
-                            final Handler<Throwable> exceptionHandler,
-                            final String bucketName,
-                            final KeyValueOptions kvo)
+    public NatsVertxKeyValueImpl(final Connection conn,
+                                 final Vertx vertx,
+                                 final Handler<Throwable> exceptionHandler,
+                                 final String bucketName,
+                                 final KeyValueOptions kvo)
     {
         super(conn, vertx, exceptionHandler, kvo == null ? null : kvo.getJetStreamOptions());
 
@@ -65,20 +66,37 @@ public NatsKeyValueImpl(final Connection conn,
         }
         else {
             writePrefix = kvo.getJetStreamOptions().getPrefix() + readPrefix;
+
+        }
+        try {
+            this.kv = conn.keyValue(bucketName, kvo);
+        }
+        catch (IOException e) {
+            if (exceptionHandler != null) {
+                exceptionHandler.handle(e);
+            }
+            throw new RuntimeException(e);
         }
     }
 
     @Override
-    public NatsKeyValueImpl exceptionHandler(Handler<Throwable> handler) {
+    public NatsImpl getImpl() {
+        return this;
+    }
+
+    @Override
+    public NatsVertxKeyValueImpl exceptionHandler(Handler<Throwable> handler) {
         exceptionHandler.set(handler);
         return this;
     }
 
-    String readSubject(String key) {
+    @Override
+    public String readSubject(String key) {
         return readPrefix + key;
     }
 
-    String writeSubject(String key) {
+    @Override
+    public String writeSubject(String key) {
         return writePrefix + key;
     }
 
@@ -185,33 +203,43 @@ public Future<Void> purge(String key, long expectedRevision) {
     }
 
     @Override
-    public NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) {
-        return null;
+    public Future<NatsKeyValueWatchSubscription> watch(String key, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) {
+        return watch(Collections.singletonList(key), watcher, -1, watchOptions);
     }
 
     @Override
-    public NatsKeyValueWatchSubscription watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) {
-        return null;
+    public Future<NatsKeyValueWatchSubscription> watch(String key, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) {
+        return watch(Collections.singletonList(key), watcher, -1, watchOptions);
     }
 
     @Override
-    public NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) {
-        return null;
+    public Future<NatsKeyValueWatchSubscription> watch(List<String> keys, KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) {
+        return watch(keys, watcher, -1, watchOptions);
     }
 
     @Override
-    public NatsKeyValueWatchSubscription watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) {
-        return null;
+    public Future<NatsKeyValueWatchSubscription> watch(List<String> keys, KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) {
+        final Promise<NatsKeyValueWatchSubscription> promise = context().promise();
+        context().executeBlocking(event -> {
+            try {
+                validateKvKeysWildcardAllowedRequired(keys);
+                validateNotNull(watcher, "Watcher is required");
+                promise.complete(kv.watch(keys, watcher, fromRevision, watchOptions));
+            } catch (Exception e) {
+                handleException(promise, e);
+            }
+        }, false);
+        return promise.future();
     }
 
     @Override
-    public NatsKeyValueWatchSubscription watchAll(KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) {
-        return null;
+    public Future<NatsKeyValueWatchSubscription> watchAll(KeyValueWatcher watcher, KeyValueWatchOption... watchOptions) {
+        return watch(Collections.singletonList(GREATER_THAN), watcher, -1, watchOptions);
     }
 
     @Override
-    public NatsKeyValueWatchSubscription watchAll(KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) {
-        return null;
+    public Future<NatsKeyValueWatchSubscription> watchAll(KeyValueWatcher watcher, long fromRevision, KeyValueWatchOption... watchOptions) {
+        return watch(Collections.singletonList(GREATER_THAN), watcher, fromRevision, watchOptions);
     }
 
     public Future<List<String>> keys() {