Address errors thrown while waiting for writes to drain onto the network

Author: sharadraju

lib/thin/sqlnet/ntTcp.js

@@ -240,7 +240,8 @@ class NTTCP {
     }
     this.stream = null;
     this.connected = false;
-    this.waiter = null;
+    this.drainWaiter = null;
+    this.readWaiter = null;
   }
 
   /**
@@ -323,10 +324,7 @@ class NTTCP {
       if (err) {
         this.savedErr = err;
         this.err = true;
-        if (this.waiter) {
-          this.waiter();
-          this.waiter = null;
-        }
+        this._notifyWaiters();
       }
     });
     if (!result) {
@@ -352,7 +350,9 @@ class NTTCP {
   async pauseWrite() {
     this.checkErr();
     if (this.needsDrain) {
-      await new Promise((resolve) => this.stream.once('drain', resolve));
+      await new Promise((resolve) => {
+        this.drainWaiter = resolve;
+      });
       this.checkErr();
     } else {
       await new Promise((resolve) => Timers.setImmediate(resolve));
@@ -397,9 +397,9 @@ class NTTCP {
           flags: tempBuf[5]
         };
         this.packets.push(packet);
-        if (this.waiter) {
-          this.waiter();
-          this.waiter = null;
+        if (this.readWaiter) {
+          this.readWaiter();
+          this.readWaiter = null;
         }
         if (process.env.NODE_ORACLEDB_DEBUG_PACKETS)
           this.printPacket("Receiving packet", packet.buf);
@@ -436,7 +436,7 @@ class NTTCP {
     if (this.packets.length === 0) {
       this.checkErr();
       await new Promise((resolve) => {
-        this.waiter = resolve;
+        this.readWaiter = resolve;
         this.numPacketsSinceLastWait = 0;
       });
       this.checkErr();
@@ -472,26 +472,25 @@ class NTTCP {
     this.stream.on('error', (err) => {
       this.savedErr = err;
       this.err = true;
-      if (this.waiter) {
-        this.waiter();
-        this.waiter = null;
-      }
+      this._notifyWaiters();
     });
 
     this.stream.on('end', () => {
       this.err = true;
-      if (this.waiter) {
-        this.waiter();
-        this.waiter = null;
-      }
+      this._notifyWaiters();
     });
 
     this.stream.on('close', () => {
       this.connected = false;
+      this._notifyWaiters();
     });
 
     this.stream.on('drain', () => {
       this.needsDrain = false;
+      if (this.drainWaiter) {
+        this.drainWaiter();
+        this.drainWaiter = null;
+      }
     });
 
   }
@@ -516,6 +515,20 @@ class NTTCP {
     }
   }
 
+  /**
+   * Notify the waiters (drain and read) and reset them, if applicable.
+   */
+  _notifyWaiters() {
+    if (this.drainWaiter) {
+      this.drainWaiter();
+      this.drainWaiter = null;
+    }
+    if (this.readWaiter) {
+      this.readWaiter();
+      this.readWaiter = null;
+    }
+  }
+
 }
 
 module.exports = NTTCP;