Refs #2138 Fixing a bug that was implemented when we made document sources work correctly

lib/pipeline/Pipeline.js

@@ -19,7 +19,8 @@ var Pipeline = module.exports = (function(){
 		SkipDocumentSource = require('./documentSources/SkipDocumentSource'),
 		UnwindDocumentSource = require('./documentSources/UnwindDocumentSource'),
 		GroupDocumentSource = require('./documentSources/GroupDocumentSource'),
-		SortDocumentSource = require('./documentSources/SortDocumentSource');
+		SortDocumentSource = require('./documentSources/SortDocumentSource'),
+		DocumentSource = require("./documentSources/DocumentSource");
 	
 	klass.StageDesc = {};//attaching this to the class for test cases
 	klass.StageDesc[LimitDocumentSource.limitName] = LimitDocumentSource.createFromJson;
@@ -142,7 +143,6 @@ var Pipeline = module.exports = (function(){
 			iter.optimize();
 		}
 
-
 		return pipelineInstance;
 	};
 
@@ -156,11 +156,15 @@ var Pipeline = module.exports = (function(){
 	proto.run = function run(source, callback){
 		if(!callback)
 			throw new Error("run requires callback");
+
+		if(source && !(source instanceof DocumentSource))
+			throw new Error("source must be an instance of DocumentSource");
+		
+		
 		var self = this;
 		source.setSource(undefined, function(){	//TODO: HACK: temp solution to the fact that we need to initialize our source since we're using setSource as a workaround for the lack of real async cursors
 			async.eachSeries(self.sourceVector,
 				function(item, next){
-					console.debug("Set source on %s", source.constructor.name);
 					item.setSource(source, function(err){
 						if(err) return next(err);
 						source = item;

lib/pipeline/documentSources/CursorDocumentSource.js

@@ -179,8 +179,11 @@ var CursorDocumentSource = module.exports = (function(){
      * @method	setSource
      * @param	{DocumentSource}	pSource	the underlying source to use
     **/
-    proto.setSource = function setSource(pTheSource) {
-		throw new Error("CursorDocumentSource doesn't take a source");
+    proto.setSource = function setSource(pTheSource, callback) {
+        //TODO: This needs to put back at some point. Once async is properly supported
+        //throw new Error("CursorDocumentSource doesn't take a source");
+        if (callback)
+            return process.nextTick(callback);
     };
     /**
      * Create an object that represents the document source.  The object

test/lib/aggregate.js

@@ -348,4 +348,4 @@ module.exports = {
 
 };
 
-if(!module.parent) (new (require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run();
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);

test/lib/pipeline/Pipeline.js

@@ -1,6 +1,7 @@
 "use strict";
 var assert = require("assert"),
-	Pipeline = require("../../../lib/pipeline/Pipeline");
+	Pipeline = require("../../../lib/pipeline/Pipeline"),
+	DocumentSource = require('../../../lib/pipeline/documentSources/DocumentSource');
 
 
 module.exports = {
@@ -102,7 +103,7 @@ module.exports = {
 		"#run": {
 			"should set the parent source for all sources in the pipeline except the first one":function(next){
 				var p = Pipeline.parseCommand([{$test:{coalesce:false}}, {$test:{coalesce:false}}, {$test:{coalesce:false}}]);
-				p.run({}, function(err, result){
+				p.run(new DocumentSource(), function(err, result){
 					assert.equal(p.sourceVector[1].pSource, p.sourceVector[0]);
 					assert.equal(p.sourceVector[2].pSource, p.sourceVector[1]);
 					next();
@@ -112,7 +113,7 @@ module.exports = {
 			"should iterate through sources and return resultant array":function(next){
 				var p = Pipeline.parseCommand([{$test:{coalesce:false}}, {$test:{coalesce:false}}, {$test:{coalesce:false}}]),
 					result = {};
-				p.run(result, function(err, result){
+				p.run(new DocumentSource(), function(err, result){
 
 					assert.deepEqual(result.result, [5,4,3,2,1,0]);//see the test source for why this should be so
 					next();
@@ -125,4 +126,4 @@ module.exports = {
 
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run();
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);

test/lib/pipeline/PipelineD.js

@@ -119,7 +119,7 @@ module.exports = {
 				];
 				var p = Pipeline.parseCommand(cmdObj),
 					cs = PipelineD.prepareCursorSource(p, [1,2,3,4,5]);
-					assert.equal(JSON.stringify(cs._projection), JSON.stringify({ 'a.b.c': 1, d: 1, 'e.f.g': 1, _id: 1 }));
+					assert.equal(JSON.stringify(cs._projection), JSON.stringify({_id: 1, 'a.b.c': 1, d: 1, 'e.f.g': 1}));
 			},
 
 			"should get group's deps": function(){
@@ -149,4 +149,4 @@ module.exports = {
 
 };
 
-if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).run();
+if (!module.parent)(new(require("mocha"))()).ui("exports").reporter("spec").addFile(__filename).grep(process.env.MOCHA_GREP || '').run(process.exit);