Added communicationTimeout options

lib/Client.js

@@ -260,10 +260,16 @@ Client.prototype.setClientInfo = function setClientInfo(key, val) {
   }
 };
 
+Client.prototype.isValid = function isValid(timeout, cb) {
+  // Timeout is inputted in seconds, convert to milliseconds
+  this._connection.isValid(timeout * 1000, cb);
+}
+
 Client.prototype._execute = function _execute(command, options, cb) {
   var result = this._createResult(this._connection, options);
   this._connection.executeDirect({
-    command: command
+    command: command,
+    communicationTimeout: options.communicationTimeout
   }, function handleReply(err, reply) {
     result.handle(err, reply, cb);
   });
@@ -273,7 +279,8 @@ Client.prototype._execute = function _execute(command, options, cb) {
 Client.prototype._prepare = function _prepare(command, options, cb) {
   var statement = this._createStatement(this._connection, options);
   this._connection.prepare({
-    command: command
+    command: command,
+    communicationTimeout: options.communicationTimeout
   }, function handleReply(err, reply) {
     statement.handle(err, reply, cb);
   });

lib/protocol/Connection.js

@@ -250,6 +250,13 @@ Connection.prototype._addListeners = function _addListeners(socket) {
     socket.removeListener('close', onclose);
   }
 
+  function clearStateTimeout() {
+    if (self._state.timeoutObject) {
+      clearTimeout(self._state.timeoutObject);
+      self._state.timeoutObject = undefined;
+    }
+  }
+
   // register listerners on socket
   function ondata(chunk) {
     packet.push(chunk);
@@ -258,17 +265,24 @@ Connection.prototype._addListeners = function _addListeners(socket) {
         self._state.sessionId = packet.header.sessionId;
         self._state.packetCount = -1;
       }
-      var buffer = packet.getData();
-      packet.clear();
-      var cb = self._state.receive;
-      self._state.receive = undefined;
-      self._state.messageType = undefined;
-      self.receive(buffer, cb);
+      // Ensure the packet corresponds to the current request
+      if (self._state.packetCount === -1 || self._state.packetCount === packet.header.packetCount) {
+        var buffer = packet.getData();
+        packet.clear();
+        var cb = self._state.receive;
+        self._state.receive = undefined;
+        self._state.messageType = undefined;
+        clearStateTimeout();
+        self.receive(buffer, cb);
+      } else {
+        packet.clear();
+      }
     }
   }
   socket.on('data', ondata);
 
   function onerror(err) {
+    clearStateTimeout();
     var cb = self._state && self._state.receive;
     if (cb) {
       self._state.receive = null; // a callback should be called only once
@@ -313,7 +327,7 @@ Connection.prototype._clearQueue = function _clearQueue(err) {
   }
 };
 
-Connection.prototype.send = function send(message, receive) {
+Connection.prototype.send = function send(message, options, receive) {
   if (this._statementContext) {
     message.unshift(PartKind.STATEMENT_CONTEXT, this._statementContext.getOptions());
   }
@@ -354,6 +368,21 @@ Connection.prototype.send = function send(message, receive) {
   if (this._socket) {
     this._socket.write(packet);
   }
+  var self = this;
+  var communicationTimeout;
+  function onTimeout() {
+    self._state.receive = undefined;
+    var err = new Error('Socket receive timeout (receive took longer than ' + communicationTimeout + ' ms)');
+    err.code = 'EHDBTIMEOUT';
+    receive(err);
+  }
+  if (options.communicationTimeout) {
+    communicationTimeout = options.communicationTimeout;
+    state.timeoutObject = setTimeout(onTimeout, communicationTimeout);
+  } else if (this._settings.communicationTimeout) {
+    communicationTimeout = this._settings.communicationTimeout;
+    state.timeoutObject = setTimeout(onTimeout, communicationTimeout);
+  }
 };
 
 
@@ -422,7 +451,7 @@ Connection.prototype.receive = function receive(buffer, cb) {
   }
 };
 
-Connection.prototype.enqueue = function enqueue(task, cb) {
+Connection.prototype.enqueue = function enqueue(task, options, cb) {
   var queueable;
 
   if (!this._socket || !this._queue || this.readyState === 'closed') {
@@ -439,7 +468,7 @@ Connection.prototype.enqueue = function enqueue(task, cb) {
     queueable.name = task.name;
   } else if (util.isObject(task)) {
     if (task instanceof request.Segment) {
-      queueable = this._queue.createTask(this.send.bind(this, task), cb);
+      queueable = this._queue.createTask(this.send.bind(this, task, options), cb);
       queueable.name = MessageTypeName[task.type];
     } else if (util.isFunction(task.run)) {
       queueable = task;
@@ -507,6 +536,12 @@ Connection.prototype.connect = function connect(options, cb) {
 
   function connReceive(err, reply) {
     if (err) {
+      if (err.code === 'EHDBTIMEOUT') {
+        // With connect timeouts, the connection should be reset to a closed state
+        self.destroy();
+        // Convert to connect timeout to match hana-client
+        return cb(new Error("Connect failed (connect timeout expired)"));
+      }
       return cb(err);
     }
     if (Array.isArray(reply.connectOptions)) {
@@ -523,6 +558,10 @@ Connection.prototype.connect = function connect(options, cb) {
 
   function authReceive(err, reply) {
     if (err) {
+      if (err.code === 'EHDBTIMEOUT') {
+        self.destroy();
+        return cb(new Error("Connect failed (connect timeout expired)"));
+      }
       return cb(err, reply);
     }
     manager.initialize(reply.authentication, function(err) {
@@ -574,7 +613,7 @@ Connection.prototype.connect = function connect(options, cb) {
         clientId: self.clientId,
         connectOptions: self.connectOptions.getOptions(),
         useCesu8: self.useCesu8
-      }), connReceive);
+      }), options, connReceive);
     });
   }
 
@@ -588,7 +627,7 @@ Connection.prototype.connect = function connect(options, cb) {
     authOptions.dbConnectInfo = true;
   }
 
-  this.send(request.authenticate(authOptions), authReceive);
+  this.send(request.authenticate(authOptions), options, authReceive);
 };
 
 Connection.prototype.disconnect = function disconnect(cb) {
@@ -600,7 +639,7 @@ Connection.prototype.disconnect = function disconnect(cb) {
   }
 
   function enqueueDisconnect() {
-    self.enqueue(request.disconnect(), done);
+    self.enqueue(request.disconnect(), {}, done);
   }
 
   if (this.isIdle()) {
@@ -616,7 +655,7 @@ Connection.prototype.executeDirect = function executeDirect(options, cb) {
     scrollableCursor: this.scrollableCursor,
     useCesu8: this.useCesu8
   }, options);
-  this.enqueue(request.executeDirect(options), cb);
+  this.enqueue(request.executeDirect(options), options, cb);
 };
 
 Connection.prototype.prepare = function prepare(options, cb) {
@@ -625,7 +664,7 @@ Connection.prototype.prepare = function prepare(options, cb) {
     scrollableCursor: this.scrollableCursor,
     useCesu8: this.useCesu8
   }, options);
-  this.enqueue(request.prepare(options), cb);
+  this.enqueue(request.prepare(options), options, cb);
 };
 
 Connection.prototype.readLob = function readLob(options, cb) {
@@ -635,7 +674,7 @@ Connection.prototype.readLob = function readLob(options, cb) {
     };
   }
   options.autoCommit = this.autoCommit;
-  this.enqueue(request.readLob(options), cb);
+  this.enqueue(request.readLob(options), options, cb);
 };
 
 Connection.prototype.execute = function execute(options, cb) {
@@ -646,40 +685,40 @@ Connection.prototype.execute = function execute(options, cb) {
     parameters: EMPTY_BUFFER
   }, options);
   if (options.parameters === EMPTY_BUFFER) {
-    return this.enqueue(request.execute(options), cb);
+    return this.enqueue(request.execute(options), options, cb);
   }
   this.enqueue(createExecuteTask(this, options, cb));
 };
 
 Connection.prototype.fetchNext = function fetchNext(options, cb) {
   options.autoCommit = this.autoCommit;
   options.useCesu8 = this.useCesu8;
-  this.enqueue(request.fetchNext(options), cb);
+  this.enqueue(request.fetchNext(options), options, cb);
 };
 
 Connection.prototype.closeResultSet = function closeResultSet(options, cb) {
-  this.enqueue(request.closeResultSet(options), cb);
+  this.enqueue(request.closeResultSet(options), options, cb);
 };
 
 Connection.prototype.dropStatement = function dropStatement(options, cb) {
   options.useCesu8 = this.useCesu8;
-  this.enqueue(request.dropStatementId(options), cb);
+  this.enqueue(request.dropStatementId(options), options, cb);
 };
 
 Connection.prototype.commit = function commit(options, cb) {
   if (util.isFunction(options)) {
     cb = options;
     options = {};
   }
-  this.enqueue(request.commit(options), cb);
+  this.enqueue(request.commit(options), options, cb);
 };
 
 Connection.prototype.rollback = function rollback(options, cb) {
   if (util.isFunction(options)) {
     cb = options;
     options = {};
   }
-  this.enqueue(request.rollback(options), cb);
+  this.enqueue(request.rollback(options), options, cb);
 };
 
   // The function doesn't use the queue. It's used before the queue starts running
@@ -689,7 +728,7 @@ Connection.prototype.fetchDbConnectInfo = function (options, cb) {
     err.code = 'EHDBCLOSE';
     return cb(err)
   }
-  this.send(request.dbConnectInfo(options), function(err, reply) {
+  this.send(request.dbConnectInfo(options), options, function(err, reply) {
     if (err) {
       return cb(err);
     }
@@ -728,6 +767,21 @@ Connection.prototype.destroy = function destroy(err) {
   }
 };
 
+Connection.prototype.isValid = function isValid(timeout, cb) {
+  var options = {
+    command: 'SELECT 1 FROM DUMMY WHERE 1 = 0',
+    communicationTimeout: timeout > 0 ? timeout : undefined,
+  }
+  this.executeDirect(options, function (err, reply) {
+    if (err) {
+      // Currently, isValid will swallow all errors and indicate invalid
+      cb(null, false);
+    } else {
+      cb(null, true);
+    }
+  });
+}
+
 Connection.prototype.isIdle = function isIdle() {
   return this._queue.empty && !this._queue.busy;
 };
@@ -759,6 +813,7 @@ function ConnectionState() {
   this.packetCount = -1;
   this.messageType = undefined;
   this.receive = undefined;
+  this.timeoutObject = undefined;
 }
 
 function Version(major, minor) {

lib/protocol/ExecuteTask.js

@@ -30,6 +30,7 @@ function ExecuteTask(connection, options, callback) {
   this.scrollableCursor = options.scrollableCursor;
   this.statementId = options.statementId;
   this.functionCode = options.functionCode;
+  this.communicationTimeout = options.communicationTimeout;
   this.writer = new Writer(options.parameters.types, connection.useCesu8, connection.spatialTypes);
   var values = options.parameters.values;
   if (values.length && Array.isArray(values[0])) {
@@ -203,14 +204,16 @@ ExecuteTask.prototype.sendExecute = function sendExecute(cb) {
     if (err) {
       return cb(err);
     }
-    self.connection.send(request.execute({
+    var options = {
       autoCommit: self.autoCommit,
       holdCursorsOverCommit: self.holdCursorsOverCommit,
       scrollableCursor: self.scrollableCursor,
       statementId: self.statementId,
       parameters: parameters,
-      useCesu8: self.connection.useCesu8
-    }), cb);
+      useCesu8: self.connection.useCesu8,
+      communicationTimeout: self.communicationTimeout
+    };
+    self.connection.send(request.execute(options), options, cb);
   });
 };
 
@@ -221,24 +224,28 @@ ExecuteTask.prototype.sendWriteLobRequest = function sendWriteLobRequest(cb) {
     if (err) {
       return cb(err);
     }
-    self.connection.send(request.writeLob({
-      writeLobRequest: buffer
-    }), cb);
+    var options = {
+      writeLobRequest: buffer,
+      communicationTimeout: self.communicationTimeout
+    };
+    self.connection.send(request.writeLob(options), options, cb);
   });
 };
 
 ExecuteTask.prototype.sendCommit = function sendCommit(cb) {
   var self = this;
-  self.connection.send(request.commit({
+  var options = {
     holdCursorsOverCommit: self.holdCursorsOverCommit
-  }), cb);
+  };
+  self.connection.send(request.commit(options), options, cb);
 };
 
 ExecuteTask.prototype.sendRollback = function sendRollback(cb) {
   var self = this;
-  self.connection.send(request.rollback({
+  var options = {
     holdCursorsOverCommit: self.holdCursorsOverCommit
-  }), cb);
+  };
+  self.connection.send(request.rollback(options), options, cb);
 };
 
 function createInvalidFunctionCodeError() {

lib/protocol/Statement.js

@@ -148,7 +148,8 @@ Statement.prototype._execute = function _execute(values, options, cb) {
   this._connection.execute({
     functionCode: this.functionCode,
     statementId: this.id,
-    parameters: inputParams
+    parameters: inputParams,
+    communicationTimeout: options.communicationTimeout
   }, function handleReply(err, reply) {
     result.handle(err, reply, cb);
   });