refactor(collection): move operations in collection to their own classes

Fixes NODE-1897

Author: kvwalker

lib/collection.js

@@ -0,0 +1,127 @@
+'use strict';
+
+const OperationBase = require('./operation').OperationBase;
+const AggregationCursor = require('../aggregation_cursor');
+const applyWriteConcern = require('../utils').applyWriteConcern;
+const decorateWithCollation = require('../utils').decorateWithCollation;
+const decorateWithReadConcern = require('../utils').decorateWithReadConcern;
+const handleCallback = require('../utils').handleCallback;
+const MongoError = require('mongodb-core').MongoError;
+const resolveReadPreference = require('../utils').resolveReadPreference;
+const toError = require('../utils').toError;
+
+const DB_AGGREGATE_COLLECTION = 1;
+
+class AggregateOperation extends OperationBase {
+  constructor(db, collection, pipeline, options) {
+    super(options);
+
+    this.db = db;
+    this.collection = collection;
+    this.pipeline = pipeline;
+  }
+
+  execute(callback) {
+    const db = this.db;
+    const coll = this.collection;
+    let pipeline = this.pipeline;
+    let options = this.options;
+
+    const isDbAggregate = typeof coll === 'string';
+    const target = isDbAggregate ? db : coll;
+    const topology = target.s.topology;
+    let hasOutStage = false;
+
+    if (typeof options.out === 'string') {
+      pipeline = pipeline.concat({ $out: options.out });
+      hasOutStage = true;
+    } else if (pipeline.length > 0 && pipeline[pipeline.length - 1]['$out']) {
+      hasOutStage = true;
+    }
+
+    let command;
+    let namespace;
+    let optionSources;
+
+    if (isDbAggregate) {
+      command = { aggregate: DB_AGGREGATE_COLLECTION, pipeline: pipeline };
+      namespace = `${db.s.databaseName}.${DB_AGGREGATE_COLLECTION}`;
+
+      optionSources = { db };
+    } else {
+      command = { aggregate: coll.s.name, pipeline: pipeline };
+      namespace = coll.s.namespace;
+
+      optionSources = { db: coll.s.db, collection: coll };
+    }
+
+    const takesWriteConcern = topology.capabilities().commandsTakeWriteConcern;
+
+    if (!hasOutStage) {
+      decorateWithReadConcern(command, target, options);
+    }
+
+    if (pipeline.length > 0 && pipeline[pipeline.length - 1]['$out'] && takesWriteConcern) {
+      applyWriteConcern(command, optionSources, options);
+    }
+
+    try {
+      decorateWithCollation(command, target, options);
+    } catch (err) {
+      if (typeof callback === 'function') return callback(err, null);
+      throw err;
+    }
+
+    if (options.bypassDocumentValidation === true) {
+      command.bypassDocumentValidation = options.bypassDocumentValidation;
+    }
+
+    if (typeof options.allowDiskUse === 'boolean') command.allowDiskUse = options.allowDiskUse;
+    if (typeof options.maxTimeMS === 'number') command.maxTimeMS = options.maxTimeMS;
+
+    if (options.hint) command.hint = options.hint;
+
+    options = Object.assign({}, options);
+
+    // Ensure we have the right read preference inheritance
+    options.readPreference = resolveReadPreference(options, optionSources);
+
+    if (options.explain) {
+      if (command.readConcern || command.writeConcern) {
+        throw toError(
+          '"explain" cannot be used on an aggregate call with readConcern/writeConcern'
+        );
+      }
+      command.explain = options.explain;
+    }
+
+    if (typeof options.comment === 'string') command.comment = options.comment;
+
+    // Validate that cursor options is valid
+    if (options.cursor != null && typeof options.cursor !== 'object') {
+      throw toError('cursor options must be an object');
+    }
+
+    options.cursor = options.cursor || {};
+    if (options.batchSize && !hasOutStage) options.cursor.batchSize = options.batchSize;
+    command.cursor = options.cursor;
+
+    // promiseLibrary
+    options.promiseLibrary = target.s.promiseLibrary;
+
+    // Set the AggregationCursor constructor
+    options.cursorFactory = AggregationCursor;
+
+    if (typeof callback !== 'function') {
+      if (!topology.capabilities()) {
+        throw new MongoError('cannot connect to server');
+      }
+
+      return topology.cursor(namespace, command, options);
+    }
+
+    return handleCallback(callback, null, topology.cursor(namespace, command, options));
+  }
+}
+
+module.exports = AggregateOperation;

lib/operations/aggregate_operation.js

@@ -0,0 +1,104 @@
+'use strict';
+
+const applyRetryableWrites = require('../utils').applyRetryableWrites;
+const applyWriteConcern = require('../utils').applyWriteConcern;
+const MongoError = require('mongodb-core').MongoError;
+const OperationBase = require('./operation').OperationBase;
+
+class BulkWriteOperation extends OperationBase {
+  constructor(collection, operations, options) {
+    super(options);
+
+    this.collection = collection;
+    this.operations = operations;
+  }
+
+  execute(callback) {
+    const coll = this.collection;
+    const operations = this.operations;
+    let options = this.options;
+
+    // Add ignoreUndfined
+    if (coll.s.options.ignoreUndefined) {
+      options = Object.assign({}, options);
+      options.ignoreUndefined = coll.s.options.ignoreUndefined;
+    }
+
+    // Create the bulk operation
+    const bulk =
+      options.ordered === true || options.ordered == null
+        ? coll.initializeOrderedBulkOp(options)
+        : coll.initializeUnorderedBulkOp(options);
+
+    // Do we have a collation
+    let collation = false;
+
+    // for each op go through and add to the bulk
+    try {
+      for (let i = 0; i < operations.length; i++) {
+        // Get the operation type
+        const key = Object.keys(operations[i])[0];
+        // Check if we have a collation
+        if (operations[i][key].collation) {
+          collation = true;
+        }
+
+        // Pass to the raw bulk
+        bulk.raw(operations[i]);
+      }
+    } catch (err) {
+      return callback(err, null);
+    }
+
+    // Final options for retryable writes and write concern
+    let finalOptions = Object.assign({}, options);
+    finalOptions = applyRetryableWrites(finalOptions, coll.s.db);
+    finalOptions = applyWriteConcern(finalOptions, { db: coll.s.db, collection: coll }, options);
+
+    const writeCon = finalOptions.writeConcern ? finalOptions.writeConcern : {};
+    const capabilities = coll.s.topology.capabilities();
+
+    // Did the user pass in a collation, check if our write server supports it
+    if (collation && capabilities && !capabilities.commandsTakeCollation) {
+      return callback(new MongoError('server/primary/mongos does not support collation'));
+    }
+
+    // Execute the bulk
+    bulk.execute(writeCon, finalOptions, (err, r) => {
+      // We have connection level error
+      if (!r && err) {
+        return callback(err, null);
+      }
+
+      r.insertedCount = r.nInserted;
+      r.matchedCount = r.nMatched;
+      r.modifiedCount = r.nModified || 0;
+      r.deletedCount = r.nRemoved;
+      r.upsertedCount = r.getUpsertedIds().length;
+      r.upsertedIds = {};
+      r.insertedIds = {};
+
+      // Update the n
+      r.n = r.insertedCount;
+
+      // Inserted documents
+      const inserted = r.getInsertedIds();
+      // Map inserted ids
+      for (let i = 0; i < inserted.length; i++) {
+        r.insertedIds[inserted[i].index] = inserted[i]._id;
+      }
+
+      // Upserted documents
+      const upserted = r.getUpsertedIds();
+      // Map upserted ids
+      for (let i = 0; i < upserted.length; i++) {
+        r.upsertedIds[upserted[i].index] = upserted[i]._id;
+      }
+
+      // Return the results
+      callback(null, r);
+    });
+  }
+}
+
+module.exports = BulkWriteOperation;