moved additional types to prevent circ deps in the module

Author: aricart

jetstream/consumer.ts

@@ -20,7 +20,7 @@ import {
   Timeout,
   timeout,
 } from "../nats-base-client/util.ts";
-import { ConsumerAPI, ConsumerAPIImpl } from "./jsmconsumer_api.ts";
+import { ConsumerAPIImpl } from "./jsmconsumer_api.ts";
 import { nuid } from "../nats-base-client/nuid.ts";
 import { isHeartbeatMsg } from "./jsutil.ts";
 import { QueuedIteratorImpl } from "../nats-base-client/queued_iterator.ts";
@@ -47,6 +47,7 @@ import {
   ConsumeMessages,
   ConsumeOptions,
   Consumer,
+  ConsumerAPI,
   ConsumerCallbackFn,
   ConsumerDebugEvents,
   ConsumerEvents,

jetstream/internal_mod.ts

@@ -45,6 +45,7 @@ export type {
   ConsumeMessages,
   ConsumeOptions,
   Consumer,
+  ConsumerAPI,
   ConsumerCallbackFn,
   ConsumerInfoable,
   ConsumerMessages,
@@ -78,6 +79,7 @@ export type {
   KvPutOptions,
   KvStatus,
   KvWatchOptions,
+  Lister,
   MaxBytes,
   MaxMessages,
   NextOptions,
@@ -142,7 +144,6 @@ export type {
 } from "./jsapi_types.ts";
 
 export type { JsMsg } from "./jsmsg.ts";
-export type { Lister } from "./jslister.ts";
 
 export {
   AckPolicy,
@@ -154,6 +155,5 @@ export {
   StoreCompression,
 } from "./jsapi_types.ts";
 
-export type { ConsumerAPI } from "./jsmconsumer_api.ts";
 export type { DeliveryInfo, StreamInfoRequestOptions } from "./jsapi_types.ts";
 export { jetstreamManager } from "./jsclient.ts";

jetstream/jsclient.ts

@@ -25,7 +25,7 @@ import {
   validateDurableName,
   validateStreamName,
 } from "./jsutil.ts";
-import { ConsumerAPI, ConsumerAPIImpl } from "./jsmconsumer_api.ts";
+import { ConsumerAPIImpl } from "./jsmconsumer_api.ts";
 import { JsMsg, JsMsgImpl, toJsMsg } from "./jsmsg.ts";
 import {
   MsgAdapter,
@@ -53,6 +53,7 @@ import { ConsumersImpl, StreamAPIImpl, StreamsImpl } from "./jsmstream_api.ts";
 import {
   Advisory,
   AdvisoryKind,
+  ConsumerAPI,
   ConsumerInfoable,
   ConsumerOpts,
   consumerOpts,

jetstream/jslister.ts

@@ -20,17 +20,7 @@ import {
   ConsumerListResponse,
   StreamListResponse,
 } from "./jsapi_types.ts";
-
-/**
- * An interface for listing. Returns a promise with typed list.
- */
-export interface Lister<T> {
-  [Symbol.asyncIterator](): AsyncIterator<T>;
-
-  next(): Promise<T[]>;
-}
-
-export type ListerFieldFilter<T> = (v: unknown) => T[];
+import { Lister, ListerFieldFilter } from "./types.ts";
 
 export class ListerImpl<T> implements Lister<T>, AsyncIterable<T> {
   err?: Error;
@@ -85,8 +75,7 @@ export class ListerImpl<T> implements Lister<T>, AsyncIterable<T> {
         return [];
       }
       this.offset += count;
-      const a = this.filter(r);
-      return a;
+      return this.filter(r);
     } catch (err) {
       this.err = err;
       throw err;
@@ -107,7 +96,6 @@ export class ListerImpl<T> implements Lister<T>, AsyncIterable<T> {
         // has to be a stream...
         return (r as StreamListResponse).streams?.length || 0;
     }
-    return 0;
   }
 
   async *[Symbol.asyncIterator]() {

jetstream/jsmconsumer_api.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 The NATS Authors
+ * Copyright 2021-2024 The NATS Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 import { BaseApiClientImpl } from "./jsbaseclient_api.ts";
-import { Lister, ListerFieldFilter, ListerImpl } from "./jslister.ts";
+import { ListerImpl } from "./jslister.ts";
 import {
   minValidation,
   validateDurableName,
@@ -32,60 +32,12 @@ import {
   SuccessResponse,
 } from "./jsapi_types.ts";
 
-import { JetStreamOptions } from "./types.ts";
-
-export interface ConsumerAPI {
-  /**
-   * Returns the ConsumerInfo for the specified consumer in the specified stream.
-   * @param stream
-   * @param consumer
-   */
-  info(stream: string, consumer: string): Promise<ConsumerInfo>;
-
-  /**
-   * Adds a new consumer to the specified stream with the specified consumer options.
-   * @param stream
-   * @param cfg
-   */
-  add(stream: string, cfg: Partial<ConsumerConfig>): Promise<ConsumerInfo>;
-
-  /**
-   * Updates the consumer configuration for the specified consumer on the specified
-   * stream that has the specified durable name.
-   * @param stream
-   * @param durable
-   * @param cfg
-   */
-  update(
-    stream: string,
-    durable: string,
-    cfg: Partial<ConsumerUpdateConfig>,
-  ): Promise<ConsumerInfo>;
-
-  /**
-   * Deletes the specified consumer name/durable from the specified stream.
-   * @param stream
-   * @param consumer
-   */
-  delete(stream: string, consumer: string): Promise<boolean>;
-
-  /**
-   * Lists all the consumers on the specfied streams
-   * @param stream
-   */
-  list(stream: string): Lister<ConsumerInfo>;
-
-  pause(
-    stream: string,
-    name: string,
-    until?: Date,
-  ): Promise<{ paused: boolean; pause_until?: string }>;
-
-  resume(
-    stream: string,
-    name: string,
-  ): Promise<{ paused: boolean; pause_until?: string }>;
-}
+import {
+  ConsumerAPI,
+  JetStreamOptions,
+  Lister,
+  ListerFieldFilter,
+} from "./types.ts";
 
 export class ConsumerAPIImpl extends BaseApiClientImpl implements ConsumerAPI {
   constructor(nc: NatsConnection, opts?: JetStreamOptions) {

jetstream/jsmstream_api.ts

@@ -15,7 +15,7 @@
 
 import { Empty, MsgHdrs } from "../nats-base-client/types.ts";
 import { BaseApiClientImpl, StreamNames } from "./jsbaseclient_api.ts";
-import { Lister, ListerFieldFilter, ListerImpl } from "./jslister.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";
@@ -26,10 +26,13 @@ import { Feature } from "../nats-base-client/semver.ts";
 import { NatsConnectionImpl } from "../nats-base-client/nats.ts";
 import {
   Consumer,
+  ConsumerAPI,
   Consumers,
   JetStreamOptions,
   kvPrefix,
   KvStatus,
+  Lister,
+  ListerFieldFilter,
   ObjectStoreStatus,
   OrderedConsumerOptions,
   StoredMsg,
@@ -58,7 +61,7 @@ import {
   SuccessResponse,
 } from "./jsapi_types.ts";
 import { OrderedPullConsumerImpl, PullConsumerImpl } from "./consumer.ts";
-import { ConsumerAPI, ConsumerAPIImpl } from "./jsmconsumer_api.ts";
+import { ConsumerAPIImpl } from "./jsmconsumer_api.ts";
 
 export function convertStreamSourceDomain(s?: StreamSource) {
   if (s === undefined) {
@@ -463,10 +466,9 @@ export class StreamAPIImpl extends BaseApiClientImpl implements StreamAPI {
       if (kvStreams.length) {
         cluster = this.nc.info?.cluster ?? "";
       }
-      const status = kvStreams.map((si) => {
+      return kvStreams.map((si) => {
         return new KvStatusImpl(si, cluster);
       });
-      return status;
     };
     const subj = `${this.prefix}.STREAM.LIST`;
     return new ListerImpl<KvStatus>(subj, filter, this);
@@ -483,10 +485,9 @@ export class StreamAPIImpl extends BaseApiClientImpl implements StreamAPI {
       objStreams.forEach((si) => {
         this._fixInfo(si);
       });
-      const status = objStreams.map((si) => {
+      return objStreams.map((si) => {
         return new ObjectStoreStatusImpl(si);
       });
-      return status;
     };
     const subj = `${this.prefix}.STREAM.LIST`;
     return new ListerImpl<ObjectStoreStatus>(subj, filter, this);

jetstream/kv.ts

@@ -755,7 +755,7 @@ export class Bucket implements KV, KvRemove {
   }
 
   canSetWatcherName(): boolean {
-    //@ts-ignore: don't import JetStreamClientImpl
+    //@ts-ignore: avoiding circular dependencies
     const nci = this.js.nc as NatsConnectionImpl;
     const { ok } = nci.features.get(
       Feature.JS_NEW_CONSUMER_CREATE_API,

jetstream/types.ts

@@ -30,6 +30,7 @@ import {
   AckPolicy,
   ConsumerConfig,
   ConsumerInfo,
+  ConsumerUpdateConfig,
   defaultConsumer,
   DeliverPolicy,
   DirectBatchOptions,
@@ -51,10 +52,7 @@ import {
   StreamUpdateConfig,
 } from "./jsapi_types.ts";
 import { JsMsg } from "./jsmsg.ts";
-// import { BaseApiClientImpl } from "./jsbaseclient_api.ts";
-import { ConsumerAPI } from "./jsmconsumer_api.ts";
 import { validateDurableName } from "./jsutil.ts";
-import { Lister } from "./jslister.ts";
 import { nanos } from "../nats-base-client/util.ts";
 import { NatsConnectionImpl } from "../nats-base-client/nats.ts";
 import { Codec } from "../nats-base-client/codec.ts";
@@ -241,6 +239,17 @@ export interface Views {
   ) => Promise<ObjectStore>;
 }
 
+/**
+ * An interface for listing. Returns a promise with typed list.
+ */
+export interface Lister<T> {
+  [Symbol.asyncIterator](): AsyncIterator<T>;
+
+  next(): Promise<T[]>;
+}
+
+export type ListerFieldFilter<T> = (v: unknown) => T[];
+
 export interface StreamAPI {
   /**
    * Returns the information about the specified stream
@@ -332,6 +341,59 @@ export interface StreamAPI {
   get(name: string): Promise<Stream>;
 }
 
+export interface ConsumerAPI {
+  /**
+   * Returns the ConsumerInfo for the specified consumer in the specified stream.
+   * @param stream
+   * @param consumer
+   */
+  info(stream: string, consumer: string): Promise<ConsumerInfo>;
+
+  /**
+   * Adds a new consumer to the specified stream with the specified consumer options.
+   * @param stream
+   * @param cfg
+   */
+  add(stream: string, cfg: Partial<ConsumerConfig>): Promise<ConsumerInfo>;
+
+  /**
+   * Updates the consumer configuration for the specified consumer on the specified
+   * stream that has the specified durable name.
+   * @param stream
+   * @param durable
+   * @param cfg
+   */
+  update(
+    stream: string,
+    durable: string,
+    cfg: Partial<ConsumerUpdateConfig>,
+  ): Promise<ConsumerInfo>;
+
+  /**
+   * Deletes the specified consumer name/durable from the specified stream.
+   * @param stream
+   * @param consumer
+   */
+  delete(stream: string, consumer: string): Promise<boolean>;
+
+  /**
+   * Lists all the consumers on the specfied streams
+   * @param stream
+   */
+  list(stream: string): Lister<ConsumerInfo>;
+
+  pause(
+    stream: string,
+    name: string,
+    until?: Date,
+  ): Promise<{ paused: boolean; pause_until?: string }>;
+
+  resume(
+    stream: string,
+    name: string,
+  ): Promise<{ paused: boolean; pause_until?: string }>;
+}
+
 /**
  * The API for interacting with JetStream resources
  */