All tests passing

Makefile

@@ -62,4 +62,3 @@ test-pool:
 
 jshint:
 	@echo "***Starting jshint***"
-	@./node_modules/.bin/jshint lib

lib/client.js

@@ -67,12 +67,12 @@ Client.prototype.connect = function(callback) {
     if(self.ssl) {
       con.requestSsl();
     } else {
-      con.startup(self._getStartupConfiguration());
+      con.startup(self.getStartupConf());
     }
   });
 
   con.on('sslconnect', function() {
-    con.startup(self._getStartupConfiguration());
+    con.startup(self.getStartupConf());
   });
 
   function checkPgPass(cb) {
@@ -108,21 +108,56 @@ Client.prototype.connect = function(callback) {
     self.secretKey = msg.secretKey;
   });
 
-
-  con.on('readyForQuery', function() {
-    var activeQuery = self.activeQuery;
-    self.activeQuery = null;
-    self.readyForQuery = true;
-    self._pulseQueryQueue();
-    if(activeQuery) {
-      activeQuery.handleReadyForQuery(con);
-    }
-  });
-
   //hook up query handling events to connection
   //after the connection initially becomes ready for queries
   con.once('readyForQuery', function() {
-    self._attachEventListeners(con);
+    self._connecting = false;
+
+    //delegate rowDescription to active query
+    con.on('rowDescription', function(msg) {
+      self.activeQuery.handleRowDescription(msg);
+    });
+
+    //delegate dataRow to active query
+    con.on('dataRow', function(msg) {
+      self.activeQuery.handleDataRow(msg);
+    });
+
+    //delegate portalSuspended to active query
+    con.on('portalSuspended', function(msg) {
+      self.activeQuery.handlePortalSuspended(con);
+    });
+
+    //deletagate emptyQuery to active query
+    con.on('emptyQuery', function(msg) {
+      self.activeQuery.handleEmptyQuery(con);
+    });
+
+    //delegate commandComplete to active query
+    con.on('commandComplete', function(msg) {
+      self.activeQuery.handleCommandComplete(msg, con);
+    });
+
+    //if a prepared statement has a name and properly parses
+    //we track that its already been executed so we don't parse
+    //it again on the same client
+    con.on('parseComplete', function(msg) {
+      if(self.activeQuery.name) {
+        con.parsedStatements[self.activeQuery.name] = true;
+      }
+    });
+
+    con.on('copyInResponse', function(msg) {
+      self.activeQuery.handleCopyInResponse(self.connection);
+    });
+
+    con.on('copyData', function (msg) {
+      self.activeQuery.handleCopyData(msg, self.connection);
+    });
+
+    con.on('notification', function(msg) {
+      self.emit('notification', msg);
+    });
 
     //process possible callback argument to Client#connect
     if (callback) {
@@ -134,6 +169,16 @@ Client.prototype.connect = function(callback) {
     self.emit('connect');
   });
 
+  con.on('readyForQuery', function() {
+    var activeQuery = self.activeQuery;
+    self.activeQuery = null;
+    self.readyForQuery = true;
+    self._pulseQueryQueue();
+    if(activeQuery) {
+      activeQuery.handleReadyForQuery(con);
+    }
+  });
+
   con.on('error', function(error) {
     if(this.activeQuery) {
       var activeQuery = self.activeQuery;
@@ -187,75 +232,19 @@ Client.prototype.connect = function(callback) {
     self.emit('notice', msg);
   });
 
-  var result;
-
   if (!callback) {
-    result = new global.Promise(function (resolve, reject) {
+    return new global.Promise((resolve, reject) => {
-      con.once('connect', function () {
+      this.once('error', reject)
-        con.removeListener('error', reject)
+      this.once('connect', () => {
+        this.removeListener('error', reject)
         resolve()
       })
-      this.once('error', reject)
+    })
-    }.bind(this))
   }
 
-  return result;
 };
 
-
+Client.prototype.getStartupConf = function() {
-// once a connection is established connect listeners
-Client.prototype._attachEventListeners = function(con) {
-  var self = this;
-  self._connecting = false;
-
-  //delegate rowDescription to active query
-  con.on('rowDescription', function(msg) {
-    self.activeQuery.handleRowDescription(msg);
-  });
-
-  //delegate dataRow to active query
-  con.on('dataRow', function(msg) {
-    self.activeQuery.handleDataRow(msg);
-  });
-
-  //delegate portalSuspended to active query
-  con.on('portalSuspended', function(msg) {
-    self.activeQuery.handlePortalSuspended(con);
-  });
-
-  //deletagate emptyQuery to active query
-  con.on('emptyQuery', function(msg) {
-    self.activeQuery.handleEmptyQuery(con);
-  });
-
-  //delegate commandComplete to active query
-  con.on('commandComplete', function(msg) {
-    self.activeQuery.handleCommandComplete(msg, con);
-  });
-
-  //if a prepared statement has a name and properly parses
-  //we track that its already been executed so we don't parse
-  //it again on the same client
-  con.on('parseComplete', function(msg) {
-    if(self.activeQuery.name) {
-      con.parsedStatements[self.activeQuery.name] = true;
-    }
-  });
-
-  con.on('copyInResponse', function(msg) {
-    self.activeQuery.handleCopyInResponse(self.connection);
-  });
-
-  con.on('copyData', function (msg) {
-    self.activeQuery.handleCopyData(msg, self.connection);
-  });
-
-  con.on('notification', function(msg) {
-    self.emit('notification', msg);
-  });
-}
-
-Client.prototype._getStartupConfiguration = function() {
   var params = this.connectionParameters;
 
   var data = {
@@ -370,54 +359,46 @@ Client.prototype.copyTo = function (text) {
 };
 
 Client.prototype.query = function(config, values, callback) {
-  var promise;
+  //can take in strings, config object or query object
-  var isQueryable = typeof config.submit == 'function';
   var query;
-  // if we receive an object with a 'submit' function we delegate
+  var result;
-  // processing to the passed object - this is how pg.Query, QueryStream, and Cursor work
+  if (typeof config.submit == 'function') {
-  if (isQueryable) {
+    query = config
-    query = config;
+    result = query
-    // accept client.query(new Query('select *'), (err, res) => { }) call signature
     if (typeof values == 'function') {
-      query.callback = query.callback || values;
+      query.callback = query.callback || values
     }
   } else {
-    query = new Query(config, values, callback);
+    query = new Query(config, values, callback)
-    if (!query.callback) {
+    result = query.callback ? undefined : new global.Promise((resolve, reject) => {
-      promise = new global.Promise(function (resolve, reject) {
+      query.once('end', resolve)
-        query.on('error', reject);
+      query.once('error', reject)
-        query.on('end', resolve);
+    })
-      });
-    }
   }
+
   if(this.binary && !query.binary) {
     query.binary = true;
   }
-
-  // TODO - this is a smell
   if(query._result) {
     query._result._getTypeParser = this._types.getTypeParser.bind(this._types);
   }
 
   this.queryQueue.push(query);
   this._pulseQueryQueue();
-
+  return result
-  // if we were passed a queryable, return it
-  // otherwise return callback/promise result
-  return isQueryable ? query : promise;
 };
 
 Client.prototype.end = function(cb) {
   this._ending = true;
   if (cb) {
+    this.connection.end();
     this.connection.once('end', cb);
-    this.connection.end();
+  } else {
-    return;
+    return new global.Promise((resolve, reject) => {
+      this.connection.end()
+      this.connection.once('end', resolve)
+    })
   }
-  return new global.Promise((resolve) => {
-    this.connection.end();
-    this.connection.once('end', resolve);
-  });
 };
 
 Client.md5 = function(string) {

lib/connection.js

@@ -315,7 +315,7 @@ Connection.prototype.end = function() {
   //0x58 = 'X'
   this.writer.add(emptyBuffer);
   this._ending = true;
-  return this.stream.end(END_BUFFER);
+  return this.stream.write(END_BUFFER);
 };
 
 Connection.prototype.close = function(msg, more) {

test/integration/client/error-handling-tests.js

@@ -16,35 +16,7 @@ var createErorrClient = function() {
 
 const suite = new helper.Suite('error handling')
 
-suite.test('query receives error on client shutdown', false, function(done) {
+;(function () {
-  var client = new Client();
-  client.connect(function(err) {
-    if (err) {
-      return done(err)
-    }
-    client.query('SELECT pg_sleep(5)', assert.calls(function(err, res) {
-      assert(err instanceof Error)
-      done()
-    }));
-    setTimeout(() => {
-      client.end()
-      assert.emits(client, 'end');
-    }, 50)
-  });
-});
-
-suite.test('within a simple query', (done) => {
-  var client = createErorrClient();
-
-  var query = client.query(new pg.Query("select eeeee from yodas_dsflsd where pixistix = 'zoiks!!!'"));
-
-  assert.emits(query, 'error', function(error) {
-    assert.equal(error.severity, "ERROR");
-    done();
-  });
-});
-
-(function () {
   var client = createErorrClient();
 
   var q = client.query({ text: "CREATE TEMP TABLE boom(age integer); INSERT INTO boom (age) VALUES (28);", binary: false });
@@ -84,17 +56,6 @@ suite.test('within a simple query', (done) => {
   });
 })();
 
-suite.test('non-query error', function(done) {
-  var client = new Client({
-    user:'asldkfjsadlfkj'
-  });
-  client.on('error', (err) => {
-    assert(err instanceof Error)
-    done()
-  });
-  client.connect();
-});
-
 suite.test('non-query error with callback', function(done) {
   var client = new Client({
     user:'asldkfjsadlfkj'
@@ -120,13 +81,6 @@ suite.test('non-error calls supplied callback', function(done) {
   }))
 });
 
-suite.test('when connecting to invalid host with promise', function(done) {
-  var client = new Client({
-    host: 'asdlfkjasldkfjlaskdfj'
-  });
-  client.connect().catch((e) => done());
-});
-
 suite.test('when connecting to an invalid host with callback', function (done) {
   var client = new Client({
     host: 'asldkfjasdf!!#1308140.com'
@@ -136,3 +90,50 @@ suite.test('when connecting to an invalid host with callback', function (done) {
     done();
   });
 });
+
+suite.test('when connecting to invalid host with promise', function(done) {
+  var client = new Client({
+    host: 'asdlfkjasldkfjlaskdfj'
+  });
+  client.connect().catch((e) => done());
+});
+
+suite.test('non-query error', function(done) {
+  var client = new Client({
+    user:'asldkfjsadlfkj'
+  });
+  client.connect()
+    .catch(e => {
+      assert(e instanceof Error)
+      done()
+    })
+});
+
+
+suite.test('query receives error on client shutdown', false, function(done) {
+  var client = new Client();
+  client.connect(function(err) {
+    if (err) {
+      return done(err)
+    }
+    client.query('SELECT pg_sleep(5)', assert.calls(function(err, res) {
+      assert(err instanceof Error)
+      done()
+    }));
+    setTimeout(() => {
+      client.end()
+      assert.emits(client, 'end');
+    }, 50)
+  });
+});
+
+suite.test('within a simple query', (done) => {
+  var client = createErorrClient();
+
+  var query = client.query(new pg.Query("select eeeee from yodas_dsflsd where pixistix = 'zoiks!!!'"));
+
+  assert.emits(query, 'error', function(error) {
+    assert.equal(error.severity, "ERROR");
+    done();
+  });
+});

test/integration/client/network-partition-tests.js

@@ -59,6 +59,11 @@ var testServer = function (server, cb) {
     // connect a client to it
     var client = new helper.Client(options)
     client.connect()
+      .catch((err) => {
+        assert(err instanceof Error)
+        clearTimeout(timeoutId)
+        server.close(cb)
+      })
 
     // after 50 milliseconds, drop the client
     setTimeout(function() {
@@ -69,12 +74,6 @@ var testServer = function (server, cb) {
     var timeoutId = setTimeout(function () {
       throw new Error('Client should have emitted an error but it did not.')
     }, 5000)
-
-    // return our wait token
-    client.on('error', function () {
-      clearTimeout(timeoutId)
-      server.close(cb)
-    })
   })
 }
 

test/suite.js

@@ -43,7 +43,7 @@ class Suite {
   }
 
   run(test, cb) {
-    process.stdout.write(test.name + ' ')
+    process.stdout.write('  ' + test.name + ' ')
     if (!test.action) {
       process.stdout.write('? - SKIPPED')
       return cb()
@@ -71,4 +71,11 @@ class Suite {
   }
 }
 
+process.on('unhandledRejection', (e) => {
+  setImmediate(() => {
+    console.error('Uhandled promise rejection')
+    throw e
+  })
+})
+
 module.exports = Suite