Refs #2378: fix err via callback on run; new test

lib/pipeline/Pipeline.js

@@ -205,13 +205,17 @@ var Pipeline = module.exports = (function(){
 					*/
 					// the array in which the aggregation results reside
 					var resultArray = [];
-					for(var hasDoc = !inputSource.eof(); hasDoc; hasDoc = inputSource.advance()) {
-						var document = inputSource.getCurrent();
-						resultArray.push(document);	// add the document to the result set
-
-						//Commenting out this assertion for munge.  MUHAHAHA!!!
-						// object will be too large, assert. the extra 1KB is for headers
-						//if(resultArray.len() < BSONObjMaxUserSize - 1024) throw new Error("aggregation result exceeds maximum document size (" + BSONObjMaxUserSize / (1024 * 1024) + "MB); code 16389");
+					try{
+						for(var hasDoc = !inputSource.eof(); hasDoc; hasDoc = inputSource.advance()) {
+							var document = inputSource.getCurrent();
+							resultArray.push(document);	// add the document to the result set
+
+							//Commenting out this assertion for munge.  MUHAHAHA!!!
+							// object will be too large, assert. the extra 1KB is for headers
+							//if(resultArray.len() < BSONObjMaxUserSize - 1024) throw new Error("aggregation result exceeds maximum document size (" + BSONObjMaxUserSize / (1024 * 1024) + "MB); code 16389");
+						}
+					} catch (err) {
+						return callback(err);
 					}
 					var result = {
 						result: resultArray

test/lib/pipeline/Pipeline.js

@@ -119,6 +119,14 @@ module.exports = {
 					assert.deepEqual(result.result, [5,4,3,2,1,0]);//see the test source for why this should be so
 					next();
 				});
+			},
+
+			"should call callback with errors from pipeline components": function (next) {
+				var p = Pipeline.parseCommand({pipeline:[{$match:{$foo:{bar:"baz"}}}]});
+				p.run(new DocumentSource({}), function(err, results){
+					assert(err instanceof Error);
+					return next();
+				});
 			}
 
 		}