enhancements to downstream iterator use to determine if the iterator failed

Makefile

@@ -1,6 +1,6 @@
 .PHONY: build test bundle lint
 
-DENO_JOBS=4
+export DENO_JOBS=4
 
 build: test
 

jetstream/consumer.ts

@@ -246,7 +246,8 @@ export interface Consumer extends ExportedConsumer {
 }
 
 export interface Close {
-  close(): Promise<void>;
+  close(): Promise<void | Error>;
+  closed(): Promise<void | Error>;
 }
 
 export interface ConsumerMessages extends QueuedIterator<JsMsg>, Close {
@@ -620,12 +621,12 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     this.timeout = t;
   }
 
-  close(): Promise<void> {
+  close(): Promise<void | Error> {
     this.stop();
     return this.iterClosed;
   }
 
-  closed(): Promise<void> {
+  closed(): Promise<void | Error> {
     return this.iterClosed;
   }
 
@@ -641,6 +642,9 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   }
 
   stop(err?: Error) {
+    if (this.done) {
+      return;
+    }
     this.sub?.unsubscribe();
     this.clearTimers();
     this.statusIterator?.stop();
@@ -757,11 +761,15 @@ export class OrderedConsumerMessages extends QueuedIteratorImpl<JsMsg>
     });
   }
 
-  close(): Promise<void> {
+  close(): Promise<void | Error> {
     this.stop();
     return this.iterClosed;
   }
 
+  closed(): Promise<void | Error> {
+    return this.iterClosed;
+  }
+
   status(): Promise<AsyncIterable<ConsumerStatus>> {
     const iter = new QueuedIteratorImpl<ConsumerStatus>();
     this.listeners.push(iter);

jetstream/tests/consumers_consume_test.ts

@@ -110,7 +110,6 @@ Deno.test("consumers - consume deleted consumer", async () => {
   (async () => {
     const status = await iter.status();
     for await (const s of status) {
-      console.log(s);
       if (s.type === ConsumerEvents.ConsumerDeleted) {
         deleted.resolve();
       }

nats-base-client/queued_iterator.ts

@@ -53,7 +53,7 @@ export class QueuedIteratorImpl<T> implements QueuedIterator<T> {
   // FIXME: this is updated by the protocol
   received: number;
   noIterator: boolean;
-  iterClosed: Deferred<void>;
+  iterClosed: Deferred<void | Error>;
   done: boolean;
   signal: Deferred<void>;
   yields: T[];
@@ -78,7 +78,7 @@ export class QueuedIteratorImpl<T> implements QueuedIterator<T> {
     this.done = false;
     this.signal = deferred<void>();
     this.yields = [];
-    this.iterClosed = deferred<void>();
+    this.iterClosed = deferred<void | Error>();
     this.time = 0;
     this.yielding = false;
   }
@@ -185,7 +185,7 @@ export class QueuedIteratorImpl<T> implements QueuedIterator<T> {
     this.err = err;
     this.done = true;
     this.signal.resolve();
-    this.iterClosed.resolve();
+    this.iterClosed.resolve(err);
   }
 
   getProcessed(): number {