die this way die?

stephenplusplus · stephenplusplus · commit 7b4ab4fa5190 · 2015-05-05T17:03:27.000-04:00
diff --git a/lib/storage/file.js b/lib/storage/file.js
@@ -301,9 +301,10 @@ File.prototype.createReadStream = function(options) {
     util.is(options.start, 'number') || util.is(options.end, 'number');
   var throughStream = streamEvents(through());
 
+  var requestStream;
+
   var validations = ['crc32c', 'md5'];
   var validation;
-  var socket;
 
   if (util.is(options.validation, 'string')) {
     options.validation = options.validation.toLowerCase();
@@ -333,6 +334,22 @@ File.prototype.createReadStream = function(options) {
 
   createAuthorizedReq(remoteFilePath);
 
+  // End the stream, first emitting an error or complete event.
+  var endThroughStream = once(function(err, resp) {
+    if (err) {
+      throughStream.emit('error', err, resp);
+    } else {
+      throughStream.emit('complete', resp);
+    }
+
+    throughStream.destroy();
+  });
+
+  var endRequestStream = once(function() {
+    requestStream.abort();
+    requestStream.destroy();
+  });
+
   return throughStream;
 
   // Authenticate the request, then pipe the remote API request to the stream
@@ -359,7 +376,7 @@ File.prototype.createReadStream = function(options) {
     that.bucket.storage.makeAuthorizedRequest_(reqOpts, {
       onAuthorized: function(err, authorizedReqOpts) {
         if (err) {
-          done(err, null);
+          endThroughStream(err, null);
           return;
         }
 
@@ -368,8 +385,13 @@ File.prototype.createReadStream = function(options) {
         var localCrcHash;
         var localMd5Hash = crypto.createHash('md5');
 
-        request(authorizedReqOpts)
-          .on('error', done)
+        requestStream = request(authorizedReqOpts);
+
+        requestStream
+          .on('error', function(err) {
+            endRequestStream();
+            endThroughStream(err);
+          })
 
           .on('data', function(chunk) {
             if (crc32c) {
@@ -381,20 +403,16 @@ File.prototype.createReadStream = function(options) {
             }
           })
 
-          .on('socket', function(s) {
-            socket = s;
-          })
-
           .on('complete', function(res) {
             util.handleResp(null, res, res.body, function(err, resp) {
               if (err) {
-                done(err, resp);
+                endThroughStream(err, resp);
                 return;
               }
 
               if (rangeRequest) {
                 // Range requests can't receive data integrity checks.
-                done(null, resp);
+                endThroughStream(null, resp);
                 return;
               }
 
@@ -434,34 +452,19 @@ File.prototype.createReadStream = function(options) {
                 ].join(' '));
                 mismatchError.code = 'CONTENT_DOWNLOAD_MISMATCH';
 
-                done(mismatchError, resp);
+                endThroughStream(mismatchError, resp);
               } else {
-                done(null, resp);
+                endThroughStream(null, resp);
               }
             });
           })
 
-          .pipe(throughStream);
+          .pipe(throughStream)
 
-          throughStream.on('error', function() {
-            if (socket) {
-              socket.destroy();
-            }
-          });
+          .on('error', endRequestStream);
       }
     });
   }
-
-  // End the stream, first emitting an error or complete event.
-  function done(err) {
-    if (err) {
-      throughStream.emit('error', err);
-    } else {
-      throughStream.emit('complete');
-    }
-
-    throughStream.end();
-  }
 };
 
 /**
