[CHANGE] [KV] is now its own module

Author: aricart

Makefile

@@ -6,12 +6,12 @@ lint:
 	deno lint --ignore=docs/,debug/
 
 test: clean
-	deno test --allow-all --parallel --reload --quiet --coverage=coverage tests/ jetstream/tests
+	deno test --allow-all --parallel --reload --quiet --coverage=coverage tests/ jetstream/tests kv/tests/
 	deno test --allow-all --parallel --reload --quiet --unsafely-ignore-certificate-errors --coverage=coverage unsafe_tests/
 
 
 testw: clean
-	deno test --allow-all --unstable --reload --parallel --watch --fail-fast tests/ jetstream/
+	deno test --allow-all --unstable --reload --parallel --watch --fail-fast tests/ jetstream/ kv/tests/
 
 cover:
 	deno coverage --unstable ./coverage --lcov > ./coverage/out.lcov
@@ -25,4 +25,4 @@ bundle:
 	deno bundle --log-level info --unstable src/mod.ts ./nats.js
 
 fmt:
-	deno fmt src/ doc/ bin/ nats-base-client/ examples/ tests/ debug/ unsafe_tests/ jetstream/ jetstream.md README.md services.md
+	deno fmt src/ doc/ bin/ nats-base-client/ examples/ tests/ debug/ unsafe_tests/ jetstream/ kv/ jetstream.md README.md migration.md services.md

README.md

@@ -11,7 +11,7 @@ A Deno client for the [NATS messaging system](https://nats.io).
 You can get the latest release version like this:
 
 ```typescript
-import * as nats from "https://deno.land/x/nats/src/mod.ts";
+import * as nats from "https://deno.land/x/nats/src/types.ts";
 ```
 
 To specify a specific released version, simply replace nats with
@@ -20,7 +20,7 @@ nats@_versionTag_.
 You can get the current development version by:
 
 ```typescript
-import * as nats from "https://raw.githubusercontent.com/nats-io/nats.deno/main/src/mod.ts";
+import * as nats from "https://raw.githubusercontent.com/nats-io/nats.deno/main/src/types.ts";
 ```
 
 ## Documentation
@@ -54,7 +54,7 @@ is working.
 
 ```typescript
 // import the connect function
-import { connect } from "../../src/mod.ts";
+import { connect } from "../../src/types.ts";
 
 const servers = [
   {},
@@ -123,7 +123,7 @@ the server.
 
 ```typescript
 // import the connect function
-import { connect, StringCodec } from "../../src/mod.ts";
+import { connect, StringCodec } from "../../src/types.ts";
 
 // to create a connection to a nats-server:
 const nc = await connect({ servers: "demo.nats.io:4222" });
@@ -191,7 +191,7 @@ All subscriptions are independent. If two different subscriptions match a
 subject, both will get to process the message:
 
 ```javascript
-import { connect, StringCodec, Subscription } from "../../src/mod.ts";
+import { connect, StringCodec, Subscription } from "../../src/types.ts";
 const nc = await connect({ servers: "demo.nats.io:4222" });
 const sc = StringCodec();
 
@@ -244,7 +244,7 @@ simply to illustrate not only how to create responses, but how the subject
 itself is used to dispatch different behaviors.
 
 ```typescript
-import { connect, StringCodec, Subscription } from "../../src/mod.ts";
+import { connect, StringCodec, Subscription } from "../../src/types.ts";
 
 // create a connection
 const nc = await connect({ servers: "demo.nats.io" });
@@ -319,7 +319,7 @@ Here's a simple example of a client making a simple request from the service
 above:
 
 ```typescript
-import { connect, Empty, StringCodec } from "../../src/mod.ts";
+import { connect, Empty, StringCodec } from "../../src/types.ts";
 
 // create a connection
 const nc = await connect({ servers: "demo.nats.io:4222" });
@@ -357,7 +357,7 @@ import {
   NatsConnection,
   StringCodec,
   Subscription,
-} from "../../src/mod.ts";
+} from "../../src/types.ts";
 
 // this is the definition of a service with `count` members in them.
 // if the queue is specified, the they will be part of a queue
@@ -436,7 +436,7 @@ are publishing a message with a header, it is possible for the recipient to not
 support them.
 
 ```typescript
-import { connect, createInbox, Empty, headers } from "../../src/mod.ts";
+import { connect, createInbox, Empty, headers } from "../../src/types.ts";
 import { nuid } from "../../nats-base-client/nuid.ts";
 
 const nc = await connect(
@@ -525,7 +525,7 @@ Setting the `user`/`pass` or `token` options, simply initializes an
 ```typescript
 // if the connection requires authentication, provide `user` and `pass` or
 // `token` options in the NatsConnectionOptions
-import { connect } from "src/mod.ts";
+import { connect } from "src/types.ts";
 
 const nc1 = await connect({
   servers: "127.0.0.1:4222",

jetstream/internal_mod.ts

@@ -22,18 +22,10 @@ export {
   DirectMsgHeaders,
   isConsumerOptsBuilder,
   JsHeaders,
-  KvWatchInclude,
   RepublishHeaders,
 } from "./types.ts";
 
-export { jetstream } from "./jsclient.ts";
-
-export {
-  Base64KeyCodec,
-  Bucket,
-  defaultBucketOpts,
-  NoopKvCodecs,
-} from "./kv.ts";
+export { jetstream, JetStreamClientImpl } from "./jsclient.ts";
 
 export type {
   AbortOnMissingResource,
@@ -54,6 +46,7 @@ export type {
   Consumers,
   ConsumerStatus,
   Destroyable,
+  DirectStreamAPI,
   Expires,
   FetchBytes,
   FetchMessages,
@@ -69,16 +62,6 @@ export type {
   JetStreamSubscriptionInfoable,
   JetStreamSubscriptionOptions,
   JsMsgCallback,
-  KV,
-  KvCodec,
-  KvCodecs,
-  KvDeleteOptions,
-  KvEntry,
-  KvLimits,
-  KvOptions,
-  KvPutOptions,
-  KvStatus,
-  KvWatchOptions,
   Lister,
   MaxBytes,
   MaxMessages,
@@ -95,7 +78,6 @@ export type {
   OrderedConsumerOptions,
   PubAck,
   Pullable,
-  RoKV,
   StoredMsg,
   Stream,
   StreamAPI,
@@ -149,6 +131,7 @@ export {
   AckPolicy,
   DeliverPolicy,
   DiscardPolicy,
+  PubHeaders,
   ReplayPolicy,
   RetentionPolicy,
   StorageType,

jetstream/jsbaseclient_api.ts

@@ -120,6 +120,9 @@ export class BaseApiClientImpl implements BaseClient {
       const err = checkJsErrorCode(r.error.code, r.error.description);
       if (err !== null) {
         err.api_error = r.error;
+        if (r.error.description !== "") {
+          err.message = r.error.description;
+        }
         throw err;
       }
     }

jetstream/jsclient.ts

@@ -45,7 +45,6 @@ import {
   timeout,
 } from "../nats-base-client/util.ts";
 import { headers } from "../nats-base-client/headers.ts";
-import { Bucket } from "./kv.ts";
 import { Feature } from "../nats-base-client/semver.ts";
 import { ObjectStoreImpl } from "./objectstore.ts";
 import { IdleHeartbeatMonitor } from "../nats-base-client/idleheartbeat_monitor.ts";
@@ -73,8 +72,6 @@ import {
   JetStreamSubscriptionInfoable,
   JetStreamSubscriptionOptions,
   JsHeaders,
-  KV,
-  KvOptions,
   ObjectStore,
   ObjectStoreOptions,
   PubAck,
@@ -122,15 +119,14 @@ export function jetstream(
   return new JetStreamClientImpl(nc, opts);
 }
 
-
 /**
  * Returns a {@link JetStreamManager} supported by the specified NatsConnection
  * @param nc
  * @param opts
  */
 export async function jetstreamManager(
-    nc: NatsConnection,
-    opts: JetStreamOptions | JetStreamManagerOptions = {},
+  nc: NatsConnection,
+  opts: JetStreamOptions | JetStreamManagerOptions = {},
 ): Promise<JetStreamManager> {
   const adm = new JetStreamManagerImpl(nc, opts);
   if ((opts as JetStreamManagerOptions).checkAPI !== false) {
@@ -152,20 +148,20 @@ class ViewsImpl implements Views {
   constructor(js: JetStreamClientImpl) {
     this.js = js;
   }
-  kv(name: string, opts: Partial<KvOptions> = {}): Promise<KV> {
-    const jsi = this.js as JetStreamClientImpl;
-    const { ok, min } = jsi.nc.features.get(Feature.JS_KV);
-    if (!ok) {
-      return Promise.reject(
-        new Error(`kv is only supported on servers ${min} or better`),
-      );
-    }
-    if (opts.bindOnly) {
-      return Bucket.bind(this.js, name, opts);
-    }
-
-    return Bucket.create(this.js, name, opts);
-  }
+  // kv(name: string, opts: Partial<KvOptions> = {}): Promise<KV> {
+  //   const jsi = this.js as JetStreamClientImpl;
+  //   const { ok, min } = jsi.nc.features.get(Feature.JS_KV);
+  //   if (!ok) {
+  //     return Promise.reject(
+  //       new Error(`kv is only supported on servers ${min} or better`),
+  //     );
+  //   }
+  //   if (opts.bindOnly) {
+  //     return Bucket.bind(this.js, name, opts);
+  //   }
+  //
+  //   return Bucket.create(this.js, name, opts);
+  // }
   os(
     name: string,
     opts: Partial<ObjectStoreOptions> = {},

jetstream/jsmstream_api.ts

@@ -18,7 +18,6 @@ import { BaseApiClientImpl, StreamNames } from "./jsbaseclient_api.ts";
 import { ListerImpl } from "./jslister.ts";
 import { validateStreamName } from "./jsutil.ts";
 import { headers, MsgHdrsImpl } from "../nats-base-client/headers.ts";
-import { KvStatusImpl } from "./kv.ts";
 import { ObjectStoreStatusImpl, osPrefix } from "./objectstore.ts";
 import { Codec, JSONCodec } from "../nats-base-client/codec.ts";
 import { TD } from "../nats-base-client/encoders.ts";
@@ -29,8 +28,6 @@ import {
   ConsumerAPI,
   Consumers,
   JetStreamOptions,
-  kvPrefix,
-  KvStatus,
   Lister,
   ListerFieldFilter,
   ObjectStoreStatus,
@@ -451,29 +448,6 @@ export class StreamAPIImpl extends BaseApiClientImpl implements StreamAPI {
     return this.findStream(subject);
   }
 
-  listKvs(): Lister<KvStatus> {
-    const filter: ListerFieldFilter<KvStatus> = (
-      v: unknown,
-    ): KvStatus[] => {
-      const slr = v as StreamListResponse;
-      const kvStreams = slr.streams.filter((v) => {
-        return v.config.name.startsWith(kvPrefix);
-      });
-      kvStreams.forEach((si) => {
-        this._fixInfo(si);
-      });
-      let cluster = "";
-      if (kvStreams.length) {
-        cluster = this.nc.info?.cluster ?? "";
-      }
-      return kvStreams.map((si) => {
-        return new KvStatusImpl(si, cluster);
-      });
-    };
-    const subj = `${this.prefix}.STREAM.LIST`;
-    return new ListerImpl<KvStatus>(subj, filter, this);
-  }
-
   listObjectStores(): Lister<ObjectStoreStatus> {
     const filter: ListerFieldFilter<ObjectStoreStatus> = (
       v: unknown,

jetstream/mod.ts

@@ -29,7 +29,6 @@ export {
   DirectMsgHeaders,
   DiscardPolicy,
   JsHeaders,
-  KvWatchInclude,
   ReplayPolicy,
   RepublishHeaders,
   RetentionPolicy,
@@ -82,16 +81,6 @@ export type {
   JetStreamUsageAccountLimits,
   JsMsg,
   JsMsgCallback,
-  KV,
-  KvCodec,
-  KvCodecs,
-  KvDeleteOptions,
-  KvEntry,
-  KvLimits,
-  KvOptions,
-  KvPutOptions,
-  KvStatus,
-  KvWatchOptions,
   LastForMsgRequest,
   Lister,
   LostStreamData,
@@ -121,7 +110,6 @@ export type {
   PurgeResponse,
   PurgeTrimOpts,
   Republish,
-  RoKV,
   SeqMsgRequest,
   SequenceInfo,
   StoredMsg,

jetstream/objectstore.ts

@@ -13,7 +13,6 @@
  * limitations under the License.
  */
 
-import { validateBucket } from "./kv.ts";
 import { Base64UrlPaddedCodec } from "../nats-base-client/base64.ts";
 import { JSONCodec } from "../nats-base-client/codec.ts";
 import { nuid } from "../nats-base-client/nuid.ts";
@@ -114,6 +113,12 @@ export class ObjectStoreStatusImpl implements ObjectStoreStatus {
     return false;
   }
 }
+export function validateBucket(name: string) {
+  const validBucketRe = /^[-\w]+$/;
+  if (!validBucketRe.test(name)) {
+    throw new Error(`invalid bucket name: ${name}`);
+  }
+}
 
 export type ServerObjectStoreMeta = {
   name: string;