You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
354 lines
13 KiB
354 lines
13 KiB
'use strict';
|
|
|
|
const applyWriteConcern = require('../utils').applyWriteConcern;
|
|
const Code = require('../core').BSON.Code;
|
|
const createIndexDb = require('./db_ops').createIndex;
|
|
const decorateWithCollation = require('../utils').decorateWithCollation;
|
|
const decorateWithReadConcern = require('../utils').decorateWithReadConcern;
|
|
const ensureIndexDb = require('./db_ops').ensureIndex;
|
|
const evaluate = require('./db_ops').evaluate;
|
|
const executeCommand = require('./db_ops').executeCommand;
|
|
const handleCallback = require('../utils').handleCallback;
|
|
const indexInformationDb = require('./db_ops').indexInformation;
|
|
const Long = require('../core').BSON.Long;
|
|
const MongoError = require('../core').MongoError;
|
|
const ReadPreference = require('../core').ReadPreference;
|
|
const insertDocuments = require('./common_functions').insertDocuments;
|
|
const updateDocuments = require('./common_functions').updateDocuments;
|
|
|
|
/**
|
|
* Group function helper
|
|
* @ignore
|
|
*/
|
|
// var groupFunction = function () {
|
|
// var c = db[ns].find(condition);
|
|
// var map = new Map();
|
|
// var reduce_function = reduce;
|
|
//
|
|
// while (c.hasNext()) {
|
|
// var obj = c.next();
|
|
// var key = {};
|
|
//
|
|
// for (var i = 0, len = keys.length; i < len; ++i) {
|
|
// var k = keys[i];
|
|
// key[k] = obj[k];
|
|
// }
|
|
//
|
|
// var aggObj = map.get(key);
|
|
//
|
|
// if (aggObj == null) {
|
|
// var newObj = Object.extend({}, key);
|
|
// aggObj = Object.extend(newObj, initial);
|
|
// map.put(key, aggObj);
|
|
// }
|
|
//
|
|
// reduce_function(obj, aggObj);
|
|
// }
|
|
//
|
|
// return { "result": map.values() };
|
|
// }.toString();
|
|
const groupFunction =
|
|
'function () {\nvar c = db[ns].find(condition);\nvar map = new Map();\nvar reduce_function = reduce;\n\nwhile (c.hasNext()) {\nvar obj = c.next();\nvar key = {};\n\nfor (var i = 0, len = keys.length; i < len; ++i) {\nvar k = keys[i];\nkey[k] = obj[k];\n}\n\nvar aggObj = map.get(key);\n\nif (aggObj == null) {\nvar newObj = Object.extend({}, key);\naggObj = Object.extend(newObj, initial);\nmap.put(key, aggObj);\n}\n\nreduce_function(obj, aggObj);\n}\n\nreturn { "result": map.values() };\n}';
|
|
|
|
/**
|
|
* Create an index on the db and collection.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {(string|object)} fieldOrSpec Defines the index.
|
|
* @param {object} [options] Optional settings. See Collection.prototype.createIndex for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
*/
|
|
function createIndex(coll, fieldOrSpec, options, callback) {
|
|
createIndexDb(coll.s.db, coll.collectionName, fieldOrSpec, options, callback);
|
|
}
|
|
|
|
/**
|
|
* Create multiple indexes in the collection. This method is only supported for
|
|
* MongoDB 2.6 or higher. Earlier version of MongoDB will throw a command not supported
|
|
* error. Index specifications are defined at http://docs.mongodb.org/manual/reference/command/createIndexes/.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {array} indexSpecs An array of index specifications to be created
|
|
* @param {Object} [options] Optional settings. See Collection.prototype.createIndexes for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
*/
|
|
function createIndexes(coll, indexSpecs, options, callback) {
|
|
const capabilities = coll.s.topology.capabilities();
|
|
|
|
// Ensure we generate the correct name if the parameter is not set
|
|
for (let i = 0; i < indexSpecs.length; i++) {
|
|
if (indexSpecs[i].name == null) {
|
|
const keys = [];
|
|
|
|
// Did the user pass in a collation, check if our write server supports it
|
|
if (indexSpecs[i].collation && capabilities && !capabilities.commandsTakeCollation) {
|
|
return callback(new MongoError('server/primary/mongos does not support collation'));
|
|
}
|
|
|
|
for (let name in indexSpecs[i].key) {
|
|
keys.push(`${name}_${indexSpecs[i].key[name]}`);
|
|
}
|
|
|
|
// Set the name
|
|
indexSpecs[i].name = keys.join('_');
|
|
}
|
|
}
|
|
|
|
options = Object.assign({}, options, { readPreference: ReadPreference.PRIMARY });
|
|
|
|
// Execute the index
|
|
executeCommand(
|
|
coll.s.db,
|
|
{
|
|
createIndexes: coll.collectionName,
|
|
indexes: indexSpecs
|
|
},
|
|
options,
|
|
callback
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Ensure that an index exists. If the index does not exist, this function creates it.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {(string|object)} fieldOrSpec Defines the index.
|
|
* @param {object} [options] Optional settings. See Collection.prototype.ensureIndex for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
*/
|
|
function ensureIndex(coll, fieldOrSpec, options, callback) {
|
|
ensureIndexDb(coll.s.db, coll.collectionName, fieldOrSpec, options, callback);
|
|
}
|
|
|
|
/**
|
|
* Run a group command across a collection.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {(object|array|function|code)} keys An object, array or function expressing the keys to group by.
|
|
* @param {object} condition An optional condition that must be true for a row to be considered.
|
|
* @param {object} initial Initial value of the aggregation counter object.
|
|
* @param {(function|Code)} reduce The reduce function aggregates (reduces) the objects iterated
|
|
* @param {(function|Code)} finalize An optional function to be run on each item in the result set just before the item is returned.
|
|
* @param {boolean} command Specify if you wish to run using the internal group command or using eval, default is true.
|
|
* @param {object} [options] Optional settings. See Collection.prototype.group for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
* @deprecated MongoDB 3.6 or higher will no longer support the group command. We recommend rewriting using the aggregation framework.
|
|
*/
|
|
function group(coll, keys, condition, initial, reduce, finalize, command, options, callback) {
|
|
// Execute using the command
|
|
if (command) {
|
|
const reduceFunction = reduce && reduce._bsontype === 'Code' ? reduce : new Code(reduce);
|
|
|
|
const selector = {
|
|
group: {
|
|
ns: coll.collectionName,
|
|
$reduce: reduceFunction,
|
|
cond: condition,
|
|
initial: initial,
|
|
out: 'inline'
|
|
}
|
|
};
|
|
|
|
// if finalize is defined
|
|
if (finalize != null) selector.group['finalize'] = finalize;
|
|
// Set up group selector
|
|
if ('function' === typeof keys || (keys && keys._bsontype === 'Code')) {
|
|
selector.group.$keyf = keys && keys._bsontype === 'Code' ? keys : new Code(keys);
|
|
} else {
|
|
const hash = {};
|
|
keys.forEach(key => {
|
|
hash[key] = 1;
|
|
});
|
|
selector.group.key = hash;
|
|
}
|
|
|
|
options = Object.assign({}, options);
|
|
// Ensure we have the right read preference inheritance
|
|
options.readPreference = ReadPreference.resolve(coll, options);
|
|
|
|
// Do we have a readConcern specified
|
|
decorateWithReadConcern(selector, coll, options);
|
|
|
|
// Have we specified collation
|
|
try {
|
|
decorateWithCollation(selector, coll, options);
|
|
} catch (err) {
|
|
return callback(err, null);
|
|
}
|
|
|
|
// Execute command
|
|
executeCommand(coll.s.db, selector, options, (err, result) => {
|
|
if (err) return handleCallback(callback, err, null);
|
|
handleCallback(callback, null, result.retval);
|
|
});
|
|
} else {
|
|
// Create execution scope
|
|
const scope = reduce != null && reduce._bsontype === 'Code' ? reduce.scope : {};
|
|
|
|
scope.ns = coll.collectionName;
|
|
scope.keys = keys;
|
|
scope.condition = condition;
|
|
scope.initial = initial;
|
|
|
|
// Pass in the function text to execute within mongodb.
|
|
const groupfn = groupFunction.replace(/ reduce;/, reduce.toString() + ';');
|
|
|
|
evaluate(coll.s.db, new Code(groupfn, scope), null, options, (err, results) => {
|
|
if (err) return handleCallback(callback, err, null);
|
|
handleCallback(callback, null, results.result || results);
|
|
});
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Retrieve all the indexes on the collection.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {Object} [options] Optional settings. See Collection.prototype.indexes for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
*/
|
|
function indexes(coll, options, callback) {
|
|
options = Object.assign({}, { full: true }, options);
|
|
indexInformationDb(coll.s.db, coll.collectionName, options, callback);
|
|
}
|
|
|
|
/**
|
|
* Check if one or more indexes exist on the collection. This fails on the first index that doesn't exist.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {(string|array)} indexes One or more index names to check.
|
|
* @param {Object} [options] Optional settings. See Collection.prototype.indexExists for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
*/
|
|
function indexExists(coll, indexes, options, callback) {
|
|
indexInformation(coll, options, (err, indexInformation) => {
|
|
// If we have an error return
|
|
if (err != null) return handleCallback(callback, err, null);
|
|
// Let's check for the index names
|
|
if (!Array.isArray(indexes))
|
|
return handleCallback(callback, null, indexInformation[indexes] != null);
|
|
// Check in list of indexes
|
|
for (let i = 0; i < indexes.length; i++) {
|
|
if (indexInformation[indexes[i]] == null) {
|
|
return handleCallback(callback, null, false);
|
|
}
|
|
}
|
|
|
|
// All keys found return true
|
|
return handleCallback(callback, null, true);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Retrieve this collection's index info.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {object} [options] Optional settings. See Collection.prototype.indexInformation for a list of options.
|
|
* @param {Collection~resultCallback} [callback] The command result callback
|
|
*/
|
|
function indexInformation(coll, options, callback) {
|
|
indexInformationDb(coll.s.db, coll.collectionName, options, callback);
|
|
}
|
|
|
|
/**
|
|
* Return N parallel cursors for a collection to allow parallel reading of the entire collection. There are
|
|
* no ordering guarantees for returned results.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {object} [options] Optional settings. See Collection.prototype.parallelCollectionScan for a list of options.
|
|
* @param {Collection~parallelCollectionScanCallback} [callback] The command result callback
|
|
*/
|
|
function parallelCollectionScan(coll, options, callback) {
|
|
// Create command object
|
|
const commandObject = {
|
|
parallelCollectionScan: coll.collectionName,
|
|
numCursors: options.numCursors
|
|
};
|
|
|
|
// Do we have a readConcern specified
|
|
decorateWithReadConcern(commandObject, coll, options);
|
|
|
|
// Store the raw value
|
|
const raw = options.raw;
|
|
delete options['raw'];
|
|
|
|
// Execute the command
|
|
executeCommand(coll.s.db, commandObject, options, (err, result) => {
|
|
if (err) return handleCallback(callback, err, null);
|
|
if (result == null)
|
|
return handleCallback(
|
|
callback,
|
|
new Error('no result returned for parallelCollectionScan'),
|
|
null
|
|
);
|
|
|
|
options = Object.assign({ explicitlyIgnoreSession: true }, options);
|
|
|
|
const cursors = [];
|
|
// Add the raw back to the option
|
|
if (raw) options.raw = raw;
|
|
// Create command cursors for each item
|
|
for (let i = 0; i < result.cursors.length; i++) {
|
|
const rawId = result.cursors[i].cursor.id;
|
|
// Convert cursorId to Long if needed
|
|
const cursorId = typeof rawId === 'number' ? Long.fromNumber(rawId) : rawId;
|
|
// Add a command cursor
|
|
cursors.push(coll.s.topology.cursor(coll.namespace, cursorId, options));
|
|
}
|
|
|
|
handleCallback(callback, null, cursors);
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Save a document.
|
|
*
|
|
* @method
|
|
* @param {Collection} a Collection instance.
|
|
* @param {object} doc Document to save
|
|
* @param {object} [options] Optional settings. See Collection.prototype.save for a list of options.
|
|
* @param {Collection~writeOpCallback} [callback] The command result callback
|
|
* @deprecated use insertOne, insertMany, updateOne or updateMany
|
|
*/
|
|
function save(coll, doc, options, callback) {
|
|
// Get the write concern options
|
|
const finalOptions = applyWriteConcern(
|
|
Object.assign({}, options),
|
|
{ db: coll.s.db, collection: coll },
|
|
options
|
|
);
|
|
// Establish if we need to perform an insert or update
|
|
if (doc._id != null) {
|
|
finalOptions.upsert = true;
|
|
return updateDocuments(coll, { _id: doc._id }, doc, finalOptions, callback);
|
|
}
|
|
|
|
// Insert the document
|
|
insertDocuments(coll, [doc], finalOptions, (err, result) => {
|
|
if (callback == null) return;
|
|
if (doc == null) return handleCallback(callback, null, null);
|
|
if (err) return handleCallback(callback, err, null);
|
|
handleCallback(callback, null, result);
|
|
});
|
|
}
|
|
|
|
module.exports = {
|
|
createIndex,
|
|
createIndexes,
|
|
ensureIndex,
|
|
group,
|
|
indexes,
|
|
indexExists,
|
|
indexInformation,
|
|
parallelCollectionScan,
|
|
save
|
|
};
|