Enhanced JetStreamManagement

Added ability to get KV/Management and OS/Management context directly from JetStreamManagement context

Author: scottf

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

@@ -375,4 +375,33 @@ public interface JetStreamManagement {
      * @return a JetStream instance.
      */
     JetStream jetStream();
-}
+
+    /**
+     * Gets a context for working with a Key Value bucket
+     * @param bucketName the bucket name
+     * @return a KeyValue instance.
+     * @throws IOException various IO exception such as timeout or interruption
+     */
+    KeyValue keyValue(String bucketName) throws IOException;
+
+    /**
+     * Gets a context for managing Key Value buckets
+     * @return a KeyValueManagement instance.
+     * @throws IOException various IO exception such as timeout or interruption
+     */
+    KeyValueManagement keyValueManagement() throws IOException;
+
+    /**
+     * Gets a context for working with an Object Store.
+     * @param bucketName the bucket name
+     * @return an ObjectStore instance.
+     * @throws IOException various IO exception such as timeout or interruption
+     */
+    ObjectStore objectStore(String bucketName) throws IOException;
+
+    /**
+     * Gets a context for managing Object Stores
+     * @return an ObjectStoreManagement instance.
+     * @throws IOException various IO exception such as timeout or interruption
+     */
+    ObjectStoreManagement objectStoreManagement() throws IOException;}

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

@@ -543,7 +543,8 @@ public class Options {
      * Property used to set class name for the Callback Executor Service (executor) class
      * {@link Builder#callbackExecutor(ExecutorService) callbackExecutor}.
      */
-    public static final String PROP_CALLBACK_EXECUTOR_SERVICE_CLASS = "callback.executor.service.class";    /**
+    public static final String PROP_CALLBACK_EXECUTOR_SERVICE_CLASS = "callback.executor.service.class";
+    /**
      * Property used to set class name for the Connect Thread Factory
      * {@link Builder#connectThreadFactory(ThreadFactory) connectThreadFactory}.
      */

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

@@ -2481,7 +2481,7 @@ public KeyValue keyValue(@NonNull String bucketName) throws IOException {
     public KeyValue keyValue(@NonNull String bucketName, @Nullable KeyValueOptions options) throws IOException {
         Validator.validateBucketName(bucketName, true);
         ensureNotClosing();
-        return new NatsKeyValue(this, bucketName, options);
+        return new NatsKeyValue(bucketName, this, options, null);
     }
 
     /**
@@ -2500,7 +2500,7 @@ public KeyValueManagement keyValueManagement() throws IOException {
     @NonNull
     public KeyValueManagement keyValueManagement(@Nullable KeyValueOptions options) throws IOException {
         ensureNotClosing();
-        return new NatsKeyValueManagement(this, options);
+        return new NatsKeyValueManagement(this, options, null);
     }
 
     /**
@@ -2520,7 +2520,7 @@ public ObjectStore objectStore(@NonNull String bucketName) throws IOException {
     public ObjectStore objectStore(@NonNull String bucketName, @Nullable ObjectStoreOptions options) throws IOException {
         Validator.validateBucketName(bucketName, true);
         ensureNotClosing();
-        return new NatsObjectStore(this, bucketName, options);
+        return new NatsObjectStore(bucketName, this, options, null);
     }
 
     /**
@@ -2530,7 +2530,7 @@ public ObjectStore objectStore(@NonNull String bucketName, @Nullable ObjectStore
     @NonNull
     public ObjectStoreManagement objectStoreManagement() throws IOException {
         ensureNotClosing();
-        return new NatsObjectStoreManagement(this, null);
+        return new NatsObjectStoreManagement(this, null, null);
     }
 
     /**
@@ -2540,7 +2540,7 @@ public ObjectStoreManagement objectStoreManagement() throws IOException {
     @NonNull
     public ObjectStoreManagement objectStoreManagement(@Nullable ObjectStoreOptions options) throws IOException {
         ensureNotClosing();
-        return new NatsObjectStoreManagement(this, options);
+        return new NatsObjectStoreManagement(this, options, null);
     }
 
     private void ensureNotClosing() throws IOException {

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

@@ -606,6 +606,9 @@ void encounteredProtocolError(Exception ex) throws IOException {
 
     //For testing
     void fakeReadForTest(byte[] bytes) {
+        for (int x = 0; x < this.buffer.length; x++) {
+            this.buffer[x] = 0;
+        }
         System.arraycopy(bytes, 0, this.buffer, 0, bytes.length);
         this.bufferPosition = 0;
         this.op = UNKNOWN_OP;

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

@@ -32,14 +32,18 @@ public class NatsFeatureBase {
     protected final NatsJetStreamManagement jsm;
     protected String streamName;
 
-    NatsFeatureBase(NatsConnection connection, FeatureOptions fo) throws IOException {
-        if (fo == null) {
+    NatsFeatureBase(NatsConnection connection, FeatureOptions fo, NatsJetStreamManagement jsm) throws IOException {
+        if (jsm != null) {
+            this.jsm = jsm;
+            js = (NatsJetStream)jsm.jetStream();
+        }
+        else if (fo == null) {
             js = new NatsJetStream(connection, null);
-            jsm = new NatsJetStreamManagement(connection, null);
+            this.jsm = new NatsJetStreamManagement(connection, null);
         }
         else {
             js = new NatsJetStream(connection, fo.getJetStreamOptions());
-            jsm = new NatsJetStreamManagement(connection, fo.getJetStreamOptions());
+            this.jsm = new NatsJetStreamManagement(connection, fo.getJetStreamOptions());
         }
     }
 

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

@@ -16,6 +16,7 @@
 import io.nats.client.*;
 import io.nats.client.api.*;
 import io.nats.client.api.Error;
+import io.nats.client.support.Validator;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
@@ -386,4 +387,26 @@ public JetStream jetStream() {
         }
         return js;
     }
+
+    @Override
+    public KeyValue keyValue(String bucketName) throws IOException {
+        Validator.validateBucketName(bucketName, true);
+        return new NatsKeyValue(bucketName, null, null, this);
+    }
+
+    @Override
+    public KeyValueManagement keyValueManagement() throws IOException {
+        return new NatsKeyValueManagement(null, null, this);
+    }
+
+    @Override
+    public ObjectStore objectStore(String bucketName) throws IOException {
+        Validator.validateBucketName(bucketName, true);
+        return new NatsObjectStore(bucketName, null, null, this);
+    }
+
+    @Override
+    public ObjectStoreManagement objectStoreManagement() throws IOException {
+        return new NatsObjectStoreManagement(null, null, this);
+    }
 }

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

@@ -40,13 +40,13 @@ public class NatsKeyValue extends NatsFeatureBase implements KeyValue {
     private final String readPrefix;
     private final String writePrefix;
 
-    NatsKeyValue(NatsConnection connection, String bucketName, KeyValueOptions kvo) throws IOException {
-        super(connection, kvo);
+    NatsKeyValue(String bucketName, NatsConnection connection, KeyValueOptions kvo, NatsJetStreamManagement jsm) throws IOException {
+        super(connection, kvo, jsm);
         this.bucketName = Validator.validateBucketName(bucketName, true);
         streamName = toStreamName(bucketName);
         StreamInfo si;
         try {
-             si = jsm.getStreamInfo(streamName);
+             si = this.jsm.getStreamInfo(streamName);
         } catch (JetStreamApiException e) {
             // can't throw directly, that would be a breaking change
             throw new IOException(e);

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

@@ -31,9 +31,14 @@ public class NatsKeyValueManagement implements KeyValueManagement {
     private final NatsJetStreamManagement jsm;
     private final boolean serverOlderThan272;
 
-    NatsKeyValueManagement(NatsConnection connection, KeyValueOptions kvo) throws IOException {
-        jsm = new NatsJetStreamManagement(connection, kvo == null ? null : kvo.getJetStreamOptions());
-        serverOlderThan272 = jsm.conn.getServerInfo().isOlderThanVersion("2.7.2");
+    NatsKeyValueManagement(NatsConnection connection, KeyValueOptions kvo, NatsJetStreamManagement jsm) throws IOException {
+        if (jsm == null) {
+            this.jsm = new NatsJetStreamManagement(connection, kvo == null ? null : kvo.getJetStreamOptions());
+        }
+        else {
+            this.jsm = jsm;
+        }
+        serverOlderThan272 = this.jsm.conn.getServerInfo().isOlderThanVersion("2.7.2");
     }
 
     /**

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

@@ -37,8 +37,8 @@ public class NatsObjectStore extends NatsFeatureBase implements ObjectStore {
     private final String rawChunkPrefix;
     private final String rawMetaPrefix;
 
-    NatsObjectStore(NatsConnection connection, String bucketName, ObjectStoreOptions oso) throws IOException {
-        super(connection, oso);
+    NatsObjectStore(String bucketName, NatsConnection connection, ObjectStoreOptions oso, NatsJetStreamManagement jsm) throws IOException {
+        super(connection, oso, jsm);
         this.oso = oso;
         this.bucketName = Validator.validateBucketName(bucketName, true);
         streamName = toStreamName(bucketName);

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

@@ -30,8 +30,13 @@
 public class NatsObjectStoreManagement implements ObjectStoreManagement {
     private final NatsJetStreamManagement jsm;
 
-    NatsObjectStoreManagement(NatsConnection connection, ObjectStoreOptions oso) throws IOException {
-        jsm = new NatsJetStreamManagement(connection, oso == null ? null : oso.getJetStreamOptions());
+    NatsObjectStoreManagement(NatsConnection connection, ObjectStoreOptions oso, NatsJetStreamManagement jsm) throws IOException {
+        if (jsm == null) {
+            this.jsm = new NatsJetStreamManagement(connection, oso == null ? null : oso.getJetStreamOptions());
+        }
+        else {
+            this.jsm = jsm;
+        }
     }
 
     /**