fix: grow subscriptions (#130)

Author: gleip

src/Client.ts

@@ -81,7 +81,7 @@ export class Client<E extends Emitter = Emitter> extends Root {
   private async startBatchWatch(fetcher: StreamFetcher, listener: EventEmitter, eventName: string) {
     while (true) {
       const batch: Partial<EmitterStreamEvent<any>>[] = [];
-      const events = fetcher.fetch();
+      const events = await fetcher.fetch();
 
       for await (const event of events) {
         let data: unknown;
@@ -97,7 +97,7 @@ export class Client<E extends Emitter = Emitter> extends Root {
         batch.push(message);
       }
       if (batch.length > 0) listener.emit(eventName, batch);
-      events.stop();
+      await events.close();
     }
   }
 

src/StreamFetcher.ts

@@ -1,4 +1,4 @@
-import { JetStreamClient, JsMsg, QueuedIterator } from 'nats';
+import { ConsumerMessages, JetStreamClient } from 'nats';
 
 interface BatcherOptions {
   batchSize?: number;
@@ -13,11 +13,11 @@ export class StreamFetcher {
     private options: BatcherOptions,
   ) {}
 
-  public fetch(noWait?: boolean, size?: number, expires?: number): QueuedIterator<JsMsg> {
-    return this.jsClient.fetch(this.streamName, this.consumerName, {
-      batch: size ?? this.options.batchSize,
+  public async fetch(size?: number, expires?: number): Promise<ConsumerMessages> {
+    const consumer = await this.jsClient.consumers.get(this.streamName, this.consumerName);
+    return await consumer.fetch({
+      max_messages: size ?? this.options.batchSize,
       expires: expires ?? this.options.batchTimeout,
-      no_wait: noWait ?? this.options.noWait,
     });
   }
 }

src/__tests__/Client.ts

@@ -18,7 +18,11 @@ describe('Testing Client class methods', () => {
     request: jest.fn(),
     jetstream: () => ({
       subscribe: jetstreamSubscribeMock,
-      fetch: jetstreamFetchMock,
+      consumers: {
+        get: jest.fn().mockResolvedValue({
+          fetch: jetstreamFetchMock,
+        }),
+      },
     }),
     jetstreamManager: jest.fn().mockResolvedValue({
       streams: {
@@ -97,10 +101,10 @@ describe('Testing Client class methods', () => {
       const payload = { data: { elapsed: 42 } };
       const subscribe = new PassThrough({ objectMode: true });
       const secondSubscribe = new PassThrough({ objectMode: true });
-      subscribe['stop'] = jest.fn();
-      secondSubscribe['stop'] = jest.fn();
-      jetstreamFetchMock.mockReturnValueOnce(subscribe);
-      jetstreamFetchMock.mockReturnValueOnce(secondSubscribe);
+      subscribe['close'] = jest.fn().mockResolvedValue('Ok');
+      secondSubscribe['close'] = jest.fn().mockResolvedValue('Ok');
+      jetstreamFetchMock.mockResolvedValueOnce(subscribe);
+      jetstreamFetchMock.mockResolvedValueOnce(secondSubscribe);
 
       const result = mathClient.getListener('Test', { batch: true });