refs #5134: New DocumentSources async

lib/pipeline/documentSources/DocumentSource.js

@@ -254,6 +254,27 @@ klass.parseDeps = function parseDeps(deps) {
 	return md;
 };
 
+/**
+ * A function compatible as a getNext for document sources.
+ * Does nothing except pass the documents through. To use,
+ * Attach this function on a DocumentSource prototype.
+ *
+ * @method GET_NEXT_PASS_THROUGH
+ * @param callback {Function}
+ * @param callback.err {Error} An error or falsey
+ * @param callback.doc {Object} The source's next object or DocumentSource.EOF
+ **/
+klass.GET_NEXT_PASS_THROUGH = function GET_NEXT_PASS_THROUGH(callback) {
+	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
+
+	var out;
+	this.source.getNext(function(err, doc) {
+		out = doc;
+		return callback(err, doc);
+	});
+	return out; // For the sync people in da house
+};
+
 klass.documentFromJsonWithDeps = function documentFromJsonWithDeps(bson, neededFields) {
 	var arrayHelper = function(bson, neededFields) {
 		var values = [];

lib/pipeline/documentSources/GeoNearDocumentSource.js

@@ -29,11 +29,7 @@ proto.getSourceName = function() {
 	return klass.geoNearName;
 };
 
-proto.getNext = function(callback) {
-	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
-
-	return this.source.getNext(callback);
-};
+proto.getNext = DocumentSource.GET_NEXT_PASS_THROUGH;
 
 proto.setSource = function(docSource) {
 	throw new Error('code 16602; $geoNear is only allowed as the first pipeline stage');

lib/pipeline/documentSources/OutDocumentSource.js

@@ -24,11 +24,7 @@ proto.getSourceName = function() {
 	return klass.outName;
 };
 
-proto.getNext = function(callback) {
-	if (!callback) throw new Error(this.getSourceName() + ' #getNext() requires callback');
-
-	return this.source.getNext(callback);
-};
+proto.getNext = DocumentSource.GET_NEXT_PASS_THROUGH;
 
 proto.serialize = function(explain) {
 	var doc = {},

lib/pipeline/documentSources/RedactDocumentSource.js

@@ -48,7 +48,7 @@ proto.getNext = function getNext(callback) {
 				self._variables.setValue(self._currentId, input);
 				var result = self.redactObject();
 				if (result !== DocumentSource.EOF)
-					return cb(result);
+					return cb(result); //Using the err argument to pass the result document; let's break out without having EOF
 				return cb();
 			});
 		},
@@ -58,6 +58,7 @@ proto.getNext = function getNext(callback) {
 			return callback(null, DocumentSource.EOF);
 		}
 	);
+	return doc;
 };
 
 proto.redactValue = function redactValue(input) {