Merge branch 'feature/mongo_2.6.5_documentSource' into feature/mongo_2.6.5_documentSource_Project

lib/pipeline/Document.js

@@ -151,7 +151,7 @@ klass.cloneDeep = function cloneDeep(doc) {	//there are casese this is actually
 	for (var key in doc) {
 		if (doc.hasOwnProperty(key)) {
 			var val = doc[key];
-			obj[key] = val instanceof Object && val.constructor === Object ? Document.clone(val) : val;
+			obj[key] = val instanceof Object && val.constructor === Object ? Document.cloneDeep(val) : val;
 		}
 	}
 	return obj;

lib/pipeline/documentSources/DocumentSource.js

@@ -169,8 +169,10 @@ proto.optimize = function optimize() {
 
 klass.GetDepsReturn = {
 	NOT_SUPPORTED: "NOT_SUPPORTED", // This means the set should be ignored
-	EXHAUSTIVE: "EXHAUSTIVE", // This means that everything needed should be in the set
-	SEE_NEXT: "SEE_NEXT" // Add the next Source's deps to the set
+	SEE_NEXT: "SEE_NEXT", // Add the next Source's deps to the set
+	EXHAUSTIVE_FIELDS:"EXHAUSTIVE_FIELDS", // Later stages won"t need more fields from input
+	EXHAUSTIVE_META: "EXHAUSTIVE_META", // Later stages won"t need more metadata from input
+	EXHAUSTIVE_ALL: "EXHAUSTIVE_ALL" // Later stages won"t need either NOTE: This is an | of FIELDS and META in mongo C++
 };
 
 /**

lib/pipeline/documentSources/LimitDocumentSource.js

@@ -10,10 +10,10 @@ var DocumentSource = require('./DocumentSource');
  * @constructor
  * @param [ctx] {ExpressionContext}
  **/
-var LimitDocumentSource = module.exports = function LimitDocumentSource(ctx){
-	if (arguments.length > 1) throw new Error("up to one arg expected");
+var LimitDocumentSource = module.exports = function LimitDocumentSource(ctx, limit){
+	if (arguments.length > 2) throw new Error("up to two args expected");
 	base.call(this, ctx);
-	this.limit = 0;
+	this.limit = limit;
 	this.count = 0;
 }, klass = LimitDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
@@ -44,9 +44,17 @@ proto.coalesce = function coalesce(nextSource) {
 	return true;
 };
 
+/* Returns the execution of the callback against
+* the next documentSource
+* @param {function} callback
+* @return {bool} indicating end of document reached
+*/
 proto.getNext = function getNext(callback) {
 	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
 
+	if (this.expCtx instanceof Object && this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt() === false)
+		return callback(new Error("Interrupted"));
+
 	if (++this.count > this.limit) {
 		this.source.dispose();
 		callback(null, DocumentSource.EOF);
@@ -57,19 +65,25 @@ proto.getNext = function getNext(callback) {
 };
 
 /**
- * Creates a new LimitDocumentSource with the input number as the limit
- * @param {Number} JsonElement this thing is *called* Json, but it expects a number
- **/
-klass.createFromJson = function createFromJson(jsonElement, ctx) {
-	if (typeof jsonElement !== "number") throw new Error("code 15957; the limit must be specified as a number");
+Create a limiting DocumentSource from JSON.
 
-	var Limit = proto.getFactory(),
-		nextLimit = new Limit(ctx);
+This is a convenience method that uses the above, and operates on
+a JSONElement that has been deteremined to be an Object with an
+element named $limit.
 
-	nextLimit.limit = jsonElement;
-	if ((nextLimit.limit <= 0) || isNaN(nextLimit.limit)) throw new Error("code 15958; the limit must be positive");
+@param jsonElement the JSONELement that defines the limit
+@param ctx the expression context
+@returns the grouping DocumentSource
+*/
+klass.createFromJson = function createFromJson(jsonElement, ctx) {
+	if (typeof jsonElement !== "number") throw new Error("code 15957; the limit must be specified as a number");
+	var limit = jsonElement;
+	return this.create(ctx, limit);
+};
 
-	return nextLimit;
+klass.create = function create(ctx, limit){
+	if ((limit <= 0) || isNaN(limit)) throw new Error("code 15958; the limit must be positive");
+	return new LimitDocumentSource(ctx, limit);
 };
 
 proto.getLimit = function getLimit(newLimit) {

lib/pipeline/documentSources/MatchDocumentSource.js

@@ -22,6 +22,13 @@ var MatchDocumentSource = module.exports = function MatchDocumentSource(query, c
 	base.call(this, ctx);
 	this.query = query; // save the query, so we can check it for deps later. THIS IS A DEVIATION FROM THE MONGO IMPLEMENTATION
 	this.matcher = new matcher(query);
+
+	// not supporting currently $text operator
+	// set _isTextQuery to false.
+	// TODO: update after we implement $text.
+	if (klass.isTextQuery(query)) throw new Error("$text pipeline operation not supported");
+	this._isTextQuery = false;
+
 }, klass = MatchDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 klass.matchName = "$match";
@@ -33,6 +40,10 @@ proto.getSourceName = function getSourceName(){
 proto.getNext = function getNext(callback) {
 	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
 
+	if (this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt() === false) {
+		return callback(new Error('Interrupted'));
+	}
+
 	var self = this,
 		next,
 		test = function test(doc) {
@@ -86,11 +97,11 @@ klass.uassertNoDisallowedClauses = function uassertNoDisallowedClauses(query) {
 	for(var key in query){
 		if(query.hasOwnProperty(key)){
 			// can't use the Matcher API because this would segfault the constructor
-			if (query[key] == "$where") throw new Error("code 16395; $where is not allowed inside of a $match aggregation expression");
+			if (key === "$where") throw new Error("code 16395; $where is not allowed inside of a $match aggregation expression");
 			// geo breaks if it is not the first portion of the pipeline
-			if (query[key] == "$near") throw new Error("code 16424; $near is not allowed inside of a $match aggregation expression");
-			if (query[key] == "$within") throw new Error("code 16425; $within is not allowed inside of a $match aggregation expression");
-			if (query[key] == "$nearSphere") throw new Error("code 16426; $nearSphere is not allowed inside of a $match aggregation expression");
+			if (key === "$near") throw new Error("code 16424; $near is not allowed inside of a $match aggregation expression");
+			if (key === "$within") throw new Error("code 16425; $within is not allowed inside of a $match aggregation expression");
+			if (key === "$nearSphere") throw new Error("code 16426; $nearSphere is not allowed inside of a $match aggregation expression");
 			if (query[key] instanceof Object && query[key].constructor === Object) this.uassertNoDisallowedClauses(query[key]);
 		}
 	}
@@ -103,6 +114,25 @@ klass.createFromJson = function createFromJson(jsonElement, ctx) {
 	return matcher;
 };
 
+proto.isTextQuery = function isTextQuery() {
+    return this._isTextQuery;
+};
+
+klass.isTextQuery = function isTextQuery(query) {
+    for (var key in query) {
+        var fieldName = key;
+        if (fieldName === "$text") return true;
+        if (query[key] instanceof Object && query[key].constructor === Object && this.isTextQuery(query[key])) {
+            return true;
+        }
+    }
+    return false;
+};
+
+klass.setSource = function setSource (source) {
+	this.setSource(source);
+};
+
 proto.getQuery = function getQuery() {
 	return this.matcher._pattern;
 };

lib/pipeline/documentSources/OutDocumentSource.js

@@ -10,13 +10,13 @@ var DocumentSource = require('./DocumentSource');
  * @param [ctx] {ExpressionContext}
  **/
 var OutDocumentSource = module.exports = function OutDocumentSource(outputNs, ctx){
-	if (arguments.length > 2) throw new Error("up to two arg expected");
+	if (arguments.length > 2) throw new Error("up to two args expected");
 	base.call(this, ctx);
 	// defaults
 	this._done = false;
 	this._outputNs = outputNs;
 	this._collectionName = "";
-}, klass = OutDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
+}, klass = OutDocumentSource, base = DocumentSource, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
 klass.outName = "$out";
 
@@ -47,6 +47,12 @@ klass.createFromJson = function(jsonElement, ctx) {
 	return out;
 };
 
+// SplittableDocumentSource implementation.
+klass.isSplittableDocumentSource = true;
+
+//NeedsMongodDocumentSource implementation
+klass.needsMongodDocumentSource = true;
+
 proto.getDependencies = function(deps) {
 	deps.needWholeDocument = true;
 	return DocumentSource.GetDepsReturn.EXHAUSTIVE;

lib/pipeline/documentSources/RedactDocumentSource.js

@@ -64,7 +64,7 @@ proto.getNext = function getNext(callback) {
 proto.redactValue = function redactValue(input) {
 	// reorder to make JS happy with types
 	if (input instanceof Array) {
-		var newArr,
+		var newArr = [],
 			arr = input;
 		for (var i = 0; i < arr.length; i++) {
 			if ((arr[i] instanceof Object && arr[i].constructor === Object) || arr[i] instanceof Array) {
@@ -99,7 +99,7 @@ proto.redactObject = function redactObject() {
 		return DocumentSource.EOF;
 	} else if (expressionResult === DESCEND_VAL) {
 		var input = this._variables.getDocument(this._currentId);
-		var out;
+		var out = {};
 
 		var inputKeys = Object.keys(input);
 		for (var i = 0; i < inputKeys.length; i++) {

lib/pipeline/documentSources/SkipDocumentSource.js

@@ -4,64 +4,95 @@ var async = require('async'),
 	DocumentSource = require('./DocumentSource');
 
 /**
- * A document source skipper
+ * A document source skipper.
+ *
  * @class SkipDocumentSource
  * @namespace mungedb-aggregate.pipeline.documentSources
  * @module mungedb-aggregate
  * @constructor
  * @param [ctx] {ExpressionContext}
  **/
-var SkipDocumentSource = module.exports = function SkipDocumentSource(ctx){
-	if (arguments.length > 1) throw new Error("up to one arg expected");
+var SkipDocumentSource = module.exports = function SkipDocumentSource(ctx) {
+	if (arguments.length > 1) {
+		throw new Error('Up to one argument expected.');
+	}
+
 	base.call(this, ctx);
+
 	this.skip = 0;
 	this.count = 0;
-}, klass = SkipDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-klass.skipName = "$skip";
-proto.getSourceName = function getSourceName(){
+	this.needToSkip = true;
+}, klass = SkipDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor: {value: klass}});
+
+klass.skipName = '$skip';
+
+/**
+ * Return the source name.
+ *
+ * @returns {string}
+ */
+proto.getSourceName = function getSourceName() {
 	return klass.skipName;
 };
 
 /**
- * Coalesce skips together
- * @param {Object} nextSource the next source
- * @return {bool} return whether we can coalese together
- **/
+ * Coalesce skips together.
+ *
+ * @param nextSource
+ * @returns {boolean}
+ */
 proto.coalesce = function coalesce(nextSource) {
-	var nextSkip =	nextSource.constructor === SkipDocumentSource?nextSource:null;
+	var nextSkip =	nextSource.constructor === SkipDocumentSource ? nextSource : null;
 
-	// if it's not another $skip, we can't coalesce
-	if (!nextSkip) return false;
+	// If it's not another $skip, we can't coalesce.
+	if (!nextSkip) {
+		return false;
+	}
 
-	// we need to skip over the sum of the two consecutive $skips
+	// We need to skip over the sum of the two consecutive $skips.
 	this.skip += nextSkip.skip;
+
 	return true;
 };
 
+/**
+ * Get next source.
+ *
+ * @param callback
+ * @returns {*}
+ */
 proto.getNext = function getNext(callback) {
-	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
+	if (!callback) {
+		throw new Error(this.getSourceName() + ' #getNext() requires callback.');
+	}
+
+	if (this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt() === false) {
+		return callback(new Error('Interrupted'));
+	}
 
 	var self = this,
 		next;
 
-	if (this.count < this.skip) {
+	if (this.needToSkip) { // May be unnecessary.
+		this.needToSkip = false;
 
 		async.doWhilst(
-			function(cb) {
-				self.source.getNext(function(err, val) {
-					if(err) return cb(err);
-					self.count++;
+			function (cb) {
+				self.source.getNext(function (err, val) {
+					if (err) { return cb(err); }
+
+					++self.count;
 					next = val;
+
 					return cb();
 				});
 			},
 			function() {
 				return self.count < self.skip || next === DocumentSource.EOF;
 			},
-			function(err) {
-				if (err)
-					return callback(err);
+			function (err) {
+				if (err) { return callback(err); }
 			}
 		);
 	}
@@ -69,28 +100,96 @@ proto.getNext = function getNext(callback) {
 	return this.source.getNext(callback);
 };
 
+/**
+ * Serialize the source.
+ *
+ * @param explain
+ * @returns {{}}
+ */
 proto.serialize = function serialize(explain) {
 	var out = {};
+
 	out[this.getSourceName()] = this.skip;
+
 	return out;
 };
 
+/**
+ * Get skip value.
+ *
+ * @returns {number}
+ */
 proto.getSkip = function getSkip() {
 	return this.skip;
 };
 
 /**
- * Creates a new SkipDocumentSource with the input number as the skip
+ * Set skip value.
  *
- * @param {Number} JsonElement this thing is *called* Json, but it expects a number
+ * @param newSkip
+ */
+proto.setSkip = function setSkip(newSkip) {
+	this.skip = newSkip;
+};
+
+/**
+ * Create a new SkipDocumentSource.
+ *
+ * @param expCtx
+ * @returns {SkipDocumentSource}
+ */
+klass.create = function create(expCtx) {
+	return new SkipDocumentSource(expCtx);
+};
+
+/**
+ * Creates a new SkipDocumentSource with the input number as the skip.
+ *
+ * @param {Number} JsonElement this thing is *called* JSON, but it expects a number.
  **/
 klass.createFromJson = function createFromJson(jsonElement, ctx) {
-	if (typeof jsonElement !== "number") throw new Error("code 15972; the value to skip must be a number");
+	if (typeof jsonElement !== 'number') {
+		throw new Error('code 15972; the value to skip must be a number');
+	}
 
 	var nextSkip = new SkipDocumentSource(ctx);
 
 	nextSkip.skip = jsonElement;
-	if (nextSkip.skip < 0 || isNaN(nextSkip.skip)) throw new Error("code 15956; the number to skip cannot be negative");
+
+	if (nextSkip.skip < 0 || isNaN(nextSkip.skip)) {
+		throw new Error('code 15956; the number to skip cannot be negative');
+	}
 
 	return nextSkip;
 };
+
+// SplittableDocumentSource implementation.
+klass.isSplittableDocumentSource = true;
+
+/**
+ * Get dependencies.
+ *
+ * @param deps
+ * @returns {number}
+ */
+proto.getDependencies = function getDependencies(deps) {
+	return DocumentSource.GetDepsReturn.SEE_NEXT;
+};
+
+/**
+ * Get shard source.
+ *
+ * @returns {null}
+ */
+proto.getShardSource = function getShardSource() {
+	return null;
+};
+
+/**
+ * Get router source.
+ *
+ * @returns {SkipDocumentSource}
+ */
+proto.getRouterSource = function getRouterSource() {
+	return this;
+};

lib/pipeline/documentSources/UnwindDocumentSource.js

@@ -1,6 +1,11 @@
 "use strict";
 
-var async = require("async");
+var async = require('async'),
+	DocumentSource = require('./DocumentSource'),
+	Expression = require('../expressions/Expression'),
+	FieldPath = require('../FieldPath'),
+	Value = require('../Value'),
+	Document = require('../Document');
 
 /**
  * A document source unwinder
@@ -11,61 +16,51 @@ var async = require("async");
  * @param [ctx] {ExpressionContext}
  **/
 var UnwindDocumentSource = module.exports = function UnwindDocumentSource(ctx){
-	if (arguments.length > 1) throw new Error("up to one arg expected");
-	base.call(this, ctx);
-
-	// Configuration state.
-	this._unwindPath = null;
+	if (arguments.length > 1) {
+		throw new Error('Up to one argument expected.');
+	}
 
-	// Iteration state.
-	this._unwinder = null;
+	base.call(this, ctx);
 
+	this._unwindPath = null; // Configuration state.
+	this._unwinder = null; // Iteration state.
 }, klass = UnwindDocumentSource, base = require('./DocumentSource'), proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
 
-var DocumentSource = base,
-	FieldPath = require('../FieldPath'),
-	Document = require('../Document'),
-	Expression = require('../expressions/Expression');
+klass.unwindName = '$unwind';
 
-klass.Unwinder = (function(){
+klass.Unwinder = (function() {
 	/**
-	 * Helper class to unwind arrays within a series of documents.
-	 * @param	{String}	unwindPath is the field path to the array to unwind.
-	 **/
-	var klass = function Unwinder(unwindPath){
-		// Path to the array to unwind.
-		this._unwindPath = unwindPath;
-		// The souce document to unwind.
-		this._document = null;
-		// Document indexes of the field path components.
-		this._unwindPathFieldIndexes = [];
-		// Iterator over the array within _document to unwind.
-		this._unwindArrayIterator = null;
-		// The last value returned from _unwindArrayIterator.
-		//this._unwindArrayIteratorCurrent = undefined; //dont define this yet
-	}, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
+	 * Construct a new Unwinder instance. Used as a parent class for UnwindDocumentSource.
+	 *
+	 * @param unwindPath
+	 * @constructor
+	 */
+	var klass = function Unwinder(unwindPath) {
+		this._unwindPath = new FieldPath(unwindPath);
 
-	/**
-	 * Reset the unwinder to unwind a new document.
-	 * @param	{Object}	document
-	 **/
-	proto.resetDocument = function resetDocument(document){
-		if (!document) throw new Error("document is required!");
+		this._inputArray = undefined;
+		this._document = undefined;
+		this._index = undefined;
+	}, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor: {value: klass}});
 
-		// Reset document specific attributes.
+	proto.resetDocument = function resetDocument(document) {
+		if (!document) throw new Error('Document is required!');
+
+		this._inputArray = [];
 		this._document = document;
-		this._unwindPathFieldIndexes.length = 0;
-		this._unwindArrayIterator = null;
-		delete this._unwindArrayIteratorCurrent;
+		this._index = 0;
 
-		var pathValue = this.extractUnwindValue(); // sets _unwindPathFieldIndexes
-		if (!pathValue || pathValue.length === 0) return;  // The path does not exist.
+		var pathValue = Document.getNestedField(this._document, this._unwindPath);
 
-		if (!(pathValue instanceof Array)) throw new Error(UnwindDocumentSource.unwindName + ":  value at end of field path must be an array; code 15978");
+		if (!pathValue || pathValue.length === 0) {
+			return;
+		}
 
-		// Start the iterator used to unwind the array.
-		this._unwindArrayIterator = pathValue.slice(0);
-		this._unwindArrayIteratorCurrent = this._unwindArrayIterator.splice(0,1)[0];
+		if (!(pathValue instanceof Array)) {
+			throw new Error(UnwindDocumentSource.unwindName + ':  value at end of field path must be an array; code 15978');
+		}
+
+		this._inputArray = pathValue;
 	};
 
 	/**
@@ -75,199 +70,144 @@ klass.Unwinder = (function(){
 	 * than the original mongo implementation, but should get updated to follow the current API.
 	 **/
 	proto.getNext = function getNext() {
-		if (this.eof())
+		if (this._inputArray === undefined || this._index === this._inputArray.length) {
 			return DocumentSource.EOF;
-
-		var output = this.getCurrent();
-		this.advance();
-		return output;
-	};
-
-	/**
-	 * eof
-	 * @returns	{Boolean}	true if done unwinding the last document passed to resetDocument().
-	 **/
-	proto.eof = function eof(){
-		return !this.hasOwnProperty("_unwindArrayIteratorCurrent");
-	};
-
-	/**
-	 * Try to advance to the next document unwound from the document passed to resetDocument().
-	 * @returns	{Boolean} true if advanced to a new unwound document, but false if done advancing.
-	 **/
-	proto.advance = function advance(){
-		if (!this._unwindArrayIterator) {
-			// resetDocument() has not been called or the supplied document had no results to
-			// unwind.
-			delete this._unwindArrayIteratorCurrent;
-		} else if (!this._unwindArrayIterator.length) {
-			// There are no more results to unwind.
-			delete this._unwindArrayIteratorCurrent;
-		} else {
-			this._unwindArrayIteratorCurrent = this._unwindArrayIterator.splice(0, 1)[0];
 		}
-	};
 
-	/**
-	 * Get the current document unwound from the document provided to resetDocument(), using
-	 * the current value in the array located at the provided unwindPath.  But return
-	 * intrusive_ptr<Document>() if resetDocument() has not been called or the results to unwind
-	 * have been exhausted.
-	 *
-	 * @returns	{Object}
-	 **/
-	proto.getCurrent = function getCurrent(){
-		if (!this.hasOwnProperty("_unwindArrayIteratorCurrent")) {
-			return null;
-		}
-
-		// Clone all the documents along the field path so that the end values are not shared across
-		// documents that have come out of this pipeline operator.  This is a partial deep clone.
-		// Because the value at the end will be replaced, everything along the path leading to that
-		// will be replaced in order not to share that change with any other clones (or the
-		// original).
-
-		var clone = Document.clone(this._document);
-		var current = clone;
-		var n = this._unwindPathFieldIndexes.length;
-		if (!n) throw new Error("unwindFieldPathIndexes are empty");
-		for (var i = 0; i < n; ++i) {
-			var fi = this._unwindPathFieldIndexes[i];
-			var fp = current[fi];
-			if (i + 1 < n) {
-				// For every object in the path but the last, clone it and continue on down.
-				var next = Document.clone(fp);
-				current[fi] = next;
-				current = next;
-			} else {
-				// In the last nested document, subsitute the current unwound value.
-				current[fi] = this._unwindArrayIteratorCurrent;
-			}
-		}
-
-		return clone;
-	};
-
-	/**
-	 * Get the value at the unwind path, otherwise an empty pointer if no such value
-	 * exists.  The _unwindPathFieldIndexes attribute will be set as the field path is traversed
-	 * to find the value to unwind.
-	 *
-	 * @returns	{Object}
-	 **/
-	proto.extractUnwindValue = function extractUnwindValue() {
-		var current = this._document;
-		var pathValue;
-		var pathLength = this._unwindPath.getPathLength();
-		for (var i = 0; i < pathLength; ++i) {
-
-			var idx = this._unwindPath.getFieldName(i);
-
-			if (!current.hasOwnProperty(idx)) return null; // The target field is missing.
-
-			// Record the indexes of the fields down the field path in order to quickly replace them
-			// as the documents along the field path are cloned.
-			this._unwindPathFieldIndexes.push(idx);
-
-			pathValue = current[idx];
+		this._document = Document.cloneDeep(this._document);
+		Document.setNestedField(this._document, this._unwindPath, this._inputArray[this._index++]);
 
-			if (i < pathLength - 1) {
-				if (typeof pathValue !== 'object') return null; // The next field in the path cannot exist (inside a non object).
-				current = pathValue; // Move down the object tree.
-			}
-		}
-
-		return pathValue;
+		return this._document;
 	};
 
 	return klass;
 })();
 
 /**
- * Specify the field to unwind.
-**/
-proto.unwindPath = function unwindPath(fieldPath){
-	// Can't set more than one unwind path.
-	if (this._unwindPath) throw new Error(this.getSourceName() + " can't unwind more than one path; code 15979");
-
-	// Record the unwind path.
-	this._unwindPath = new FieldPath(fieldPath);
-	this._unwinder = new klass.Unwinder(this._unwindPath);
-};
-
-klass.unwindName = "$unwind";
-
-proto.getSourceName = function getSourceName(){
+ * Get the document source name.
+ *
+ * @returns {string}
+ */
+proto.getSourceName = function getSourceName() {
 	return klass.unwindName;
 };
 
 /**
- * Get the fields this operation needs to do its job.
- * Deps should be in "a.b.c" notation
+ * Get the next source.
  *
- * @method	getDependencies
- * @param	{Object} deps	set (unique array) of strings
- * @returns	DocumentSource.GetDepsReturn
-**/
-proto.getDependencies = function getDependencies(deps) {
-	if (!this._unwindPath) throw new Error("unwind path does not exist!");
-	deps[this._unwindPath.getPath(false)] = 1;
-	return DocumentSource.GetDepsReturn.SEE_NEXT;
-};
-
+ * @param callback
+ * @returns {*}
+ */
 proto.getNext = function getNext(callback) {
-	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
+	if (!callback) {
+		throw new Error(this.getSourceName() + ' #getNext() requires callback.');
+	}
+
+	if (this.expCtx.checkForInterrupt && this.expCtx.checkForInterrupt() === false) {
+		return callback(new Error('Interrupted'));
+	}
 
 	var self = this,
 		out = this._unwinder.getNext(),
 		exhausted = false;
 
 	async.until(
-		function() {
-			if(out === DocumentSource.EOF && exhausted) return true;	// Really is EOF, not just an empty unwinder
-			else if(out !== DocumentSource.EOF) return true; // Return whatever we got that wasn't EOF
+		function () {
+			if (out !== DocumentSource.EOF || exhausted) {
+				return true;
+			}
+
 			return false;
 		},
-		function(cb) {
-			self.source.getNext(function(err, doc) {
-				if(err) return cb(err);
-				out = doc;
-				if(out === DocumentSource.EOF) { // Our source is out of documents, we're done
+		function (cb) {
+			self.source.getNext(function (err, doc) {
+				if (err) {
+					return cb(err);
+				}
+
+				if (doc === DocumentSource.EOF) {
 					exhausted = true;
-					return cb();
 				} else {
 					self._unwinder.resetDocument(doc);
 					out = self._unwinder.getNext();
-					return cb();
 				}
+
+				return cb();
 			});
 		},
 		function(err) {
-			if(err) return callback(err);
+			if (err) {
+				return callback(err);
+			}
+
 			return callback(null, out);
 		}
 	);
 
-	return out; //For sync mode
+	return out;
 };
 
+/**
+ * Serialize the data.
+ *
+ * @param explain
+ * @returns {{}}
+ */
 proto.serialize = function serialize(explain) {
-	if (!this._unwindPath) throw new Error("unwind path does not exist!");
+	if (!this._unwindPath) {
+		throw new Error('unwind path does not exist!');
+	}
+
 	var doc = {};
+
 	doc[this.getSourceName()] = this._unwindPath.getPath(true);
+
 	return doc;
 };
 
+/**
+ * Get the fields this operation needs to do its job.
+ *
+ * @param deps
+ * @returns {DocumentSource.GetDepsReturn.SEE_NEXT|*}
+ */
+proto.getDependencies = function getDependencies(deps) {
+	if (!this._unwindPath) {
+		throw new Error('unwind path does not exist!');
+	}
+
+	deps[this._unwindPath.getPath(false)] = 1;
+
+	return DocumentSource.GetDepsReturn.SEE_NEXT;
+};
+
+/**
+ * Unwind path.
+ *
+ * @param fieldPath
+ */
+proto.unwindPath = function unwindPath(fieldPath) {
+	if (this._unwindPath) {
+		throw new Error(this.getSourceName() + ' can\'t unwind more than one path; code 15979');
+	}
+
+	// Record the unwind path.
+	this._unwindPath = new FieldPath(fieldPath);
+	this._unwinder = new klass.Unwinder(fieldPath);
+};
+
 /**
  * Creates a new UnwindDocumentSource with the input path as the path to unwind
  * @param {String} JsonElement this thing is *called* Json, but it expects a string
 **/
 klass.createFromJson = function createFromJson(jsonElement, ctx) {
-	// The value of $unwind should just be a field path.
-	if (jsonElement.constructor !== String) throw new Error("the " + klass.unwindName + " field path must be specified as a string; code 15981");
+	if (jsonElement.constructor !== String) {
+		throw new Error('the ' + klass.unwindName + ' field path must be specified as a string; code 15981');
+	}
+
+	var pathString = Expression.removeFieldPrefix(jsonElement),
+		unwind = new UnwindDocumentSource(ctx);
 
-	var pathString = Expression.removeFieldPrefix(jsonElement);
-	var unwind = new UnwindDocumentSource(ctx);
 	unwind.unwindPath(pathString);
 
 	return unwind;

lib/pipeline/expressions/index.js

@@ -24,6 +24,8 @@ module.exports = {
 	ObjectExpression: require("./ObjectExpression.js"),
 	OrExpression: require("./OrExpression.js"),
 	SecondExpression: require("./SecondExpression.js"),
+	SetIntersectionExpression: require("./SetIntersectionExpression.js"),
+	SizeExpression: require("./SizeExpression.js"),
 	StrcasecmpExpression: require("./StrcasecmpExpression.js"),
 	SubstrExpression: require("./SubstrExpression.js"),
 	SubtractExpression: require("./SubtractExpression.js"),

test/lib/pipeline/documentSources/DocumentSource.js

@@ -7,6 +7,8 @@ module.exports = {
 
 	"DocumentSource": {
 
+		"should be tested via subclasses": function() {}
+
 	}
 
 };

test/lib/pipeline/documentSources/LimitDocumentSource.js

@@ -10,74 +10,80 @@ module.exports = {
 
 		"constructor()": {
 
-			"should not throw Error when constructing without args": function testConstructor(){
+			"should not throw Error when constructing without args": function testConstructor(next){
 				assert.doesNotThrow(function(){
 					new LimitDocumentSource();
+					return next();
 				});
 			}
-
 		},
 
+ 		/** A limit does not introduce any dependencies. */
 		"#getDependencies": {
-			"limits do not create dependencies": function() {
-				var lds = LimitDocumentSource.createFromJson(1),
+			"limits do not create dependencies": function(next) {
+				var lds = LimitDocumentSource.createFromJson(1, null),
 					deps = {};
 
 				assert.equal(DocumentSource.GetDepsReturn.SEE_NEXT, lds.getDependencies(deps));
 				assert.equal(0, Object.keys(deps).length);
+				return next();
 			}
 		},
 
 		"#getSourceName()": {
 
-			"should return the correct source name; $limit": function testSourceName(){
+			"should return the correct source name; $limit": function testSourceName(next){
 				var lds = new LimitDocumentSource();
 				assert.strictEqual(lds.getSourceName(), "$limit");
+				return next();
 			}
-
 		},
 
 		"#getFactory()": {
 
-			"should return the constructor for this class": function factoryIsConstructor(){
+			"should return the constructor for this class": function factoryIsConstructor(next){
 				assert.strictEqual(new LimitDocumentSource().getFactory(), LimitDocumentSource);
+				return next();
 			}
-
 		},
 
 		"#coalesce()": {
 
-			"should return false if nextSource is not $limit": function dontSkip(){
+			"should return false if nextSource is not $limit": function dontSkip(next){
 				var lds = new LimitDocumentSource();
 				assert.equal(lds.coalesce({}), false);
+				return next();
 			},
-			"should return true if nextSource is $limit": function changeLimit(){
+			"should return true if nextSource is $limit": function changeLimit(next){
 				var lds = new LimitDocumentSource();
 				assert.equal(lds.coalesce(new LimitDocumentSource()), true);
+				return next();
 			}
-
 		},
 
 		"#getNext()": {
 
-			"should throw an error if no callback is given": function() {
+			"should throw an error if no callback is given": function(next) {
 				var lds = new LimitDocumentSource();
 				assert.throws(lds.getNext.bind(lds));
+				return next();
 			},
 
+			/** Exhausting a DocumentSourceLimit disposes of the limit's source. */
 			"should return the current document source": function currSource(next){
-				var lds = new LimitDocumentSource();
+				var lds = new LimitDocumentSource({"$limit":[{"a":1},{"a":2}]});
 				lds.limit = 1;
 				lds.source = {getNext:function(cb){cb(null,{ item:1 });}};
 				lds.getNext(function(err,val) {
 					assert.deepEqual(val, { item:1 });
-					next();
+					return next();
 				});
 			},
 
+			/** Exhausting a DocumentSourceLimit disposes of the pipeline's DocumentSourceCursor. */
 			"should return EOF for no sources remaining": function noMoar(next){
-				var lds = new LimitDocumentSource();
-				lds.limit = 10;
+				var lds = new LimitDocumentSource({"$match":[{"a":1},{"a":1}]});
+				lds.limit = 1;
 				lds.source = {
 					calls: 0,
 					getNext:function(cb) {
@@ -91,7 +97,7 @@ module.exports = {
 				lds.getNext(function(){});
 				lds.getNext(function(err,val) {
 					assert.strictEqual(val, DocumentSource.EOF);
-					next();
+					return next();
 				});
 			},
 
@@ -110,36 +116,32 @@ module.exports = {
 				lds.getNext(function(){});
 				lds.getNext(function (err,val) {
 					assert.strictEqual(val, DocumentSource.EOF);
-					next();
+					return next();
 				});
 			}
-
 		},
 
 		"#serialize()": {
 
-			"should create an object with a key $limit and the value equal to the limit": function sourceToJsonTest(){
+			"should create an object with a key $limit and the value equal to the limit": function sourceToJsonTest(next){
 				var lds = new LimitDocumentSource();
 				lds.limit = 9;
 				var actual = lds.serialize(false);
 				assert.deepEqual(actual, { "$limit": 9 });
+				return next();
 			}
-
 		},
 
 		"#createFromJson()": {
 
-			"should return a new LimitDocumentSource object from an input number": function createTest(){
+			"should return a new LimitDocumentSource object from an input number": function createTest(next){
 				var t = LimitDocumentSource.createFromJson(5);
 				assert.strictEqual(t.constructor, LimitDocumentSource);
 				assert.strictEqual(t.limit, 5);
+				return next();
 			}
-
 		}
-
-
 	}
-
 };
 
 if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run(process.exit);

test/lib/pipeline/documentSources/MatchDocumentSource.js

@@ -20,6 +20,12 @@ module.exports = {
 				assert.throws(function(){
 					new MatchDocumentSource();
 				});
+			},
+
+			"should throw Error when trying to using a $text operator": function testTextOp () {
+				assert.throws(function(){
+					new MatchDocumentSource({packet:{ $text:"thisIsntImplemented" } });
+				});
 			}
 
 		},
@@ -353,6 +359,87 @@ module.exports = {
 					{});
 			}
 
+		},
+
+		"#isTextQuery()": {
+
+			"should return true when $text operator is first stage in pipeline": function () {
+				var query = {$text:'textQuery'};
+				assert.ok(MatchDocumentSource.isTextQuery(query)); // true
+			},
+
+			"should return true when $text operator is nested in the pipeline": function () {
+				var query = {$stage:{$text:'textQuery'}};
+				assert.ok(MatchDocumentSource.isTextQuery(query)); // true
+			},
+
+			"should return false when $text operator is not in pipeline": function () {
+				var query = {$notText:'textQuery'};
+				assert.ok(!MatchDocumentSource.isTextQuery(query)); // false
+			}
+
+		},
+
+		"#uassertNoDisallowedClauses()": {
+
+			"should throw if invalid stage is in match expression": function () {
+				var whereQuery = {$where:'where'};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(whereQuery);
+				});
+
+				var nearQuery = {$near:'near'};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(nearQuery);
+				});
+
+				var withinQuery = {$within:'within'};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(withinQuery);
+				});
+
+				var nearSphereQuery = {$nearSphere:'nearSphere'};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(nearSphereQuery);
+				});
+			},
+
+			"should throw if invalid stage is nested in the match expression": function () {
+				var whereQuery = {$validStage:{$where:'where'}};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(whereQuery);
+				});
+
+				var nearQuery = {$validStage:{$near:'near'}};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(nearQuery);
+				});
+
+				var withinQuery = {$validStage:{$within:'within'}};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(withinQuery);
+				});
+
+				var nearSphereQuery = {$validStage:{$nearSphere:'nearSphere'}};
+				assert.throws(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(nearSphereQuery);
+				});
+			},
+
+			"should not throw if invalid stage is not in match expression": function () {
+				var query = {$valid:'valid'};
+				assert.doesNotThrow(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(query);
+				});
+			},
+
+			"should not throw if invalid stage is not nested in the match expression": function () {
+				var query = {$valid:{$anotherValid:'valid'}};
+				assert.doesNotThrow(function(){
+					MatchDocumentSource.uassertNoDisallowedClauses(query);
+				});
+			},
+
 		}
 
 	}

test/lib/pipeline/documentSources/OutDocumentSource.js

@@ -41,7 +41,7 @@ module.exports = {
 				assert.throws(ods.getNext.bind(ods));
 			},
 
-			"should act ass passthrough (for now)": function(next) {
+			"should act as passthrough (for now)": function(next) {
 				var ods = OutDocumentSource.createFromJson("test"),
 					cwc = new CursorDocumentSource.CursorWithContext(),
 					l = [{_id:0,a:[{b:1},{b:2}]}, {_id:1,a:[{b:1},{b:1}]} ];

test/lib/pipeline/documentSources/RedactDocumentSource.js

@@ -4,23 +4,28 @@ var assert = require("assert"),
 	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
 	RedactDocumentSource = require("../../../../lib/pipeline/documentSources/RedactDocumentSource"),
 	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
-	Cursor = require("../../../../lib/Cursor");
-
-var exampleRedact = {$cond: [
-	{$gt:[3, 0]},
-	"$$DESCEND",
-	"$$PRUNE"]
+	Cursor = require("../../../../lib/Cursor"),
+	Expressions = require("../../../../lib/pipeline/expressions");
+
+var exampleRedact = {$cond:{
+	if:{$gt:[0,4]},
+	then:"$$DESCEND",
+	else:"$$PRUNE"
+}};
+
+var createCursorDocumentSource = function createCursorDocumentSource (input) {
+	if (!input || input.constructor !== Array) throw new Error('invalid');
+	var cwc = new CursorDocumentSource.CursorWithContext();
+	cwc._cursor = new Cursor(input);
+	return new CursorDocumentSource(cwc);
 };
 
-////////////////////////////////////////////////////////////////////////////////
-////////////////////////////////////////////////////////////////////////////////
-//////////////////////////////////// BUSTED ////////////////////////////////////
-//           This DocumentSource is busted without new Expressions            //
-////////////////////////////////////////////////////////////////////////////////
-////////////////////////////////////////////////////////////////////////////////
-////////////////////////////////////////////////////////////////////////////////
+var createRedactDocumentSource = function createRedactDocumentSource (src, expression) {
+	var rds = RedactDocumentSource.createFromJson(expression);
+	rds.setSource(src);
+	return rds;
+};
 
-//TESTS
 module.exports = {
 
 	"RedactDocumentSource": {
@@ -83,6 +88,7 @@ module.exports = {
 				var rds = new RedactDocumentSource();
 				assert.throws(rds.getNext.bind(rds));
 			},
+
 		},
 
 		"#optimize()": {
@@ -109,9 +115,124 @@ module.exports = {
 			}
 
 		},
+
+		"#redact()": {
+
+			"should redact subsection where tag does not match": function (done) {
+				var cds = createCursorDocumentSource([{
+					_id: 1,
+					title: "123 Department Report",
+					tags: ["G", "STLW"],
+					year: 2014,
+					subsections: [
+						{
+							subtitle: "Section 1: Overview",
+							tags: ["SI", "G"],
+							content: "Section 1: This is the content of section 1."
+						},
+						{
+							subtitle: "Section 2: Analysis",
+							tags: ["STLW"],
+							content: "Section 2: This is the content of section 2."
+						},
+						{
+							subtitle: "Section 3: Budgeting",
+							tags: ["TK"],
+							content: {
+								text: "Section 3: This is the content of section3.",
+								tags: ["HCS"]
+							}
+						}
+					]
+				}]);
+
+				var expression = {$cond:{
+					if:{$gt: [{$size: {$setIntersection: ["$tags", [ "STLW", "G" ]]}},0]},
+					then:"$$DESCEND",
+					else:"$$PRUNE"
+				}};
+
+				var rds = createRedactDocumentSource(cds, expression);
+
+				var result = {
+					"_id": 1,
+					"title": "123 Department Report",
+					"tags": ["G", "STLW"],
+					"year": 2014,
+					"subsections": [{
+						"subtitle": "Section 1: Overview",
+						"tags": ["SI", "G"],
+						"content": "Section 1: This is the content of section 1."
+					}, {
+						"subtitle": "Section 2: Analysis",
+						"tags": ["STLW"],
+						"content": "Section 2: This is the content of section 2."
+					}]
+				};
+
+				rds.getNext(function (err, actual) {
+					assert.deepEqual(actual, result);
+					done();
+				});
+
+			},
+
+			"should redact an entire subsection based on a defined access level": function (done) {
+				var cds = createCursorDocumentSource([{
+					_id: 1,
+					level: 1,
+					acct_id: "xyz123",
+					cc: {
+						level: 5,
+						type: "yy",
+						exp_date: new Date("2015-11-01"),
+						billing_addr: {
+							level: 5,
+							addr1: "123 ABC Street",
+							city: "Some City"
+						},
+						shipping_addr: [
+							{
+								level: 3,
+								addr1: "987 XYZ Ave",
+								city: "Some City"
+							},
+							{
+								level: 3,
+								addr1: "PO Box 0123",
+								city: "Some City"
+							}
+						]
+					},
+					status: "A"
+				}]);
+
+				var expression = {$cond:{
+					if:{$eq:["$level",5]},
+					then:"$$PRUNE",
+					else:"$$DESCEND"
+				}};
+
+				var rds = createRedactDocumentSource(cds, expression);
+
+				var result = {
+					_id:1,
+					level:1,
+					acct_id:"xyz123",
+					status:"A"
+				};
+
+				rds.getNext(function (err, actual) {
+					assert.deepEqual(actual, result);
+					done();
+				});
+
+			}
+
+		}
+
 	}
 
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);
-
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);

test/lib/pipeline/documentSources/SkipDocumentSource.js

@@ -21,6 +21,15 @@ module.exports = {
 
 		},
 
+		'#create()': {
+			'should create a direct copy of a SkipDocumentSource created through the constructor': function () {
+				var sds1 = new SkipDocumentSource(),
+					sds2 = SkipDocumentSource.create();
+
+				assert.strictEqual(JSON.stringify(sds1), JSON.stringify(sds2));
+			}
+		},
+
 		"#getSourceName()": {
 
 			"should return the correct source name; $skip": function testSourceName(){
@@ -30,6 +39,24 @@ module.exports = {
 
 		},
 
+		'#getSkip()': {
+			'should return the skips': function () {
+				var sds = new SkipDocumentSource();
+
+				assert.strictEqual(sds.getSkip(), 0);
+			}
+		},
+
+		'#setSkip()': {
+			'should return the skips': function () {
+				var sds = new SkipDocumentSource();
+
+				sds.setSkip(10);
+
+				assert.strictEqual(sds.getSkip(), 10);
+			}
+		},
+
 		"#coalesce()": {
 
 			"should return false if nextSource is not $skip": function dontSkip(){
@@ -164,9 +191,31 @@ module.exports = {
 				assert.strictEqual(t.skip, 5);
 			}
 
-		}
+		},
+
+		'#getDependencies()': {
+			'should return 1 (GET_NEXT)': function () {
+				var sds = new SkipDocumentSource();
 
+				assert.strictEqual(sds.getDependencies(), DocumentSource.GetDepsReturn.SEE_NEXT); // Hackish. We may be getting an enum in somewhere.
+			}
+		},
+
+		'#getShardSource()': {
+			'should return the instance of the SkipDocumentSource': function () {
+				var sds = new SkipDocumentSource();
 
+				assert.strictEqual(sds.getShardSource(), null);
+			}
+		},
+
+		'#getRouterSource()': {
+			'should return null': function () {
+				var sds = new SkipDocumentSource();
+
+				assert.strictEqual(sds.getRouterSource(), sds);
+			}
+		}
 	}
 
 };

test/lib/pipeline/expressions/AddExpression_test.js

@@ -238,7 +238,6 @@ exports.AddExpression = {
 			assert.strictEqual(expr.operands.length, 2, "should optimize operands away");
 			assert(expr.operands[0] instanceof FieldPathExpression);
 			assert(expr.operands[1] instanceof ConstantExpression);
-			debugger
 			assert.strictEqual(expr.operands[1].evaluate(), 1 + 2 + 3 + 4 + 5 + 6);
 		},
 

test/lib/pipeline/expressions/ConcatExpression_test.js

@@ -80,7 +80,7 @@ exports.ConcatExpression = {
 		},
 
 		"should throw if an operand is a boolean": function() {
-			var expr = Expression.parseOperand({$concat:["my","$a"]}, this.vps)
+			var expr = Expression.parseOperand({$concat:["my","$a"]}, this.vps);
 			assert.throws(function() {
 				expr.evaluate({a:true});
 			});