Merge pull request #1364 from yuxizhe/feat-query-timeout

feat(query): add timeout option

Author: sidorares

lib/commands/execute.js

@@ -15,6 +15,8 @@ class Execute extends Command {
     this.onResult = callback;
     this.parameters = options.values;
     this.insertId = 0;
+    this.timeout = options.timeout;
+    this.queryTimeout = null;
     this._rows = [];
     this._fields = [];
     this._result = [];
@@ -35,6 +37,7 @@ class Execute extends Command {
   start(packet, connection) {
     this._connection = connection;
     this.options = Object.assign({}, connection.config, this._executeOptions);
+    this._setTimeout();
     const executePacket = new Packets.Execute(
       this.statement.id,
       this.parameters,
@@ -96,6 +99,8 @@ Execute.prototype.resultsetHeader = Query.prototype.resultsetHeader;
 Execute.prototype._findOrCreateReadStream =
   Query.prototype._findOrCreateReadStream;
 Execute.prototype._streamLocalInfile = Query.prototype._streamLocalInfile;
+Execute.prototype._setTimeout = Query.prototype._setTimeout;
+Execute.prototype._handleTimeoutError = Query.prototype._handleTimeoutError;
 Execute.prototype.row = Query.prototype.row;
 Execute.prototype.stream = Query.prototype.stream;
 

lib/commands/query.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const process = require('process');
+const Timers = require('timers');
 
 const Readable = require('stream').Readable;
 
@@ -21,6 +22,8 @@ class Query extends Command {
     this._queryOptions = options;
     this.namedPlaceholders = options.namedPlaceholders || false;
     this.onResult = callback;
+    this.timeout = options.timeout;
+    this.queryTimeout = null;
     this._fieldCount = 0;
     this._rowParser = null;
     this._fields = [];
@@ -48,6 +51,8 @@ class Query extends Command {
     }
     this._connection = connection;
     this.options = Object.assign({}, connection.config, this._queryOptions);
+    this._setTimeout();
+
     const cmdPacket = new Packets.Query(
       this.sql,
       connection.config.charsetNumber
@@ -58,6 +63,15 @@ class Query extends Command {
 
   done() {
     this._unpipeStream();
+    // if all ready timeout, return null directly
+    if (this.timeout && !this.queryTimeout) {
+      return null;
+    }
+    // else clear timer
+    if (this.queryTimeout) {
+      Timers.clearTimeout(this.queryTimeout);
+      this.queryTimeout = null;
+    }
     if (this.onResult) {
       let rows, fields;
       if (this._resultIndex === 0) {
@@ -272,6 +286,34 @@ class Query extends Command {
     });
     return stream;
   }
+
+  _setTimeout() {
+    if (this.timeout) {
+      const timeoutHandler = this._handleTimeoutError.bind(this);
+      this.queryTimeout = Timers.setTimeout(
+        timeoutHandler,
+        this.timeout
+      );
+    }
+  }
+
+  _handleTimeoutError() {
+    if (this.queryTimeout) {
+      Timers.clearTimeout(this.queryTimeout);
+      this.queryTimeout = null;
+    }
+    
+    const err = new Error('Query inactivity timeout');
+    err.errorno = 'PROTOCOL_SEQUENCE_TIMEOUT';
+    err.code = 'PROTOCOL_SEQUENCE_TIMEOUT';
+    err.syscall = 'query';
+
+    if (this.onResult) {
+      this.onResult(err);
+    } else {
+      this.emit('error', err);
+    }
+  }
 }
 
 Query.prototype.catch = Query.prototype.then;

test/common.js

@@ -121,7 +121,8 @@ exports.createConnection = function(args) {
     dateStrings: args && args.dateStrings,
     authSwitchHandler: args && args.authSwitchHandler,
     typeCast: args && args.typeCast,
-    namedPlaceholders: args && args.namedPlaceholders
+    namedPlaceholders: args && args.namedPlaceholders,
+    connectTimeout: args && args.connectTimeout,
   };
 
   const conn = driver.createConnection(params);

test/integration/connection/test-connect-timeout.js

@@ -0,0 +1,27 @@
+'use strict';
+
+const common = require('../../common');
+const connection = common.createConnection({ 
+  host: '10.255.255.1',
+  debug: false,
+  connectTimeout: 100, 
+});
+
+const assert = require('assert');
+
+connection.query('SELECT sleep(3) as a', (err, res) => {
+  assert.equal(res, null);
+  assert.ok(err);
+  assert.equal(err.code, 'ETIMEDOUT');
+  assert.equal(err.message, 'connect ETIMEDOUT');
+});
+
+connection.query({ sql: 'SELECT sleep(3) as a' , timeout: 50}, (err, res) => {
+  assert.equal(res, null);
+  assert.ok(err);
+  assert.equal(err.code, 'ETIMEDOUT');
+  assert.equal(err.message, 'connect ETIMEDOUT');
+});
+
+
+

test/integration/connection/test-query-timeout.js

@@ -0,0 +1,50 @@
+'use strict';
+
+const common = require('../../common');
+const connection = common.createConnection({ debug: false });
+const assert = require('assert');
+
+connection.query({ sql: 'SELECT sleep(3) as a', timeout: 500 }, (err, res) => {
+  assert.equal(res, null);
+  assert.ok(err);
+  assert.equal(err.code, 'PROTOCOL_SEQUENCE_TIMEOUT');
+  assert.equal(err.message, 'Query inactivity timeout');
+});
+
+connection.query({ sql: 'SELECT sleep(1) as a', timeout: 5000 }, (err, res) => {
+  assert.deepEqual(res, [{ a: 0 }]);
+});
+
+connection.query('SELECT sleep(1) as a', (err, res) => {
+  assert.deepEqual(res, [{ a: 0 }]);
+});
+
+connection.execute({ sql: 'SELECT sleep(3) as a', timeout: 500 }, (err, res) => {
+  assert.equal(res, null);
+  assert.ok(err);
+  assert.equal(err.code, 'PROTOCOL_SEQUENCE_TIMEOUT');
+  assert.equal(err.message, 'Query inactivity timeout');
+});
+
+connection.execute({ sql: 'SELECT sleep(1) as a', timeout: 5000 }, (err, res) => {
+  assert.deepEqual(res, [{ a: 0 }]);
+});
+
+connection.execute('SELECT sleep(1) as a', (err, res) => {
+  assert.deepEqual(res, [{ a: 0 }]);
+  connection.end();
+});
+
+const connectionTimeout = common.createConnection({ 
+  host: '10.255.255.1',
+  debug: false,
+  connectTimeout: 100, 
+});
+
+// return connect timeout error first
+connectionTimeout.query({ sql: 'SELECT sleep(3) as a', timeout: 50 }, (err, res) => {
+  assert.equal(res, null);
+  assert.ok(err);
+  assert.equal(err.code, 'ETIMEDOUT');
+  assert.equal(err.message, 'connect ETIMEDOUT');
+});