Merge pull request #166 from stephenplusplus/nexttick

connection: use setImmediate to simulate async.

Author: stephenplusplus

lib/common/connection.js

@@ -161,7 +161,8 @@ Connection.prototype.fetchToken = function(callback) {
     }, function(err, res, body) {
       if (err || !body.access_token) {
         // TODO: Provide better context about the error here.
-        return callback(err);
+        callback(err);
+        return;
       }
       var exp = new Date(body.token_expires * 1000);
       callback(null, new Token(body.access_token, exp));
@@ -172,7 +173,8 @@ Connection.prototype.fetchToken = function(callback) {
     // read key file once and cache the contents.
     fs.readFile(this.opts.keyFilename, function(err, data) {
       if (err) {
-        return callback(err);
+        callback(err);
+        return;
       }
       that.credentials = JSON.parse(data);
       that.fetchServiceAccountToken_(callback);
@@ -197,11 +199,13 @@ Connection.prototype.fetchServiceAccountToken_ = function(callback) {
     scope: this.scopes.join(' ')
   }, function(err) {
     if (err) {
-      return callback(err);
+      callback(err);
+      return;
     }
     gapi.getToken(function(err) {
       if (err) {
-        return callback(err);
+        callback(err);
+        return;
       }
       var exp = new Date(gapi.token_expires * 1000);
       callback(null, new Token(gapi.token, exp));
@@ -260,7 +264,7 @@ Connection.prototype.createAuthorizedReq = function(reqOpts, callback) {
   }
 
   if (this.isConnected()) {
-    onConnected();
+    setImmediate(onConnected);
     return;
   }
 

lib/datastore/dataset.js

@@ -290,7 +290,8 @@ Dataset.prototype.runInTransaction = function(fn, callback) {
   var newTransaction = this.createTransaction_();
   newTransaction.begin(function(err) {
     if (err) {
-      return callback(err);
+      callback(err);
+      return;
     }
     fn(newTransaction, newTransaction.finalize.bind(newTransaction, callback));
   });
@@ -329,7 +330,8 @@ Dataset.prototype.allocateIds = function(incompleteKey, n, callback) {
       new pb.AllocateIdsRequest({ key: incompleteKeys }),
       pb.AllocateIdsResponse, function(err, resp) {
     if (err) {
-      return callback(err);
+      callback(err);
+      return;
     }
     var keys = [];
     (resp.key || []).forEach(function(k) {

lib/datastore/transaction.js

@@ -95,10 +95,12 @@ Transaction.prototype.begin = function(callback) {
   var req = new pb.BeginTransactionRequest();
   var res = pb.BeginTransactionResponse;
   this.makeReq('beginTransaction', req, res, function(err, resp) {
-    if (!err) {
-      that.id = resp.transaction;
+    if (err) {
+      callback(err);
+      return;
     }
-    callback(err);
+    that.id = resp.transaction;
+    callback(null);
   });
 };
 
@@ -122,10 +124,12 @@ Transaction.prototype.rollback = function(callback) {
   var req = new pb.RollbackRequest({ transaction: this.id });
   var res = pb.RollbackResponse;
   this.makeReq('rollback', req, res, function(err) {
-    if (!err) {
-      that.isFinalized = true;
+    if (err) {
+      callback(err);
+      return;
     }
-    callback(err);
+    that.isFinalized = true;
+    callback(null);
   });
 };
 
@@ -149,10 +153,12 @@ Transaction.prototype.commit = function(callback) {
   var req = new pb.CommitRequest({ transaction: this.id });
   var res = pb.CommitResponse;
   this.makeReq('commit', req, res, function(err) {
-    if (!err) {
-      that.isFinalized = true;
+    if (err) {
+      callback(err);
+      return;
     }
-    callback(err);
+    that.isFinalized = true;
+    callback(null);
   });
 };
 
@@ -172,9 +178,10 @@ Transaction.prototype.commit = function(callback) {
  */
 Transaction.prototype.finalize = function(callback) {
   if (!this.isFinalized) {
-    return this.commit(callback);
+    this.commit(callback);
+    return;
   }
-  callback();
+  setImmediate(callback);
 };
 
 /**
@@ -303,7 +310,8 @@ Transaction.prototype.save = function(entities, callback) {
   var res = pb.CommitResponse;
   this.makeReq('commit', req, res, function(err, resp) {
     if (err || !resp) {
-      return callback(err);
+      callback(err);
+      return;
     }
     var autoInserted = (resp.mutation_result.insert_auto_id_key || []);
     autoInserted.forEach(function(key, index) {
@@ -387,7 +395,8 @@ Transaction.prototype.runQuery = function(q, callback) {
   var res = pb.RunQueryResponse;
   this.makeReq('runQuery', req, res, function(err, resp) {
     if (err || !resp.batch || !resp.batch.entity_result) {
-      return callback(err);
+      callback(err);
+      return;
     }
     var nextQuery = null;
     if (resp.batch.end_cursor) {
@@ -451,7 +460,7 @@ Transaction.prototype.makeReq = function(method, req, respType, callback) {
             callback(err);
             return;
           }
-          callback(null, respType.decode(buffer));        
+          callback(null, respType.decode(buffer));
         });
       });
     });

lib/pubsub/index.js

@@ -137,10 +137,11 @@ Subscription.prototype.del = function(callback) {
   });
   this.conn.makeReq('DELETE', path, null, true, function(err) {
     if (err) {
-      return callback(err);
+      callback(err);
+      return;
     }
     that.closed = true;
-    callback(err);
+    callback(null);
   });
 };
 
@@ -264,7 +265,8 @@ Connection.prototype.listSubscriptions = function(query, callback) {
 
   this.makeReq('GET', 'subscriptions', q, true, function(err, result) {
     if (err) {
-      return callback(err);
+      callback(err);
+      return;
     }
     var items = result.subscription || [];
     var subscriptions = items.map(function(item) {
@@ -351,7 +353,10 @@ Connection.prototype.listTopics = function(query, callback) {
   var q = util.extend({}, query);
   q.query = 'cloud.googleapis.com/project in (' + this.fullProjectName_() + ')';
   this.makeReq('GET', 'topics', q, true, function(err, result) {
-    if (err) { return callback(err); }
+    if (err) {
+      callback(err);
+      return;
+    }
     var items = result.topic || [];
     var topics = items.map(function(item) {
       return new Topic(that, item.name);
@@ -372,13 +377,14 @@ Connection.prototype.listTopics = function(query, callback) {
  */
 Connection.prototype.getTopic = function(name, callback) {
   var that = this;
-  var cb = callback || util.noop;
+  callback = callback || util.noop;
   var fullName = this.fullTopicName_(name);
   this.makeReq('GET', 'topics/' + fullName, null, true, function(err) {
     if (err) {
-      return cb(err);
+      callback(err);
+      return;
     }
-    cb(null, new Topic(that, fullName));
+    callback(null, new Topic(that, fullName));
   });
 };
 
@@ -389,10 +395,14 @@ Connection.prototype.getTopic = function(name, callback) {
  */
 Connection.prototype.createTopic = function(name, callback) {
   var that = this;
-  var cb = callback || util.noop;
+  callback = callback || util.noop;
   var fullName = this.fullTopicName_(name);
   this.makeReq('POST', 'topics', null, { name: fullName }, function(err) {
-    cb(err, new Topic(that, fullName));
+    if (err) {
+      callback(err);
+      return;
+    }
+    callback(null, new Topic(that, fullName));
   });
 };
 

lib/storage/index.js

@@ -161,7 +161,8 @@ Bucket.prototype.list = function(query, callback) {
   }
   this.makeReq('GET', 'o', query, true, function(err, resp) {
     if (err) {
-      return callback(err);
+      callback(err);
+      return;
     }
     var nextQuery = null;
     if (resp.nextPageToken) {