Start to add transaction support

lib/mysql.js

@@ -133,12 +133,16 @@ MySQL.prototype.executeSQL = function (sql, params, options, callback) {
     debug('SQL: %s, params: %j', sql, params);
   }
 
-  function releaseConnectionAndCallback(connection, err, result) {
+  var transaction = options.transaction;
+
+  function handleResponse(connection, err, result) {
+    if (!transaction) {
       connection.release();
+    }
     callback && callback(err, result);
   }
 
-  function runQuery(connection) {
+  function runQuery(connection, release) {
     connection.query(sql, params, function(err, data) {
       if (debugEnabled) {
         if (err) {
@@ -146,11 +150,11 @@ MySQL.prototype.executeSQL = function (sql, params, options, callback) {
         }
         debug('Data: ', data);
       }
-      releaseConnectionAndCallback(connection, err, data);
+      handleResponse(connection, err, data);
     });
   }
 
-  client.getConnection(function (err, connection) {
+  function executeWithConnection(err, connection) {
     if (err) {
       return callback && callback(err);
     }
@@ -168,12 +172,12 @@ MySQL.prototype.executeSQL = function (sql, params, options, callback) {
                   runQuery(connection);
                 });
               } else {
-                releaseConnectionAndCallback(connection, err);
+                handleResponse(connection, err);
               }
             });
             return;
           } else {
-            releaseConnectionAndCallback(connection, err);
+            handleResponse(connection, err);
             return;
           }
         }
@@ -183,7 +187,17 @@ MySQL.prototype.executeSQL = function (sql, params, options, callback) {
       // Bypass USE db
       runQuery(connection);
     }
-  });
+  }
+
+  if (transaction && transaction.connection &&
+    transaction.connector === this) {
+    if (debugEnabled) {
+      debug('Execute SQL within a transaction');
+    }
+    executeWithConnection(null, transaction.connection);
+  } else {
+    client.getConnection(executeWithConnection);
+  }
 };
 
 /**
@@ -427,7 +441,6 @@ MySQL.prototype.getCountForAffectedRows = function(model, info) {
   return affectedRows;
 };
 
-
 /**
  * Disconnect from MySQL
  */
@@ -448,4 +461,5 @@ MySQL.prototype.ping = function(cb) {
 
 require('./migration')(MySQL, mysql);
 require('./discovery')(MySQL, mysql);
+require('./transaction')(MySQL, mysql);
 

lib/transaction.js

@@ -0,0 +1,63 @@
+var debug = require('debug')('loopback:connector:mysql:transaction');
+module.exports = mixinTransaction;
+
+/*!
+ * @param {MySQL} MySQL connector class
+ * @param {Object} mysql mysql driver
+ */
+function mixinTransaction(MySQL, mysql) {
+
+  /**
+   * Begin a new transaction
+   * @param isolationLevel
+   * @param cb
+   */
+  MySQL.prototype.beginTransaction = function(isolationLevel, cb) {
+    debug('Begin a transaction with isolation level: %s', isolationLevel);
+    this.client.getConnection(function(err, connection) {
+      if(err) return cb(err);
+      if(isolationLevel) {
+        connection.query(
+          'SET SESSION TRANSACTION ISOLATION LEVEL ' + isolationLevel,
+          function(err) {
+            if (err) return cb(err);
+            connection.beginTransaction(function(err) {
+              if (err) return cb(err);
+              return cb(null, connection);
+            });
+          });
+      } else {
+        connection.beginTransaction(function(err) {
+          if (err) return cb(err);
+          return cb(null, connection);
+        });
+      }
+    });
+  };
+
+  /**
+   *
+   * @param connection
+   * @param cb
+   */
+  MySQL.prototype.commit = function(connection, cb) {
+    debug('Commit a transaction');
+    connection.commit(function(err) {
+      connection.release();
+      cb(err);
+    });
+  };
+
+  /**
+   *
+   * @param connection
+   * @param cb
+   */
+  MySQL.prototype.rollback = function(connection, cb) {
+    debug('Rollback a transaction');
+    connection.rollback(function(err) {
+      connection.release();
+      cb(err);
+    });
+  };
+}

test/transaction.promise.test.js

@@ -0,0 +1,180 @@
+if (typeof Promise === 'undefined') {
+  global.Promise = require('bluebird');
+}
+var Transaction = require('loopback-datasource-juggler').Transaction;
+require('./init.js');
+require('should');
+
+var db, Post, Review;
+
+describe('transactions', function() {
+
+  before(function(done) {
+    db = getDataSource({collation: 'utf8_general_ci', createDatabase: true});
+    db.once('connected', function() {
+      Post = db.define('PostTX', {
+        title: {type: String, length: 255, index: true},
+        content: {type: String}
+      }, {mysql: {engine: 'INNODB'}});
+      Review = db.define('ReviewTX', {
+        author: String,
+        content: {type: String}
+      }, {mysql: {engine: 'INNODB'}});
+      Post.hasMany(Review, {as: 'reviews', foreignKey: 'postId'});
+      db.automigrate(['PostTX', 'ReviewTX'], done);
+    });
+  });
+
+  var currentTx;
+  var hooks = [];
+  // Return an async function to start a transaction and create a post
+  function createPostInTx(post, timeout) {
+    return function(done) {
+      // Transaction.begin(db.connector, Transaction.READ_COMMITTED,
+      var promise = Post.beginTransaction({
+        isolationLevel: Transaction.READ_COMMITTED,
+        timeout: timeout
+      });
+      promise.then(function(tx) {
+        (typeof tx.id).should.be.eql('string');
+        currentTx = tx;
+        hooks = [];
+        tx.observe('before commit', function(context, next) {
+          hooks.push('before commit');
+          next();
+        });
+        tx.observe('after commit', function(context, next) {
+          hooks.push('after commit');
+          next();
+        });
+        tx.observe('before rollback', function(context, next) {
+          hooks.push('before rollback');
+          next();
+        });
+        tx.observe('after rollback', function(context, next) {
+          hooks.push('after rollback');
+          next();
+        });
+      }).then(function() {
+        Post.create(post, {transaction: currentTx}).then(
+          function(p) {
+            p.reviews.create({
+              author: 'John',
+              content: 'Review for ' + p.title
+            }, {transaction: currentTx}).then(
+              function(c) {
+                done(null, c);
+              });
+          });
+      }).catch(done);
+    };
+  }
+
+// Return an async function to find matching posts and assert number of
+// records to equal to the count
+  function expectToFindPosts(where, count, inTx) {
+    return function(done) {
+      var options = {};
+      if (inTx) {
+        options.transaction = currentTx;
+      }
+      Post.find({where: where}, options).then(
+        function(posts) {
+          posts.length.should.be.eql(count);
+          if (count) {
+            // Find related reviews
+            // Please note the empty {} is required, otherwise, the options
+            // will be treated as a filter
+            posts[0].reviews({}, options).then(function(reviews) {
+              reviews.length.should.be.eql(count);
+              done();
+            });
+          } else {
+            done();
+          }
+        }).catch(done);
+    };
+  }
+
+  describe('commit', function() {
+
+    var post = {title: 't1', content: 'c1'};
+    before(createPostInTx(post));
+
+    it('should not see the uncommitted insert', expectToFindPosts(post, 0));
+
+    it('should see the uncommitted insert from the same transaction',
+      expectToFindPosts(post, 1, true));
+
+    it('should commit a transaction', function(done) {
+      currentTx.commit().then(function() {
+        hooks.should.be.eql(['before commit', 'after commit']);
+        done();
+      }).catch(done);
+    });
+
+    it('should see the committed insert', expectToFindPosts(post, 1));
+
+    it('should report error if the transaction is not active', function(done) {
+      currentTx.commit().catch(function(err) {
+        (err).should.be.instanceof(Error);
+        done();
+      });
+    });
+  });
+
+  describe('rollback', function() {
+
+    var post = {title: 't2', content: 'c2'};
+    before(createPostInTx(post));
+
+    it('should not see the uncommitted insert', expectToFindPosts(post, 0));
+
+    it('should see the uncommitted insert from the same transaction',
+      expectToFindPosts(post, 1, true));
+
+    it('should rollback a transaction', function(done) {
+      currentTx.rollback().then(function() {
+        hooks.should.be.eql(['before rollback', 'after rollback']);
+        done();
+      }).catch(done);
+    });
+
+    it('should not see the rolledback insert', expectToFindPosts(post, 0));
+
+    it('should report error if the transaction is not active', function(done) {
+      currentTx.rollback().catch(function(err) {
+        (err).should.be.instanceof(Error);
+        done();
+      });
+    });
+  });
+
+  describe('timeout', function() {
+
+    var post = {title: 't3', content: 'c3'};
+    before(createPostInTx(post, 500));
+
+    it('should report timeout', function(done) {
+      setTimeout(function() {
+        Post.find({where: {title: 't3'}}, {transaction: currentTx},
+          function(err, posts) {
+            if (err) return done(err);
+            posts.length.should.be.eql(1);
+            done();
+          });
+      }, 1000);
+      done();
+    });
+
+    it('should invoke the timeout hook', function(done) {
+      currentTx.observe('timeout', function(context, next) {
+        next();
+        done();
+      });
+    });
+
+  });
+})
+;
+

test/transaction.test.js

@@ -0,0 +1,182 @@
+var Transaction = require('loopback-datasource-juggler').Transaction;
+require('./init.js');
+require('should');
+
+var db, Post, Review;
+
+describe('transactions', function() {
+
+  before(function(done) {
+    db = getDataSource({collation: 'utf8_general_ci', createDatabase: true});
+    db.once('connected', function() {
+      Post = db.define('PostTX', {
+        title: {type: String, length: 255, index: true},
+        content: {type: String}
+      }, {mysql: {engine: 'INNODB'}});
+      Review = db.define('ReviewTX', {
+        author: String,
+        content: {type: String}
+      }, {mysql: {engine: 'INNODB'}});
+      Post.hasMany(Review, {as: 'reviews', foreignKey: 'postId'});
+      db.automigrate(['PostTX', 'ReviewTX'], done);
+    });
+  });
+
+  var currentTx;
+  var hooks = [];
+  // Return an async function to start a transaction and create a post
+  function createPostInTx(post, timeout) {
+    return function(done) {
+      // Transaction.begin(db.connector, Transaction.READ_COMMITTED,
+      Post.beginTransaction({
+          isolationLevel: Transaction.READ_COMMITTED,
+          timeout: timeout
+        },
+        function(err, tx) {
+          if (err) return done(err);
+          (typeof tx.id).should.be.eql('string');
+          hooks = [];
+          tx.observe('before commit', function(context, next) {
+            hooks.push('before commit');
+            next();
+          });
+          tx.observe('after commit', function(context, next) {
+            hooks.push('after commit');
+            next();
+          });
+          tx.observe('before rollback', function(context, next) {
+            hooks.push('before rollback');
+            next();
+          });
+          tx.observe('after rollback', function(context, next) {
+            hooks.push('after rollback');
+            next();
+          });
+          currentTx = tx;
+          Post.create(post, {transaction: tx},
+            function(err, p) {
+              if (err) {
+                done(err);
+              } else {
+                p.reviews.create({
+                    author: 'John',
+                    content: 'Review for ' + p.title
+                  }, {transaction: tx},
+                  function(err, c) {
+                    done(err);
+                  });
+              }
+            });
+        });
+    };
+  }
+
+  // Return an async function to find matching posts and assert number of
+  // records to equal to the count
+  function expectToFindPosts(where, count, inTx) {
+    return function(done) {
+      var options = {};
+      if (inTx) {
+        options.transaction = currentTx;
+      }
+      Post.find({where: where}, options,
+        function(err, posts) {
+          if (err) return done(err);
+          posts.length.should.be.eql(count);
+          if (count) {
+            // Find related reviews
+            // Please note the empty {} is required, otherwise, the options
+            // will be treated as a filter
+            posts[0].reviews({}, options, function(err, reviews) {
+              if (err) return done(err);
+              reviews.length.should.be.eql(count);
+              done();
+            });
+          } else {
+            done();
+          }
+        });
+    };
+  }
+
+  describe('commit', function() {
+
+    var post = {title: 't1', content: 'c1'};
+    before(createPostInTx(post));
+
+    it('should not see the uncommitted insert', expectToFindPosts(post, 0));
+
+    it('should see the uncommitted insert from the same transaction',
+      expectToFindPosts(post, 1, true));
+
+    it('should commit a transaction', function(done) {
+      currentTx.commit(function(err) {
+        hooks.should.be.eql(['before commit', 'after commit']);
+        done(err);
+      });
+    });
+
+    it('should see the committed insert', expectToFindPosts(post, 1));
+
+    it('should report error if the transaction is not active', function(done) {
+      currentTx.commit(function(err) {
+        (err).should.be.instanceof(Error);
+        done();
+      });
+    });
+  });
+
+  describe('rollback', function() {
+
+    var post = {title: 't2', content: 'c2'};
+    before(createPostInTx(post));
+
+    it('should not see the uncommitted insert', expectToFindPosts(post, 0));
+
+    it('should see the uncommitted insert from the same transaction',
+      expectToFindPosts(post, 1, true));
+
+    it('should rollback a transaction', function(done) {
+      currentTx.rollback(function(err) {
+        hooks.should.be.eql(['before rollback', 'after rollback']);
+        done(err);
+      });
+    });
+
+    it('should not see the rolledback insert', expectToFindPosts(post, 0));
+
+    it('should report error if the transaction is not active', function(done) {
+      currentTx.rollback(function(err) {
+        (err).should.be.instanceof(Error);
+        done();
+      });
+    });
+  });
+
+  describe('timeout', function() {
+
+    var post = {title: 't3', content: 'c3'};
+    before(createPostInTx(post, 500));
+
+    it('should report timeout', function(done) {
+      setTimeout(function() {
+        Post.find({where: {title: 't3'}}, {transaction: currentTx},
+          function(err, posts) {
+            if (err) return done(err);
+            posts.length.should.be.eql(1);
+            done();
+          });
+      }, 1000);
+      done();
+    });
+
+    it('should invoke the timeout hook', function(done) {
+      currentTx.observe('timeout', function(context, next) {
+        next();
+        done();
+      });
+    });
+
+  });
+});
+