Fix #6 - Allow early ending table/createReadStream (#18)

* Fix #6 - Allow early ending table/createReadStream Fixed data transformation stream to stop processing data on `end` event Still, need to fix GRPC stream closing.
googleapis · Dec 13, 2017 · bfe42dd · bfe42dd
1 parent e4770e6
commit bfe42dd
Show file tree

Hide file tree

Showing 3 changed files with 71 additions and 1 deletion.
diff --git a/src/table.js b/src/table.js
@@ -489,7 +489,7 @@ Table.prototype.createReadStream = function(options) {
     reqOpts.rowsLimit = options.limit;
   }
 
-  return pumpify.obj([
+  var stream = pumpify.obj([
     this.requestStream(grpcOpts, reqOpts),
     through.obj(function(data, enc, next) {
       var throughStream = this;
@@ -498,6 +498,10 @@ Table.prototype.createReadStream = function(options) {
       });
 
       rows.forEach(function(rowData) {
+        if (stream._ended) {
+          return;
+        }
+
         var row = self.row(rowData.key);
 
         row.data = rowData.data;
@@ -507,6 +511,8 @@ Table.prototype.createReadStream = function(options) {
       next();
     }),
   ]);
+
+  return stream;
 };
 
 /**

diff --git a/system-test/bigtable.js b/system-test/bigtable.js
@@ -668,6 +668,54 @@ describe('Bigtable', function() {
           });
       });
 
+      it('should end stream early', function(done) {
+        var entries = [
+          {
+            key: 'gwashington',
+            data: {
+              follows: {
+                jadams: 1,
+              },
+            },
+          },
+          {
+            key: 'tjefferson',
+            data: {
+              follows: {
+                gwashington: 1,
+                jadams: 1,
+              },
+            },
+          },
+          {
+            key: 'jadams',
+            data: {
+              follows: {
+                gwashington: 1,
+                tjefferson: 1,
+              },
+            },
+          },
+        ];
+
+        TABLE.insert(entries, function(err) {
+          assert.ifError(err);
+
+          var rows = [];
+
+          TABLE.createReadStream()
+            .on('error', done)
+            .on('data', function(row) {
+              rows.push(row);
+              this.end();
+            })
+            .on('end', function() {
+              assert.strictEqual(rows.length, 1);
+              done();
+            });
+        });
+      });
+
       describe('filters', function() {
         it('should get rows via column data', function(done) {
           var filter = {

diff --git a/test/table.js b/test/table.js
@@ -634,6 +634,22 @@ describe('Bigtable/Table', function() {
             done();
           });
       });
+
+      it('should allow a stream to end early', function(done) {
+        var rows = [];
+
+        table
+          .createReadStream()
+          .on('error', done)
+          .on('data', function(row) {
+            rows.push(row);
+            this.end();
+          })
+          .on('end', function() {
+            assert.strictEqual(rows.length, 1);
+            done();
+          });
+      });
     });
 
     describe('error', function() {