gerrit.onap Code Review - aai/esr-gui.git/blob

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