Merge pull request #568 from sidorares/pr-567-tweaks

Make Promise Pool wrapper extend EventEmitter

Author: sidorares

promise.js

@@ -1,4 +1,18 @@
 var core = require('./index.js');
+var EventEmitter = require('events').EventEmitter;
+var util = require('util');
+
+function inheritEvents(source, target, events) {
+  events
+    .forEach(function (eventName) {
+      source.on(eventName, function () {
+        var args = [].slice.call(arguments);
+        args.unshift(eventName);
+
+        target.emit.apply(target, args);
+      });
+    });
+}
 
 function createConnection (opts) {
   var coreConnection = core.createConnection(opts);
@@ -19,7 +33,10 @@ function createConnection (opts) {
 function PromiseConnection (connection, promiseImpl) {
   this.connection = connection;
   this.Promise = promiseImpl;
+
+  inheritEvents(connection, this, ['error', 'drain', 'connect', 'end', 'enqueue']);
 }
+util.inherits(PromiseConnection, EventEmitter);
 
 PromiseConnection.prototype.release = function () {
   this.connection.release();
@@ -165,6 +182,62 @@ PromiseConnection.prototype.prepare = function () {
   'unprepare'
 ]);
 
+function PromisePool(pool, Promise) {
+  this.pool = pool;
+  this.Promise = Promise;
+  
+  inheritEvents(pool, this, ['acquire', 'connection', 'enqueue', 'release']);
+}
+util.inherits(PromisePool, EventEmitter);
+
+PromisePool.prototype.getConnection = function () {
+  var corePool = this.pool;
+
+  return new this.Promise(function (resolve, reject) {
+    corePool.getConnection(function (err, coreConnection) {
+      if (err) {
+        reject(err);
+      } else {
+        resolve(new PromiseConnection(coreConnection, Promise));
+      }
+    });
+  });
+};
+
+PromisePool.prototype.query = function (sql, args) {
+  var corePool = this.pool;
+
+  return new this.Promise(function (resolve, reject) {
+    var done = makeDoneCb(resolve, reject);
+    if (args) {
+      corePool.query(sql, args, done);
+    } else {
+      corePool.query(sql, done);
+    }
+  });
+};
+
+PromisePool.prototype.execute = function (sql, values) {
+  var corePool = this.pool;
+
+  return new Promise(function (resolve, reject) {
+    corePool.execute(sql, values, makeDoneCb(resolve, reject));
+  });
+};
+
+PromisePool.prototype.end = function () {
+  var corePool = this.pool;
+
+  return new Promise(function (resolve, reject) {
+    corePool.end(function (err) {
+      if (err) {
+        reject(err);
+      } else {
+        resolve();
+      }
+    });
+  });
+};
 
 function createPool (opts) {
   var corePool = core.createPool(opts);
@@ -175,50 +248,7 @@ function createPool (opts) {
       ' implementation as parameter, for example: { Promise: require(\'bluebird\') }');
   }
 
-  var promisePool = {
-    getConnection: function () {
-      return new Promise(function (resolve, reject) {
-        corePool.getConnection(function (err, coreConnection) {
-          if (err) {
-            reject(err);
-          } else {
-            resolve(new PromiseConnection(coreConnection, Promise));
-          }
-        });
-      });
-    },
-
-    query: function (sql, args) {
-      return new Promise(function (resolve, reject) {
-        var done = makeDoneCb(resolve, reject);
-        if (args) {
-          corePool.query(sql, args, done);
-        } else {
-          corePool.query(sql, done);
-        }
-      });
-    },
-
-    execute: function (sql, values) {
-      return new Promise(function (resolve, reject) {
-        corePool.execute(sql, values, makeDoneCb(resolve, reject));
-      });
-    },
-
-    end: function () {
-      return new Promise(function (resolve, reject) {
-        corePool.end(function (err) {
-          if (err) {
-            reject(err);
-          } else {
-            resolve();
-          }
-        });
-      });
-    }
-  };
-
-  return promisePool;
+  return new PromisePool(corePool, Promise);
 }
 
 module.exports.createConnection = createConnection;

test/common.js

@@ -56,25 +56,22 @@ module.exports.createConnection = function(args, callback) {
     // c.on('connect', function() {
     //
     // });
-    setTimeout(
-      function() {
-        console.log('altering client...');
-        c.oldQuery = c.query;
-        c.query = function(sql, callback) {
-          var rows = [];
-          var q = c.oldQuery(sql);
-          q.on('result', function(res) {
-            res.on('row', function(row) {
-              rows.push(row);
-            });
-            res.on('end', function() {
-              callback(null, rows);
-            });
+    setTimeout(function() {
+      console.log('altering client...');
+      c.oldQuery = c.query;
+      c.query = function(sql, callback) {
+        var rows = [];
+        var q = c.oldQuery(sql);
+        q.on('result', function(res) {
+          res.on('row', function(row) {
+            rows.push(row);
           });
-        };
-      },
-      1000
-    );
+          res.on('end', function() {
+            callback(null, rows);
+          });
+        });
+      };
+    }, 1000);
     return c;
   }
 

test/integration/test-promise-wrappers.js

@@ -18,9 +18,11 @@ assert.equal(mainExport, createConnection);
 
 var doneCalled = false;
 var exceptionCaught = false;
+var doneEventsConnect = false;
 
 var doneCalledPool = false;
 var exceptionCaughtPool = false;
+var doneEventsPool = false;
 
 function testBasic() {
   var connResolved;
@@ -102,6 +104,38 @@ function testObjParams() {
     });
 }
 
+function testEventsConnect() {
+  var connPromise = createConnection(config).then(function(conn) {
+    var events = 0;
+
+    conn
+      .once('error', function(connection) {
+        ++events;
+      })
+      .once('drain', function(connection) {
+        ++events;
+      })
+      .once('connect', function() {
+        ++events;
+      })
+      .once('enqueue', function() {
+        ++events;
+      })
+      .once('end', function() {
+        ++events;
+
+        doneEventsConnect = events === 5;
+      });
+
+    conn.connection.emit('error');
+    conn.connection.emit('drain');
+    conn.connection.emit('connect');
+    conn.connection.emit('enqueue');
+    conn.connection.emit('end');
+    conn.end();
+  });
+}
+
 function testBasicPool() {
   var pool = createPool(config);
   pool
@@ -163,21 +197,51 @@ function testObjParamsPool() {
     });
 }
 
+function testEventsPool() {
+  var pool = createPool(config);
+  var events = 0;
+
+  pool
+    .once('acquire', function(connection) {
+      ++events;
+    })
+    .once('connection', function(connection) {
+      ++events;
+    })
+    .once('enqueue', function() {
+      ++events;
+    })
+    .once('release', function() {
+      ++events;
+
+      doneEventsPool = events === 4;
+    });
+
+  pool.pool.emit('acquire');
+  pool.pool.emit('connection');
+  pool.pool.emit('enqueue');
+  pool.pool.emit('release');
+}
+
 testBasic();
 testErrors();
 testObjParams();
+testEventsConnect();
 testBasicPool();
 testErrorsPool();
 testObjParamsPool();
+testEventsPool();
 
 process.on('exit', function() {
   if (skipTest) {
     return;
   }
   assert.equal(doneCalled, true);
   assert.equal(exceptionCaught, true);
+  assert.equal(doneEventsConnect, true);
   assert.equal(doneCalledPool, true);
   assert.equal(exceptionCaughtPool, true);
+  assert.equal(doneEventsPool, true);
 });
 
 process.on('unhandledRejection', function(err) {

test/run.js

@@ -12,3 +12,16 @@ if (process.env.FILTER) {
 process.env.TZ = 'UTC';
 
 require('urun')(__dirname, options);
+
+
+process.on('exit', (code) => {
+  console.log(`About to exit with code: ${code}`);
+});
+
+process.on('unhandledRejection', (reason) => {
+  console.log('unhandledRejection', reason);
+});
+
+process.on('uncaughtException', (err) => {
+  console.log('uncaughtException', err);
+});