handle backpressure when loading data from file

Author: kalenikalexander

lib/connection.js

@@ -32,17 +32,14 @@ class Connection extends EventEmitter {
       if (opts.config.socketPath) {
         this.stream = Net.connect(opts.config.socketPath);
       } else {
-        this.stream = Net.connect(
-          opts.config.port,
-          opts.config.host
-        );
+        this.stream = Net.connect(opts.config.port, opts.config.host);
 
         // Enable keep-alive on the socket.  It's disabled by default, but the
         // user can enable it and supply an initial delay.
         this.stream.setKeepAlive(true, this.config.keepAliveInitialDelay);
       }
       // if stream is a function, treat it as "stream agent / factory"
-    } else if (typeof opts.config.stream === 'function')  {
+    } else if (typeof opts.config.stream === 'function') {
       this.stream = opts.config.stream(opts);
     } else {
       this.stream = opts.config.stream;
@@ -223,11 +220,15 @@ class Connection extends EventEmitter {
   }
 
   write(buffer) {
-    this.stream.write(buffer, err => {
+    const result = this.stream.write(buffer, err => {
       if (err) {
         this._handleNetworkError(err);
       }
     });
+
+    if (!result) {
+      this.stream.emit('pause');
+    }
   }
 
   // http://dev.mysql.com/doc/internals/en/sequence-id.html
@@ -260,11 +261,19 @@ class Connection extends EventEmitter {
       if (this.config.debug) {
         // eslint-disable-next-line no-console
         console.log(
-          `${this._internalId} ${this.connectionId} <== ${this._command._commandName}#${this._command.stateName()}(${[this.sequenceId, packet._name, packet.length()].join(',')})`
+          `${this._internalId} ${this.connectionId} <== ${
+            this._command._commandName
+          }#${this._command.stateName()}(${[
+            this.sequenceId,
+            packet._name,
+            packet.length()
+          ].join(',')})`
         );
         // eslint-disable-next-line no-console
         console.log(
-          `${this._internalId} ${this.connectionId} <== ${packet.buffer.toString('hex')}`
+          `${this._internalId} ${
+            this.connectionId
+          } <== ${packet.buffer.toString('hex')}`
         );
       }
       this._bumpSequenceId(1);
@@ -277,7 +286,13 @@ class Connection extends EventEmitter {
         );
         // eslint-disable-next-line no-console
         console.log(
-          `${this._internalId} ${this.connectionId} <== ${this._command._commandName}#${this._command.stateName()}(${[this.sequenceId, packet._name, packet.length()].join(',')})`
+          `${this._internalId} ${this.connectionId} <== ${
+            this._command._commandName
+          }#${this._command.stateName()}(${[
+            this.sequenceId,
+            packet._name,
+            packet.length()
+          ].join(',')})`
         );
       }
       for (offset = 4; offset < 4 + length; offset += MAX_PACKET_LENGTH) {
@@ -402,7 +417,13 @@ class Connection extends EventEmitter {
           : '(no command)';
         // eslint-disable-next-line no-console
         console.log(
-          `${this._internalId} ${this.connectionId} ==> ${commandName}#${stateName}(${[packet.sequenceId, packet.type(), packet.length()].join(',')})`
+          `${this._internalId} ${
+            this.connectionId
+          } ==> ${commandName}#${stateName}(${[
+            packet.sequenceId,
+            packet.type(),
+            packet.length()
+          ].join(',')})`
         );
       }
     }
@@ -497,7 +518,10 @@ class Connection extends EventEmitter {
       cmdQuery = Connection.createQuery(sql, values, cb, this.config);
     }
     this._resolveNamedPlaceholders(cmdQuery);
-    const rawSql = this.format(cmdQuery.sql, cmdQuery.values !== undefined ? cmdQuery.values : []);
+    const rawSql = this.format(
+      cmdQuery.sql,
+      cmdQuery.values !== undefined ? cmdQuery.values : []
+    );
     cmdQuery.sql = rawSql;
     return this.addCommand(cmdQuery);
   }
@@ -839,9 +863,9 @@ class Connection extends EventEmitter {
   }
 
   static statementKey(options) {
-    return (
-      `${typeof options.nestTables}/${options.nestTables}/${options.rowsAsArray}${options.sql}`
-    );
+    return `${typeof options.nestTables}/${options.nestTables}/${
+      options.rowsAsArray
+    }${options.sql}`;
   }
 }