[FIX] pull adds a status listener on every call, and has no mechanism…

… for removing it - this modifies it so that if the status iterator closes it is removed from the list

jetstream/consumer.ts

@@ -23,6 +23,7 @@ import {
   MsgHdrs,
   NatsError,
   QueuedIterator,
+  Status,
   Subscription,
 } from "../nats-base-client/core.ts";
 import * as core from "../nats-base-client/idleheartbeat.ts";
@@ -228,6 +229,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   timeout: Timeout<unknown> | null;
   cleanupHandler?: () => void;
   listeners: QueuedIterator<ConsumerStatus>[];
+  statusIterator?: QueuedIteratorImpl<Status>;
 
   // callback: ConsumerCallbackFn;
   constructor(
@@ -400,7 +402,9 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     // now if we disconnect, the consumer could be gone
     // or we were slow consumer'ed by the server
     (async () => {
-      for await (const s of c.api.nc.status()) {
+      const status = c.api.nc.status();
+      this.statusIterator = status as QueuedIteratorImpl<Status>;
+      for await (const s of status) {
         switch (s.type) {
           case Events.Disconnect:
             // don't spam hb errors if we are disconnected
@@ -552,6 +556,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   stop(err?: Error) {
     this.sub?.unsubscribe();
     this.clearTimers();
+    this.statusIterator?.stop();
     //@ts-ignore: fn
     this._push(() => {
       super.stop(err);

nats-base-client/nats.ts

@@ -452,6 +452,10 @@ export class NatsConnectionImpl implements NatsConnection {
 
   status(): AsyncIterable<Status> {
     const iter = new QueuedIteratorImpl<Status>();
+    iter.iterClosed.then(() => {
+      const idx = this.listeners.indexOf(iter);
+      this.listeners.splice(idx, 1);
+    });
     this.listeners.push(iter);
     return iter;
   }