Node.js MongoDB Driver API

Source: lib/aggregation_cursor.js

  1. "use strict";
  3. var inherits = require('util').inherits
  4.   , MongoError = require('mongodb-core').MongoError
  5.   , Readable = require('stream').Readable || require('readable-stream').Readable
  6.   , Define = require('./metadata')
  7.   , CoreCursor = require('./cursor');
  9. /**
  10.  * @fileOverview The **AggregationCursor** class is an internal class that embodies an aggregation cursor on MongoDB
  11.  * allowing for iteration over the results returned from the underlying query. It supports
  12.  * one by one document iteration, conversion to an array or can be iterated as a Node 0.10.X
  13.  * or higher stream
  14.  *
  15.  * **AGGREGATIONCURSOR Cannot directly be instantiated**
  16.  * @example
  17.  * var MongoClient = require('mongodb').MongoClient,
  18.  *   test = require('assert');
  19.  * // Connection url
  20.  * var url = 'mongodb://localhost:27017/test';
  21.  * // Connect using MongoClient
  22.  * MongoClient.connect(url, function(err, db) {
  23.  *   // Create a collection we want to drop later
  24.  *   var col = db.collection('createIndexExample1');
  25.  *   // Insert a bunch of documents
  26.  *   col.insert([{a:1, b:1}
  27.  *     , {a:2, b:2}, {a:3, b:3}
  28.  *     , {a:4, b:4}], {w:1}, function(err, result) {
  29.  *     test.equal(null, err);
  30.  *     // Show that duplicate records got dropped
  31.  *     col.aggregation({}, {cursor: {}}).toArray(function(err, items) {
  32.  *       test.equal(null, err);
  33.  *       test.equal(4, items.length);
  34.  *       db.close();
  35.  *     });
  36.  *   });
  37.  * });
  38.  */
  40. /**
  41.  * Namespace provided by the browser.
  42.  * @external Readable
  43.  */
  45. /**
  46.  * Creates a new Aggregation Cursor instance (INTERNAL TYPE, do not instantiate directly)
  47.  * @class AggregationCursor
  48.  * @extends external:Readable
  49.  * @fires AggregationCursor#data
  50.  * @fires AggregationCursor#end
  51.  * @fires AggregationCursor#close
  52.  * @fires AggregationCursor#readable
  53.  * @return {AggregationCursor} an AggregationCursor instance.
  54.  */
  55. var AggregationCursor = function(bson, ns, cmd, options, topology, topologyOptions) {
  56.   CoreCursor.apply(this, Array.prototype.slice.call(arguments, 0));
  57.   var state = AggregationCursor.INIT;
  58.   var streamOptions = {};
  60.   // MaxTimeMS
  61.   var maxTimeMS = null;
  63.   // Get the promiseLibrary
  64.   var promiseLibrary = options.promiseLibrary;
  66.   // No promise library selected fall back
  67.   if(!promiseLibrary) {
  68.     promiseLibrary = typeof global.Promise == 'function' ?
  69.       global.Promise : require('es6-promise').Promise;
  70.   }
  72.   // Set up
  73.   Readable.call(this, {objectMode: true});
  75.   // Internal state
  76.   this.s = {
  77.     // MaxTimeMS
  78.       maxTimeMS: maxTimeMS
  79.     // State
  80.     , state: state
  81.     // Stream options
  82.     , streamOptions: streamOptions
  83.     // BSON
  84.     , bson: bson
  85.     // Namespace
  86.     , ns: ns
  87.     // Command
  88.     , cmd: cmd
  89.     // Options
  90.     , options: options
  91.     // Topology
  92.     , topology: topology
  93.     // Topology Options
  94.     , topologyOptions: topologyOptions
  95.     // Promise library
  96.     , promiseLibrary: promiseLibrary
  97.   }
  98. }
  100. /**
  101.  * AggregationCursor stream data event, fired for each document in the cursor.
  102.  *
  103.  * @event AggregationCursor#data
  104.  * @type {object}
  105.  */
  107. /**
  108.  * AggregationCursor stream end event
  109.  *
  110.  * @event AggregationCursor#end
  111.  * @type {null}
  112.  */
  114. /**
  115.  * AggregationCursor stream close event
  116.  *
  117.  * @event AggregationCursor#close
  118.  * @type {null}
  119.  */
  121. /**
  122.  * AggregationCursor stream readable event
  123.  *
  124.  * @event AggregationCursor#readable
  125.  * @type {null}
  126.  */
  128. // Inherit from Readable
  129. inherits(AggregationCursor, Readable);
  131. // Extend the Cursor
  132. for(var name in CoreCursor.prototype) {
  133.   AggregationCursor.prototype[name] = CoreCursor.prototype[name];
  134. }
  136. var define = AggregationCursor.define = new Define('AggregationCursor', AggregationCursor, true);
  138. /**
  139.  * Set the batch size for the cursor.
  140.  * @method
  141.  * @param {number} value The batchSize for the cursor.
  142.  * @throws {MongoError}
  143.  * @return {AggregationCursor}
  144.  */
  145. AggregationCursor.prototype.batchSize = function(value) {
  146.   if(this.s.state == AggregationCursor.CLOSED || this.isDead()) throw MongoError.create({message: "Cursor is closed", driver:true });
  147.   if(typeof value != 'number') throw MongoError.create({message: "batchSize requires an integer", driver:true });
  148.   if(this.s.cmd.cursor) this.s.cmd.cursor.batchSize = value;
  149.   this.setCursorBatchSize(value);
  150.   return this;
  151. }
  153. define.classMethod('batchSize', {callback: false, promise:false, returns: [AggregationCursor]});
  155. /**
  156.  * Add a geoNear stage to the aggregation pipeline
  157.  * @method
  158.  * @param {object} document The geoNear stage document.
  159.  * @return {AggregationCursor}
  160.  */
  161. AggregationCursor.prototype.geoNear = function(document) {
  162.   this.s.cmd.pipeline.push({$geoNear: document});
  163.   return this;
  164. }
  166. define.classMethod('geoNear', {callback: false, promise:false, returns: [AggregationCursor]});
  168. /**
  169.  * Add a group stage to the aggregation pipeline
  170.  * @method
  171.  * @param {object} document The group stage document.
  172.  * @return {AggregationCursor}
  173.  */
  174. AggregationCursor.prototype.group = function(document) {
  175.   this.s.cmd.pipeline.push({$group: document});
  176.   return this;
  177. }
  179. define.classMethod('group', {callback: false, promise:false, returns: [AggregationCursor]});
  181. /**
  182.  * Add a limit stage to the aggregation pipeline
  183.  * @method
  184.  * @param {number} value The state limit value.
  185.  * @return {AggregationCursor}
  186.  */
  187. AggregationCursor.prototype.limit = function(value) {
  188.   this.s.cmd.pipeline.push({$limit: value});
  189.   return this;
  190. }
  192. define.classMethod('limit', {callback: false, promise:false, returns: [AggregationCursor]});
  194. /**
  195.  * Add a match stage to the aggregation pipeline
  196.  * @method
  197.  * @param {object} document The match stage document.
  198.  * @return {AggregationCursor}
  199.  */
  200. AggregationCursor.prototype.match = function(document) {
  201.   this.s.cmd.pipeline.push({$match: document});
  202.   return this;
  203. }
  205. define.classMethod('match', {callback: false, promise:false, returns: [AggregationCursor]});
  207. /**
  208.  * Add a maxTimeMS stage to the aggregation pipeline
  209.  * @method
  210.  * @param {number} value The state maxTimeMS value.
  211.  * @return {AggregationCursor}
  212.  */
  213. AggregationCursor.prototype.maxTimeMS = function(value) {
  214.   if(this.s.topology.lastIsMaster().minWireVersion > 2) {
  215.     this.s.cmd.maxTimeMS = value;
  216.   }
  217.   return this;
  218. }
  220. define.classMethod('maxTimeMS', {callback: false, promise:false, returns: [AggregationCursor]});
  222. /**
  223.  * Add a out stage to the aggregation pipeline
  224.  * @method
  225.  * @param {number} destination The destination name.
  226.  * @return {AggregationCursor}
  227.  */
  228. AggregationCursor.prototype.out = function(destination) {
  229.   this.s.cmd.pipeline.push({$out: destination});
  230.   return this;
  231. }
  233. define.classMethod('out', {callback: false, promise:false, returns: [AggregationCursor]});
  235. /**
  236.  * Add a project stage to the aggregation pipeline
  237.  * @method
  238.  * @param {object} document The project stage document.
  239.  * @return {AggregationCursor}
  240.  */
  241. AggregationCursor.prototype.project = function(document) {
  242.   this.s.cmd.pipeline.push({$project: document});
  243.   return this;
  244. }
  246. define.classMethod('project', {callback: false, promise:false, returns: [AggregationCursor]});
  248. /**
  249.  * Add a lookup stage to the aggregation pipeline
  250.  * @method
  251.  * @param {object} document The lookup stage document.
  252.  * @return {AggregationCursor}
  253.  */
  254. AggregationCursor.prototype.lookup = function(document) {
  255.   this.s.cmd.pipeline.push({$lookup: document});
  256.   return this;
  257. }
  259. define.classMethod('lookup', {callback: false, promise:false, returns: [AggregationCursor]});
  261. /**
  262.  * Add a redact stage to the aggregation pipeline
  263.  * @method
  264.  * @param {object} document The redact stage document.
  265.  * @return {AggregationCursor}
  266.  */
  267. AggregationCursor.prototype.redact = function(document) {
  268.   this.s.cmd.pipeline.push({$redact: document});
  269.   return this;
  270. }
  272. define.classMethod('redact', {callback: false, promise:false, returns: [AggregationCursor]});
  274. /**
  275.  * Add a skip stage to the aggregation pipeline
  276.  * @method
  277.  * @param {number} value The state skip value.
  278.  * @return {AggregationCursor}
  279.  */
  280. AggregationCursor.prototype.skip = function(value) {
  281.   this.s.cmd.pipeline.push({$skip: value});
  282.   return this;
  283. }
  285. define.classMethod('skip', {callback: false, promise:false, returns: [AggregationCursor]});
  287. /**
  288.  * Add a sort stage to the aggregation pipeline
  289.  * @method
  290.  * @param {object} document The sort stage document.
  291.  * @return {AggregationCursor}
  292.  */
  293. AggregationCursor.prototype.sort = function(document) {
  294.   this.s.cmd.pipeline.push({$sort: document});
  295.   return this;
  296. }
  298. define.classMethod('sort', {callback: false, promise:false, returns: [AggregationCursor]});
  300. /**
  301.  * Add a unwind stage to the aggregation pipeline
  302.  * @method
  303.  * @param {number} field The unwind field name.
  304.  * @return {AggregationCursor}
  305.  */
  306. AggregationCursor.prototype.unwind = function(field) {
  307.   this.s.cmd.pipeline.push({$unwind: field});
  308.   return this;
  309. }
  311. define.classMethod('unwind', {callback: false, promise:false, returns: [AggregationCursor]});
  313. AggregationCursor.prototype.get = AggregationCursor.prototype.toArray;
  315. // Inherited methods
  316. define.classMethod('toArray', {callback: true, promise:true});
  317. define.classMethod('each', {callback: true, promise:false});
  318. define.classMethod('forEach', {callback: true, promise:false});
  319. define.classMethod('hasNext', {callback: true, promise:true});
  320. define.classMethod('next', {callback: true, promise:true});
  321. define.classMethod('close', {callback: true, promise:true});
  322. define.classMethod('isClosed', {callback: false, promise:false, returns: [Boolean]});
  323. define.classMethod('rewind', {callback: false, promise:false});
  324. define.classMethod('bufferedCount', {callback: false, promise:false, returns: [Number]});
  325. define.classMethod('readBufferedDocuments', {callback: false, promise:false, returns: [Array]});
  327. /**
  328.  * Get the next available document from the cursor, returns null if no more documents are available.
  329.  * @function AggregationCursor.prototype.next
  330.  * @param {AggregationCursor~resultCallback} [callback] The result callback.
  331.  * @throws {MongoError}
  332.  * @return {Promise} returns Promise if no callback passed
  333.  */
  335. /**
  336.  * Check if there is any document still available in the cursor
  337.  * @function AggregationCursor.prototype.hasNext
  338.  * @param {AggregationCursor~resultCallback} [callback] The result callback.
  339.  * @throws {MongoError}
  340.  * @return {Promise} returns Promise if no callback passed
  341.  */
  343. /**
  344.  * The callback format for results
  345.  * @callback AggregationCursor~toArrayResultCallback
  346.  * @param {MongoError} error An error instance representing the error during the execution.
  347.  * @param {object[]} documents All the documents the satisfy the cursor.
  348.  */
  350. /**
  351.  * Returns an array of documents. The caller is responsible for making sure that there
  352.  * is enough memory to store the results. Note that the array only contain partial
  353.  * results when this cursor had been previously accessed. In that case,
  354.  * cursor.rewind() can be used to reset the cursor.
  355.  * @method AggregationCursor.prototype.toArray
  356.  * @param {AggregationCursor~toArrayResultCallback} [callback] The result callback.
  357.  * @throws {MongoError}
  358.  * @return {Promise} returns Promise if no callback passed
  359.  */
  361. /**
  362.  * The callback format for results
  363.  * @callback AggregationCursor~resultCallback
  364.  * @param {MongoError} error An error instance representing the error during the execution.
  365.  * @param {(object|null)} result The result object if the command was executed successfully.
  366.  */
  368. /**
  369.  * Iterates over all the documents for this cursor. As with **{cursor.toArray}**,
  370.  * not all of the elements will be iterated if this cursor had been previously accessed.
  371.  * In that case, **{cursor.rewind}** can be used to reset the cursor. However, unlike
  372.  * **{cursor.toArray}**, the cursor will only hold a maximum of batch size elements
  373.  * at any given time if batch size is specified. Otherwise, the caller is responsible
  374.  * for making sure that the entire result can fit the memory.
  375.  * @method AggregationCursor.prototype.each
  376.  * @param {AggregationCursor~resultCallback} callback The result callback.
  377.  * @throws {MongoError}
  378.  * @return {null}
  379.  */
  381. /**
  382.  * Close the cursor, sending a AggregationCursor command and emitting close.
  383.  * @method AggregationCursor.prototype.close
  384.  * @param {AggregationCursor~resultCallback} [callback] The result callback.
  385.  * @return {Promise} returns Promise if no callback passed
  386.  */
  388. /**
  389.  * Is the cursor closed
  390.  * @method AggregationCursor.prototype.isClosed
  391.  * @return {boolean}
  392.  */
  394. /**
  395.  * Execute the explain for the cursor
  396.  * @method AggregationCursor.prototype.explain
  397.  * @param {AggregationCursor~resultCallback} [callback] The result callback.
  398.  * @return {Promise} returns Promise if no callback passed
  399.  */
  401. /**
  402.  * Clone the cursor
  403.  * @function AggregationCursor.prototype.clone
  404.  * @return {AggregationCursor}
  405.  */
  407. /**
  408.  * Resets the cursor
  409.  * @function AggregationCursor.prototype.rewind
  410.  * @return {AggregationCursor}
  411.  */
  413. /**
  414.  * The callback format for the forEach iterator method
  415.  * @callback AggregationCursor~iteratorCallback
  416.  * @param {Object} doc An emitted document for the iterator
  417.  */
  419. /**
  420.  * The callback error format for the forEach iterator method
  421.  * @callback AggregationCursor~endCallback
  422.  * @param {MongoError} error An error instance representing the error during the execution.
  423.  */
  425. /*
  426.  * Iterates over all the documents for this cursor using the iterator, callback pattern.
  427.  * @method AggregationCursor.prototype.forEach
  428.  * @param {AggregationCursor~iteratorCallback} iterator The iteration callback.
  429.  * @param {AggregationCursor~endCallback} callback The end callback.
  430.  * @throws {MongoError}
  431.  * @return {null}
  432.  */
  434. AggregationCursor.INIT = 0;
  435. AggregationCursor.OPEN = 1;
  436. AggregationCursor.CLOSED = 2;
  438. module.exports = AggregationCursor;