Refs #5128: Update SkipDocumentSource

lib/pipeline/documentSources/SkipDocumentSource.js

@@ -1,5 +1,8 @@
 "use strict";
 
+var async = require('async'),
+	DocumentSource = require('./DocumentSource');
+
 /**
  * A document source skipper
  * @class SkipDocumentSource
@@ -36,72 +39,37 @@ proto.coalesce = function coalesce(nextSource) {
 	return true;
 };
 
-proto.skipper = function skipper() {
-	if (this.count === 0) {
-		while (!this.source.eof() && this.count++ < this.skip) {
-			this.source.advance();
-		}
-	}
-
-	if (this.source.eof()) {
-		this.current = null;
-		return;
-	}
-
-	this.current = this.source.getCurrent();
-};
-
-
-/**
- * Is the source at EOF?
- * @method	eof
- **/
-proto.eof = function eof() {
-	this.skipper();
-	return this.source.eof();
-};
-
-/**
- * some implementations do the equivalent of verify(!eof()) so check eof() first
- * @method	getCurrent
- * @returns	{Document}	the current Document without advancing
- **/
-proto.getCurrent = function getCurrent() {
-	this.skipper();
-	return this.source.getCurrent();
-};
-
-/**
- * Advance the state of the DocumentSource so that it will return the next Document.
- * The default implementation returns false, after checking for interrupts.
- * Derived classes can call the default implementation in their own implementations in order to check for interrupts.
- *
- * @method	advance
- * @returns	{Boolean}	whether there is another document to fetch, i.e., whether or not getCurrent() will succeed.  This default implementation always returns false.
- **/
-proto.advance = function advance() {
-	base.prototype.advance.call(this); // check for interrupts
-	if (this.eof()) {
-		this.current = null;
-		return false;
+proto.getNext = function getNext(callback) {
+	var self = this;
+
+	if (this.count < this.skip) {
+
+		async.doWhilst(
+			function(cb) {
+				self.source.getNext(function(err, val) {
+					self.count++;
+					if (val === DocumentSource.EOF)
+						return cb(val);
+					return cb();
+				});
+			},
+			function() {
+				return self.count < self.skip;
+			},
+			function(err) {
+				if (err)
+					return callback(err);
+			}
+		);
 	}
 
-	this.current = this.source.getCurrent();
-	return this.source.advance();
+	return this.source.getNext(callback);
 };
 
-/**
- * Create an object that represents the document source.  The object
- * will have a single field whose name is the source's name.  This
- * will be used by the default implementation of addToJsonArray()
- * to add this object to a pipeline being represented in JSON.
- *
- * @method	sourceToJson
- * @param	{Object} builder	JSONObjBuilder: a blank object builder to write to
- * @param	{Boolean}	explain	create explain output
- **/
-proto.sourceToJson = function sourceToJson(builder, explain) {
-	builder.$skip = this.skip;
+proto.serialize = function(explain) {
+	var out = {};
+	out[this.getSourceName()] = this.skip;
+	return out;
 };
 
 /**

test/lib/pipeline/documentSources/SkipDocumentSource.js

@@ -1,5 +1,9 @@
 "use strict";
 var assert = require("assert"),
+	async = require("async"),
+	Cursor = require("../../../../lib/Cursor"),
+	DocumentSource = require("../../../../lib/pipeline/documentSources/DocumentSource"),
+	CursorDocumentSource = require("../../../../lib/pipeline/documentSources/CursorDocumentSource"),
 	SkipDocumentSource = require("../../../../lib/pipeline/documentSources/SkipDocumentSource");
 
 
@@ -39,103 +43,109 @@ module.exports = {
 
 		},
 
-		"#eof()": {
+		"#getNext()": {
 
-			"should return true if there are no more sources": function noSources(){
+			"should return EOF if there are no more sources": function noSources(next){
 				var lds = new SkipDocumentSource();
-				lds.skip = 9;
+				lds.skip = 3;
 				lds.count = 0;
-				lds.source = {
-					eof: function(){
-						return true;
+
+				var expected = [
+					{val:4},
+					DocumentSource.EOF
+				];
+
+                var cwc = new CursorDocumentSource.CursorWithContext();
+                var input = [
+					{val:1},
+					{val:2},
+					{val:3},
+					{val:4},
+				];
+                cwc._cursor = new Cursor( input );
+                var cds = new CursorDocumentSource(cwc);
+				lds.setSource(cds);
+
+				async.series([
+						lds.getNext.bind(lds),
+						lds.getNext.bind(lds),
+					],
+					function(err,res) {
+						assert.deepEqual(expected, res);
+						next();
 					}
-				};
-				assert.equal(lds.eof(), true);
+				);
+				lds.getNext(function(err, actual) {
+					assert.equal(actual, DocumentSource.EOF);
+				});
+			},
+			"should return documents if skip count is not hit and there are more documents": function hitSkip(next){
+				var lds = SkipDocumentSource.createFromJson(1);
+
+                var cwc = new CursorDocumentSource.CursorWithContext();
+                var input = [{val:1},{val:2},{val:3}];
+                cwc._cursor = new Cursor( input );
+                var cds = new CursorDocumentSource(cwc);
+				lds.setSource(cds);
+
+				lds.getNext(function(err,actual) {
+					assert.notEqual(actual, DocumentSource.EOF);
+					assert.deepEqual(actual, {val:2});
+					next();
+				});
 			},
-			"should return false if skip count is not hit and there are more documents": function hitSkip(){
-				var lds = new SkipDocumentSource();
-				lds.skip = 10;
-				lds.count = 9;
-				
-				var i = 1;
-				lds.source = {
-					getCurrent:function(){return { item:i };},
-					eof: function(){return false;},
-					advance: function(){i++; return true;}
-				};
-				assert.equal(lds.eof(), false);
-			}
-
-		},
-
-		"#getCurrent()": {
 
 			"should return the current document source": function currSource(){
-				var lds = new SkipDocumentSource();
-				lds.skip = 1;
-				
-				var i = 0;
-				lds.source = {
-					getCurrent:function(){return { item:i };},
-					eof: function(){return false;},
-					advance: function(){i++; return true;}
-				};
-				assert.deepEqual(lds.getCurrent(), { item:1 });
-			}
-
-		},
+				var lds = SkipDocumentSource.createFromJson(1);
 
-		"#advance()": {
+                var cwc = new CursorDocumentSource.CursorWithContext();
+                var input = [{val:1},{val:2},{val:3}];
+                cwc._cursor = new Cursor( input );
+                var cds = new CursorDocumentSource(cwc);
+				lds.setSource(cds);
 
-			"should return true for moving to the next source": function nextSource(){
-				var lds = new SkipDocumentSource();
-				lds.skip = 1;
-				
-				var i = 0;
-				lds.source = {
-					getCurrent:function(){return { item:i };},
-					eof: function(){return false;},
-					advance: function(){i++; return true;}
-				};
-				assert.strictEqual(lds.advance(), true);
-			},
-
-			"should return false for no sources remaining": function noMoar(){
-				var lds = new SkipDocumentSource();
-				lds.skip = 1;
-				
-				var i = 0;
-				lds.source = {
-					getCurrent:function(){return { item:i };},
-					eof: function(){return true;},
-					advance: function(){return false;}
-				};
-				assert.strictEqual(lds.advance(), false);
+				lds.getNext(function(err, actual) {
+					assert.deepEqual(actual, { val:2 });
+				});
 			},
 
-			"should return false if we hit our limit": function noMoar(){
+			"should return false if we hit our limit": function noMoar(next){
 				var lds = new SkipDocumentSource();
 				lds.skip = 3;
-				
-				var i = 0;
+
+				var expected = [
+					{item:4},
+					DocumentSource.EOF
+				];
+
+				var i = 1;
 				lds.source = {
-					getCurrent:function(){return { item:i };},
-					eof: function(){return i>=5;},
-					advance: function(){i++; return i<5;}
+					getNext:function(cb){
+						if (i>=5)
+							return cb(null,DocumentSource.EOF);
+						return cb(null, { item:i++ });
+					}
 				};
-				assert.strictEqual(lds.advance(), true);
-				assert.strictEqual(lds.advance(), false);
+
+				async.series([
+						lds.getNext.bind(lds),
+						lds.getNext.bind(lds),
+					],
+					function(err,res) {
+						assert.deepEqual(expected, res);
+						next();
+					}
+				);
 			}
 
 		},
 
-		"#sourceToJson()": {
+		"#serialize()": {
 
 			"should create an object with a key $skip and the value equal to the skip": function sourceToJsonTest(){
 				var lds = new SkipDocumentSource();
 				lds.skip = 9;
-				var t = {};
-				lds.sourceToJson(t, false);
+				var t = lds.serialize(false);
 				assert.deepEqual(t, { "$skip": 9 });
 			}