core: ending streams manually is unnecessary, unless they errored

lib/bigquery/table.js

@@ -328,7 +328,6 @@ Table.prototype.createWriteStream = function(metadata) {
       job.metadata = data;
 
       dup.emit('complete', job);
-      dup.end();
     });
   });
 

lib/common/stream-router.js

@@ -184,8 +184,7 @@ streamRouter.runAsStream_ = function(parsedArguments, originalMethod) {
 
   function onResultSet(err, results, nextQuery) {
     if (err) {
-      stream.emit('error', err);
-      stream.end();
+      stream.destroy(err);
       return;
     }
 
@@ -206,7 +205,6 @@ streamRouter.runAsStream_ = function(parsedArguments, originalMethod) {
       }
 
       stream.push(null);
-      stream.end();
     });
   }
 

lib/common/util.js

@@ -242,8 +242,7 @@ function makeWritableStream(dup, options, onComplete) {
   options.makeAuthorizedRequest(reqOpts, {
     onAuthorized: function(err, authorizedReqOpts) {
       if (err) {
-        dup.emit('error', err);
-        dup.end();
+        dup.destroy(err);
         return;
       }
 
@@ -274,8 +273,7 @@ function makeWritableStream(dup, options, onComplete) {
       stream.on('complete', function(res) {
         util.handleResp(null, res, res.body, function(err, data) {
           if (err) {
-            dup.emit('error', err);
-            dup.end();
+            dup.destroy(err);
             return;
           }
           onComplete(data);
@@ -474,8 +472,7 @@ function makeAuthorizedRequestFactory(config) {
     function onAuthorized(err, authorizedReqOpts) {
       if (err) {
         if (stream) {
-          stream.emit('error', err);
-          stream.end();
+          stream.destroy(err);
         } else {
           (options.onAuthorized || options)(err);
         }

lib/datastore/request.js

@@ -167,8 +167,7 @@ DatastoreRequest.prototype.get = function(keys, callback) {
 
   function onApiResponse(err, resp) {
     if (err) {
-      stream.emit('error', err);
-      stream.end();
+      stream.destroy(err);
       return;
     }
 
@@ -186,7 +185,6 @@ DatastoreRequest.prototype.get = function(keys, callback) {
       }
 
       stream.push(null);
-      stream.end();
     });
   }
 

lib/storage/file.js

@@ -467,11 +467,10 @@ File.prototype.createReadStream = function(options) {
   var endThroughStream = once(function(err, resp) {
     if (err) {
       throughStream.emit('error', err, resp);
+      throughStream.destroy();
     } else {
       throughStream.emit('complete', resp);
     }
-
-    throughStream.destroy();
   });
 
   makeAuthorizedReq(remoteFilePath);
@@ -788,8 +787,7 @@ File.prototype.createWriteStream = function(options) {
 
     // Catch any errors from the writable stream and patch them upstream.
     .on('error', function(err) {
-      throughStream.emit('error', err);
-      throughStream.end();
+      throughStream.destroy(err);
     })
 
     // Compare our hashed version vs the completed upload's version.
@@ -840,13 +838,11 @@ File.prototype.createWriteStream = function(options) {
           error.code = code;
           error.errors = [err];
 
-          throughStream.emit('error', error);
+          throughStream.destroy(error);
         });
       } else {
         throughStream.emit('complete', metadata);
       }
-
-      throughStream.end();
     });
 
   return throughStream;
@@ -1669,7 +1665,7 @@ File.prototype.startResumableUpload_ = function(stream, metadata) {
       return;
     }
 
-    stream.emit('error', err);
+    stream.destroy(err);
   }
 };
 
@@ -1708,7 +1704,6 @@ File.prototype.startSimpleUpload_ = function(stream, metadata) {
     that.metadata = data;
 
     stream.emit('complete', data);
-    stream.end();
   });
 };
 

test/common/util.js

@@ -27,6 +27,7 @@ var request = require('request');
 var retryRequest = require('retry-request');
 var stream = require('stream');
 var streamForward = require('stream-forward');
+var through = require('through2');
 
 var googleAuthLibraryOverride;
 function fakeGoogleAuthLibrary() {
@@ -306,7 +307,7 @@ describe('common/util', function() {
     it('should emit an error', function(done) {
       var error = new Error('Error.');
 
-      var ws = new stream.Writable();
+      var ws = through();
       ws.on('error', function(err) {
         assert.equal(err, error);
         done();
@@ -840,7 +841,7 @@ describe('common/util', function() {
 
             var stream = this;
             setImmediate(function() {
-              assert.strictEqual(stream._readableState.ended, true);
+              assert.strictEqual(stream._destroyed, true);
               done();
             });
           });

test/datastore/request.js

@@ -205,7 +205,7 @@ describe('Request', function() {
 
           stream.on('error', function() {
             setImmediate(function() {
-              assert.strictEqual(stream._readableState.ended, true);
+              assert.strictEqual(stream._destroyed, true);
               done();
             });
           });

test/storage/file.js

@@ -48,6 +48,13 @@ function FakeDuplexify() {
   this.setWritable = function(setWritableStream) {
     writableStream = setWritableStream;
   };
+  this.destroy = function(err) {
+    if (err) {
+      this.emit('error', err);
+    } else {
+      this.end();
+    }
+  };
 }
 nodeutil.inherits(FakeDuplexify, stream.Duplex);
 
@@ -550,12 +557,6 @@ describe('File', function() {
               done();
             });
         });
-
-        it('should destroy the through stream', function(done) {
-          var readStream = file.createReadStream();
-          readStream.once('error', util.noop);
-          readStream.destroy = done;
-        });
       });
     });
 
@@ -650,11 +651,6 @@ describe('File', function() {
               done();
             });
         });
-
-        it('should destroy the through stream', function(done) {
-          var readStream = file.createReadStream();
-          readStream.destroy = done;
-        });
       });
     });
 
@@ -748,15 +744,14 @@ describe('File', function() {
               'bad-data', fakeResponse.crc32c);
 
           var readStream = file.createReadStream({ validation: 'md5' });
-          readStream.destroy = done;
+          readStream.end = done;
         });
 
         it('should destroy after successful validation', function(done) {
-          requestOverride = getFakeSuccessfulRequest(
-              data, fakeResponse.crc32c);
+          requestOverride = getFakeSuccessfulRequest(data, fakeResponse.crc32c);
 
           var readStream = file.createReadStream({ validation: 'crc32c' });
-          readStream.destroy = done;
+          readStream.end = done;
         });
       });
     });
@@ -822,11 +817,11 @@ describe('File', function() {
         file.createReadStream({ start: startOffset, end: endOffset });
       });
 
-      it('should destroy the through stream', function(done) {
+      it('should end the through stream', function(done) {
         requestOverride = getFakeSuccessfulRequest('body', { body: null });
 
         var readStream = file.createReadStream({ start: 100 });
-        readStream.destroy = done;
+        readStream.end = done;
       });
     });