feat(query): add timeout option

Author: yuxizhe

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,15 @@ class Query extends Command {
     }
     this._connection = connection;
     this.options = Object.assign({}, connection.config, this._queryOptions);
+
+    if (this.timeout) {
+      const timeoutHandler = this._handleTimeoutError.bind(this);
+      this.queryTimeout = Timers.setTimeout(
+        timeoutHandler,
+        this.timeout
+      );
+    }
+
     const cmdPacket = new Packets.Query(
       this.sql,
       connection.config.charsetNumber
@@ -58,6 +70,10 @@ class Query extends Command {
 
   done() {
     this._unpipeStream();
+    if (this.queryTimeout) {
+      Timers.clearTimeout(this.queryTimeout);
+      this.queryTimeout = null;
+    }
     if (this.onResult) {
       let rows, fields;
       if (this._resultIndex === 0) {
@@ -272,6 +288,24 @@ class Query extends Command {
     });
     return stream;
   }
+
+  _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;