Repository: mholt/PapaParse
Pull request #490 (2018-04-17 16:11:18)

Fix async pause/resume with ReadableStreamStreamer
I discovered this bug when trying to process and upload 1.3 million csv records to an elasticsearch instance.  The csv reader was parsing data into my queue faster than I could upload the data in my datastore and thus the process memory would crash node.  I implemented a pausing system so that the upload queue wouldn't get overloaded but then I noticed that the parser would call `complete` before the entire csv had been processed.  This fix allows for the pause/resume to work correctly when they are triggered in an async fashion.

trevorharwell: I think this issue might relate to https://github.com/mholt/PapaParse/issues/422

Patch:
diff --git a/papaparse.js b/papaparse.js
index 6fbd9bf..c8b5efa 100755
--- a/papaparse.js
+++ b/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);
 


Test patch:
diff --git a/tests/node-tests.js b/tests/node-tests.js
index 102ea75..44752ff 100644
--- a/tests/node-tests.js
+++ b/tests/node-tests.js
@@ -57,79 +57,116 @@ describe('PapaParse', function() {
 				done();
 			},
 		});
+	});
+
+	it('should support pausing and resuming on same tick 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();
+				parser.resume();
+			},
+			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('should support pausing and resuming on same tick 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();
+	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();
-				},
-				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();
-				}
-			});
+				}, 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'), {
-				beforeFirstChunk: function() {
-					throw expectedError;
-				},
-				error: function(err) {
-					assert.deepEqual(err, expectedError);
-					done();
-				}
-			});
+	it('handles errors in beforeFirstChunk', function(done) {
+		var expectedError = new Error('test');
+		Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
+			beforeFirstChunk: function() {
+				throw expectedError;
+			},
+			error: function(err) {
+				assert.deepEqual(err, expectedError);
+				done();
+			}
 		});
+	});
 
-		it('handles errors in chunk', function(done) {
-			var expectedError = new Error('test');
-			Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
-				chunk: function() {
-					throw expectedError;
-				},
-				error: function(err) {
-					assert.deepEqual(err, expectedError);
-					done();
-				}
-			});
+	it('handles errors in chunk', function(done) {
+		var expectedError = new Error('test');
+		Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
+			chunk: function() {
+				throw expectedError;
+			},
+			error: function(err) {
+				assert.deepEqual(err, expectedError);
+				done();
+			}
 		});
+	});
 
-		it('handles errors in step', function(done) {
-			var expectedError = new Error('test');
-			Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
-				step: function() {
-					throw expectedError;
-				},
-				error: function(err) {
-					assert.deepEqual(err, expectedError);
-					done();
-				}
-			});
+	it('handles errors in step', function(done) {
+		var expectedError = new Error('test');
+		Papa.parse(fs.createReadStream(__dirname + '/long-sample.csv', 'utf8'), {
+			step: function() {
+				throw expectedError;
+			},
+			error: function(err) {
+				assert.deepEqual(err, expectedError);
+				done();
+			}
 		});
 	});
 });