Debug/transaction test

lib/postgresql.js

@@ -139,6 +139,7 @@ PostgreSQL.prototype.connect = function(callback) {
   var self = this;
   self.pg.connect(function(err, client, done) {
     self.client = client;
+    self.finish = done;
     process.nextTick(done);
     callback && callback(err, client);
   });
@@ -173,6 +174,10 @@ PostgreSQL.prototype.executeSQL = function(sql, params, options, callback) {
       }
       if (self.settings.debug && data) self.debug('%j', data);
       if (done) {
+        if (err) {
+          self.client = undefined;
+        }
+        // done(err);
         process.nextTick(function() {
           // Release the connection in next tick
           done(err);
@@ -194,16 +199,30 @@ PostgreSQL.prototype.executeSQL = function(sql, params, options, callback) {
   }
 
   var transaction = options.transaction;
-  if (transaction && transaction.connection &&
-    transaction.connector === this) {
+  if (transaction && transaction.connector === this) {
+    if (!transaction.connection) {
+      return process.nextTick(function() {
+        callback(new Error(g.f('Connection does not exist')));
+      });
+    }
+    if (transaction.txId !== transaction.connection.txId) {
+      return process.nextTick(function() {
+        callback(new Error(g.f('Transaction is not active')));
+      });
+    }
     debug('Execute SQL within a transaction');
     // Do not release the connection
     executeWithConnection(transaction.connection, null);
   } else {
-    self.pg.connect(function(err, connection, done) {
-      if (err) return callback(err);
-      executeWithConnection(connection, done);
-    });
+    if (self.client) {
+      executeWithConnection(self.client, self.finish);
+    } else {
+      self.pg.connect(function(newConErr, client, newConDone) {
+        self.client = client;
+        self.finish = newConDone;
+        executeWithConnection(self.client, self.finish);
+      });
+    }
   }
 };
 

lib/transaction.js

@@ -5,6 +5,8 @@
 
 'use strict';
 var debug = require('debug')('loopback:connector:postgresql:transaction');
+var uuid = require('uuid');
+var Transaction = require('loopback-connector').Transaction;
 
 module.exports = mixinTransaction;
 
@@ -18,14 +20,18 @@ function mixinTransaction(PostgreSQL) {
    * @param cb
    */
   PostgreSQL.prototype.beginTransaction = function(isolationLevel, cb) {
+    var connector = this;
     debug('Begin a transaction with isolation level: %s', isolationLevel);
     this.pg.connect(function(err, connection, done) {
       if (err) return cb(err);
       connection.autorelease = done;
       connection.query('BEGIN TRANSACTION ISOLATION LEVEL ' + isolationLevel,
         function(err) {
           if (err) return cb(err);
-          cb(null, connection);
+          var tx = new Transaction(connector, connection);
+          tx.txId = uuid.v1();
+          connection.txId = tx.txId;
+          cb(null, tx);
         });
     });
   };
@@ -65,6 +71,7 @@ function mixinTransaction(PostgreSQL) {
 
   PostgreSQL.prototype.releaseConnection = function(connection, err) {
     if (typeof connection.autorelease === 'function') {
+      connection.txId = null;
       connection.autorelease(err);
       connection.autorelease = null;
     } else {

package.json

@@ -25,7 +25,8 @@
     "debug": "^2.1.1",
     "loopback-connector": "^4.0.0",
     "pg": "^6.0.0",
-    "strong-globalize": "^2.6.2"
+    "strong-globalize": "^2.6.2",
+    "uuid": "^3.0.1"
   },
   "devDependencies": {
     "eslint": "^2.13.1",

test/postgresql.transaction.test.js

@@ -67,15 +67,20 @@ describe('transactions', function() {
         var create = createPostInTx(post);
         Transaction.begin(db.connector, Transaction.SERIALIZABLE,
           function(err, tx) {
-            if (err) return done(err);
+            if (err) {
+              console.log('beginTransaction ' + tx);
+              return done(err);
+            }
             Post.create(post, {transaction: tx},
               function(err, p) {
                 if (err) {
-                  done(err);
+                  console.log('create ' + post + ' tx ' + tx);
+                  return done(err);
                 } else {
                   tx.commit(function(err) {
                     if (err) {
-                      done(err);
+                      console.log('commit tx ' + tx);
+                      return done(err);
                     }
                     completed++;
                     checkResults();
@@ -124,4 +129,29 @@ describe('transactions', function() {
 
     it('should not see the rolledback insert', expectToFindPosts(post, 0));
   });
+
+  describe('finished', function() {
+    var post = {title: 't2', content: 'c2'};
+    beforeEach(createPostInTx(post));
+
+    it('should throw an error when creating in a committed transaction', function(done) {
+      currentTx.commit(function(err) {
+        if (err) return done(err);
+        Post.create({title: 't4', content: 'c4'}, {transaction: currentTx}, function(err, post) {
+          if (!err) return done(new Error('should throw error'));
+          done();
+        });
+      });
+    });
+
+    it('should throw an error when creating in a rolled back transaction', function(done) {
+      currentTx.rollback(function(err) {
+        if (err) return done(err);
+        Post.create({title: 't4', content: 'c4'}, {transaction: currentTx}, function(err, post) {
+          if (!err) return done(new Error('should throw error'));
+          done();
+        });
+      });
+    });
+  });
 });