manually run runQuery as a stream

Author: stephenplusplus

lib/datastore/index.js

@@ -192,19 +192,41 @@ var util = require('../common/util.js');
  * //-
  * // <h3>Paginating Records</h3>
  * //
- * // By default, all records are returned at once. To control pagination, you
- * // can disable this on your query with {module:datastore/query#autoPaginate}.
+ * // Imagine building a website that allows a user to sift through hundreds of
+ * // their contacts. You'll likely want to only display a subset of these at
+ * // once, so you set a limit.
  * //-
- * query.autoPaginate(false);
+ * var express = require('express');
+ * var app = express();
  *
- * datastore.runQuery(query, function(err, entities, nextQuery) {
- *   // entities = [...];
+ * var NUM_RESULTS_PER_PAGE = 15;
  *
- *   if (nextQuery) {
- *     // Run `datastore.runQuery` again with the prepared query to retrieve the
- *     // next set of results.
- *     datastore.runQuery(nextQuery, function(err, entities, nextQuery) {});
+ * app.get('/contacts', function(req, res) {
+ *   var query = datastore.createQuery('Contacts')
+ *     .limit(NUM_RESULTS_PER_PAGE);
+ *
+ *   if (req.query.startCursor) {
+ *     query.start(req.query.startCursor);
  *   }
+ *
+ *   datastore.runQuery(query, function(err, entities, info) {
+ *     if (err) {
+ *       // Error handling omitted...
+ *       return;
+ *     }
+ *
+ *     // Respond to the front end with the contacts and the cursoring token
+ *     // from the query we just ran.
+ *     var frontEndResponse = {
+ *       contacts: entities
+ *     };
+ *
+ *     if (info.moreResults !== 'NO_MORE_RESULTS') {
+ *       frontEndResponse.startCursor = info.endCursor;
+ *     }
+ *
+ *     res.render('contacts', frontEndResponse);
+ *   });
  * });
  *
  * //-

lib/datastore/query.js

@@ -287,35 +287,29 @@ Query.prototype.offset = function(n) {
  *     stream instance is returned.
  * @param {?error} callback.err - An error returned while making this request
  * @param {object[]} callback.entities - A list of entities.
- * @param {?object} callback.nextQuery - If present, query with this object to
- *     check for more results.
+ * @param {object} callback.info - An object useful for pagination.
+ * @param {?string} callback.info.endCursor - Use this in a follow-up query to
+ *     begin from where these results ended.
+ * @param {string} callback.info.moreResults - Datastore responds with one of:
+ *     - `MORE_RESULTS_AFTER_LIMIT`: There *may* be more results after the
+ *       specified limit.
+ *     - `MORE_RESULTS_AFTER_CURSOR`: There *may* be more results after the
+ *       specified end cursor.
+ *     - `NO_MORE_RESULTS`: There are no more results.
  * @param {object} callback.apiResponse - The full API response.
  *
  * @example
- * query.run(function(err, entities) {});
+ * query.run(function(err, entities, info, apiResponse) {});
  *
  * //-
- * // To control how many API requests are made and page through the results
- * // manually, call `autoPaginate(false)` on your query.
- * //-
- * query.autoPaginate(false);
- *
- * function callback(err, entities, nextQuery, apiResponse) {
- *   if (nextQuery) {
- *     // More results might exist.
- *     nextQuery.run(callback);
- *   }
- * }
- *
- * query.run(callback);
- *
- * //-
- * // If you omit the callback, `run` will automatically call subsequent queries
- * // until no results remain. Entity objects will be pushed as they are found.
+ * // If you omit the callback, you will get the matching entities in a readable
+ * // object stream.
  * //-
  * query.run()
  *   .on('error', console.error)
  *   .on('data', function (entity) {})
+ *   .on('info', function(info) {})
+ *   .on('response', function(apiResponse) {})
  *   .on('end', function() {
  *     // All entities retrieved.
  *   });

lib/datastore/request.js

@@ -383,8 +383,15 @@ DatastoreRequest.prototype.insert = function(entities, callback) {
  *     stream instance is returned.
  * @param {?error} callback.err - An error returned while making this request
  * @param {object[]} callback.entities - A list of entities.
- * @param {?object} callback.nextQuery - If present, query with this object to
- *     check for more results.
+ * @param {object} callback.info - An object useful for pagination.
+ * @param {?string} callback.info.endCursor - Use this in a follow-up query to
+ *     begin from where these results ended.
+ * @param {string} callback.info.moreResults - Datastore responds with one of:
+ *     - `MORE_RESULTS_AFTER_LIMIT`: There *may* be more results after the
+ *       specified limit.
+ *     - `MORE_RESULTS_AFTER_CURSOR`: There *may* be more results after the
+ *       specified end cursor.
+ *     - `NO_MORE_RESULTS`: There are no more results.
  * @param {object} callback.apiResponse - The full API response.
  *
  * @example
@@ -394,7 +401,7 @@ DatastoreRequest.prototype.insert = function(entities, callback) {
  * //-
  * var query = datastore.createQuery('Lion');
  *
- * datastore.runQuery(query, function(err, entities) {});
+ * datastore.runQuery(query, function(err, entities, info, apiResponse) {});
  *
  * //-
  * // Or, if you're using a transaction object.
@@ -406,28 +413,14 @@ DatastoreRequest.prototype.insert = function(entities, callback) {
  * });
  *
  * //-
- * // To control how many API requests are made and page through the results
- * // manually, call `autoPaginate(false)` on your query.
- * //-
- * var manualPageQuery = datastore.createQuery('Lion').autoPaginate(false);
- *
- * function callback(err, entities, nextQuery, apiResponse) {
- *   if (nextQuery) {
- *     // More results might exist.
- *     transaction.runQuery(nextQuery, callback);
- *   }
- * }
- *
- * datastore.runQuery(manualPageQuery, callback);
- *
- * //-
- * // If you omit the callback, runQuery will automatically call subsequent
- * // queries until no results remain. Entity objects will be pushed as they are
- * // found.
+ * // If you omit the callback, you will get the matching entities in a readable
+ * // object stream.
  * //-
  * datastore.runQuery(query)
  *   .on('error', console.error)
  *   .on('data', function (entity) {})
+ *   .on('info', function(info) {})
+ *   .on('response', function(apiResponse) {})
  *   .on('end', function() {
  *     // All entities retrieved.
  *   });
@@ -453,15 +446,35 @@ DatastoreRequest.prototype.insert = function(entities, callback) {
  * });
  */
 DatastoreRequest.prototype.runQuery = function(query, options, callback) {
-  var self = this;
-
   if (is.fn(options)) {
     callback = options;
     options = {};
   }
 
   options = options || {};
 
+  var apiResponse;
+  var info;
+
+  if (is.fn(callback)) {
+    // Run this method in stream mode and send the results back to the callback.
+    this.runQuery(query, options)
+      .on('error', callback)
+      .on('response', function(apiResponse_) {
+        apiResponse = apiResponse_;
+      })
+      .on('info', function(info_) {
+        info = info_;
+      })
+      .pipe(concat(function(results) {
+        callback(null, results, info, apiResponse);
+      }));
+    return;
+  }
+
+  var self = this;
+  var stream = through.obj();
+
   var protoOpts = {
     service: 'Datastore',
     method: 'runQuery'
@@ -483,18 +496,12 @@ DatastoreRequest.prototype.runQuery = function(query, options, callback) {
     };
   }
 
-  var entities = [];
-
   function onResponse(err, resp) {
     if (err) {
-      callback(err, null, null, resp);
+      stream.destroy(err);
       return;
     }
 
-    if (resp.batch.entityResults) {
-      entities = entities.concat(entity.formatArray(resp.batch.entityResults));
-    }
-
     var info = {
       moreResults: resp.batch.moreResults
     };
@@ -503,7 +510,26 @@ DatastoreRequest.prototype.runQuery = function(query, options, callback) {
       info.endCursor = resp.batch.endCursor;
     }
 
-    if (resp.batch.moreResults === 'NOT_FINISHED') {
+    var entities = [];
+
+    if (resp.batch.entityResults) {
+      entities = entity.formatArray(resp.batch.entityResults);
+    }
+
+    // Emit each result right away, then get the rest if necessary.
+    split(entities, stream, function(streamEnded) {
+      if (streamEnded) {
+        return;
+      }
+
+      if (resp.batch.moreResults !== 'NOT_FINISHED') {
+        stream.emit('response', resp);
+        stream.emit('info', info);
+        stream.push(null);
+        return;
+      }
+
+      // The query is "NOT_FINISHED". Get the rest of the results.
       var offset = query.offsetVal === -1 ? 0 : query.offsetVal;
 
       var continuationQuery = extend(true, new Query(), query)
@@ -517,13 +543,12 @@ DatastoreRequest.prototype.runQuery = function(query, options, callback) {
 
       reqOpts.query = entity.queryToQueryProto(continuationQuery);
       self.request_(protoOpts, reqOpts, onResponse);
-      return;
-    }
-
-    callback(null, entities, info, resp);
+    });
   }
 
   this.request_(protoOpts, reqOpts, onResponse);
+
+  return stream;
 };
 
 /**

package.json

@@ -127,6 +127,7 @@
     "coveralls": "^2.11.2",
     "deep-strict-equal": "^0.1.0",
     "dox": "^0.8.0",
+    "express": "^4.13.4",
     "glob": "^5.0.9",
     "globby": "^3.0.1",
     "istanbul": "^0.3.5",

system-test/datastore.js

@@ -439,9 +439,11 @@ describe.only('Datastore', function() {
         assert.ifError(err);
         assert.strictEqual(firstEntities.length, 5);
 
-        q.start(info.endCursor).limit(q.limitVal - firstEntities.length);
+        var secondQ = datastore.createQuery('Character')
+          .hasAncestor(ancestor)
+          .start(info.endCursor);
 
-        datastore.runQuery(q, function(err, secondEntities) {
+        datastore.runQuery(secondQ, function(err, secondEntities) {
           assert.ifError(err);
           assert.strictEqual(secondEntities.length, 3);
           done();
@@ -462,7 +464,7 @@ describe.only('Datastore', function() {
       });
     });
 
-    it.skip('should run a query as a stream', function(done) {
+    it('should run a query as a stream', function(done) {
       var q = datastore.createQuery('Character').hasAncestor(ancestor);
 
       var resultsReturned = 0;
@@ -476,7 +478,7 @@ describe.only('Datastore', function() {
         });
     });
 
-    it.skip('should not go over a limit with a stream', function(done) {
+    it('should not go over a limit with a stream', function(done) {
       var limit = 3;
       var q = datastore.createQuery('Character')
         .hasAncestor(ancestor)
@@ -593,11 +595,12 @@ describe.only('Datastore', function() {
         assert.strictEqual(entities[0].data.name, 'Robb');
         assert.strictEqual(entities[2].data.name, 'Catelyn');
 
-        q.start(info.endCursor)
-          .limit(q.limitVal - entities.length)
-          .offset(0);
+        var secondQ = datastore.createQuery('Character')
+          .hasAncestor(ancestor)
+          .order('appearances')
+          .start(info.endCursor);
 
-        datastore.runQuery(q, function(err, secondEntities) {
+        datastore.runQuery(secondQ, function(err, secondEntities) {
           assert.ifError(err);
 
           assert.strictEqual(secondEntities.length, 3);
@@ -619,9 +622,12 @@ describe.only('Datastore', function() {
       datastore.runQuery(q, function(err, entities, info) {
         assert.ifError(err);
 
-        q.start(info.endCursor).limit(-1).offset(-1);
+        var secondQ = datastore.createQuery('Character')
+          .hasAncestor(ancestor)
+          .order('appearances')
+          .start(info.endCursor);
 
-        datastore.runQuery(q, function(err, secondEntities, info) {
+        datastore.runQuery(secondQ, function(err, secondEntities) {
           assert.ifError(err);
 
           assert.strictEqual(secondEntities.length, 4);