Handle notices, don't send reqs that have already been closed

src/Connection.ts

@@ -26,13 +26,18 @@ class SendQueue extends Queue {
     }
 
     if (verb === 'REQ') {
-      return this.cxn.meta.pendingRequests.size < 10
+      return this.cxn.meta.pendingRequests.size < 8
     }
 
     return true
   }
 
   handle(message: SocketMessage) {
+    // If we ended up handling a CLOSE before we handled the REQ, don't send the REQ
+    if (message[0] === 'CLOSE') {
+      this.messages = this.messages.filter(m => !(m[0] === 'REQ' && m[1] === message[1]))
+    }
+
     this.cxn.onSend(message)
   }
 }

src/ConnectionMeta.ts

@@ -44,7 +44,7 @@ export class ConnectionMeta {
   responseCount = 0
   responseTimer = 0
 
-  constructor(cxn: Connection) {
+  constructor(readonly cxn: Connection) {
     cxn.on('open', () => {
       this.lastOpen = Date.now()
     })
@@ -77,6 +77,8 @@ export class ConnectionMeta {
       if (verb === 'EVENT') this.onReceiveEvent(...payload)
       // @ts-ignore
       if (verb === 'EOSE') this.onReceiveEose(...payload)
+      // @ts-ignore
+      if (verb === 'NOTICE') this.onReceiveNotice(...payload)
     })
   }
 
@@ -133,6 +135,10 @@ export class ConnectionMeta {
     }
   }
 
+  onReceiveNotice(message: string) {
+    console.warn('NOTICE', this.cxn.url, message)
+  }
+
   clearPending = () => {
     this.pendingPublishes.clear()
     this.pendingRequests.clear()

src/util/Queue.ts

@@ -20,7 +20,14 @@ export class Queue {
   }
 
   doWork() {
-    for (const message of this.messages.splice(0, 10)) {
+    for (let i = 0; i < 10; i++) {
+      if (this.messages.length === 0) {
+        break
+      }
+
+      // Pop the messages one at a time so handle can modify the queue
+      const [message] = this.messages.splice(0, 1)
+
       if (this.shouldSend(message)) {
         this.handle(message)
       } else {