Fix async pause/resume with ReadableStreamStreamer (#490)

* Fix bug where pausing and resuming asynchronously caused the csv file to be parsed incorrectly

papaparse.js

@@ -753,6 +753,19 @@
 
 		var queue = [];
 		var parseOnData = true;
+		var streamHasEnded = false;
+
+		this.pause = function()
+		{
+			ChunkStreamer.prototype.pause.apply(this, arguments);
+			this._input.pause();
+		};
+
+		this.resume = function()
+		{
+			ChunkStreamer.prototype.resume.apply(this, arguments);
+			this._input.resume();
+		};
 
 		this.stream = function(stream)
 		{
@@ -763,8 +776,16 @@
 			this._input.on('error', this._streamError);
 		};
 
+		this._checkIsFinished = function()
+		{
+			if (streamHasEnded && queue.length === 1) {
+				this._finished = true;
+			}
+		};
+
 		this._nextChunk = function()
 		{
+			this._checkIsFinished();
 			if (queue.length)
 			{
 				this.parseChunk(queue.shift());
@@ -784,6 +805,7 @@
 				if (parseOnData)
 				{
 					parseOnData = false;
+					this._checkIsFinished();
 					this.parseChunk(queue.shift());
 				}
 			}
@@ -802,7 +824,7 @@
 		this._streamEnd = bindFunction(function()
 		{
 			this._streamCleanUp();
-			this._finished = true;
+			streamHasEnded = true;
 			this._streamData('');
 		}, this);
 

tests/node-tests.js

@@ -57,6 +57,7 @@ describe('PapaParse', function() {
 				done();
 			},
 		});
+	});
 
 	it('should support pausing and resuming on same tick when streaming', function(done) {
 		var rows = [];
@@ -93,6 +94,43 @@ describe('PapaParse', function() {
 		});
 	});
 
+	it('should support pausing and resuming asynchronously when streaming', function(done) {
+		var rows = [];
+		Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
+			chunk: function(results, parser) {
+				rows = rows.concat(results.data);
+				parser.pause();
+				setTimeout(function() {
+					parser.resume();
+				}, 200);
+			},
+			error: function(err) {
+				done(new Error(err));
+			},
+			complete: function() {
+				assert.deepEqual(rows[0], [
+					'Grant',
+					'Dyer',
+					'Donec.elementum@orciluctuset.example',
+					'2013-11-23T02:30:31-08:00',
+					'2014-05-31T01:06:56-07:00',
+					'Magna Ut Associates',
+					'ljenkins'
+				]);
+				assert.deepEqual(rows[7], [
+					'Talon',
+					'Salinas',
+					'posuere.vulputate.lacus@Donecsollicitudin.example',
+					'2015-01-31T09:19:02-08:00',
+					'2014-12-17T04:59:18-08:00',
+					'Aliquam Iaculis Incorporate',
+					'Phasellus@Quisquetincidunt.example'
+				]);
+				done();
+			}
+		});
+	});
+
 	it('handles errors in beforeFirstChunk', function(done) {
 		var expectedError = new Error('test');
 		Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
@@ -131,5 +169,4 @@ describe('PapaParse', function() {
 			}
 		});
 	});
-	});
 });