fix: handle errors from file#createReadStream

src/file.ts

@@ -1212,6 +1212,8 @@ class File extends ServiceObject<File> {
               return;
             }
 
+            rawResponseStream.on('error', onComplete);
+
             const headers = rawResponseStream.toJSON().headers;
             const isCompressed = headers['content-encoding'] === 'gzip';
             const shouldRunValidation = !rangeRequest && (crc32c || md5);
@@ -1235,9 +1237,9 @@ class File extends ServiceObject<File> {
                   rawResponseStream.pipe(pumpify.obj(throughStreams));
             }
 
-            rawResponseStream.on('end', onComplete).pipe(throughStream, {
-              end: false
-            });
+            rawResponseStream.on('error', onComplete)
+                .on('end', onComplete)
+                .pipe(throughStream, {end: false});
           };
 
       // This is hooked to the `complete` event from the request stream. This is

test/file.ts

@@ -61,6 +61,15 @@ const fakePromisify = {
 const fsCached = extend(true, {}, fs);
 const fakeFs = extend(true, {}, fsCached);
 
+const zlibCached = extend(true, {}, zlib);
+let createGunzipOverride: Function|null;
+const fakeZlib = extend(true, {}, zlib, {
+  createGunzip() {
+    return (createGunzipOverride || zlibCached.createGunzip)
+        .apply(null, arguments);
+  },
+});
+
 let hashStreamValidationOverride: Function|null;
 const hashStreamValidation = require('hash-stream-validation');
 function fakeHashStreamValidation() {
@@ -148,6 +157,7 @@ describe('File', () => {
              'hash-stream-validation': fakeHashStreamValidation,
              os: fakeOs,
              'xdg-basedir': fakeXdgBasedir,
+             zlib: fakeZlib,
            }).File;
   });
 
@@ -178,6 +188,7 @@ describe('File', () => {
     directoryFile = new File(BUCKET, 'directory/file.jpg');
     directoryFile.request = util.noop;
 
+    createGunzipOverride = null;
     handleRespOverride = null;
     hashStreamValidationOverride = null;
     makeWritableStreamOverride = null;
@@ -957,6 +968,39 @@ describe('File', () => {
                   })
               .resume();
         });
+
+        it('should emit errors from the request stream', done => {
+          const error = new Error('Error.');
+          const rawResponseStream = through();
+          // tslint:disable-next-line:no-any
+          (rawResponseStream as any).toJSON = () => {
+            return {headers: {}};
+          };
+          const requestStream = through();
+
+          handleRespOverride =
+              (err: Error, res: {}, body: {}, callback: Function) => {
+                callback(null, null, rawResponseStream);
+                setImmediate(() => {
+                  rawResponseStream.emit('error', error);
+                });
+              };
+
+          file.requestStream = () => {
+            setImmediate(() => {
+              requestStream.emit('response', rawResponseStream);
+            });
+            return requestStream;
+          };
+
+          file.createReadStream()
+              .on('error',
+                  (err: Error) => {
+                    assert.strictEqual(err, error);
+                    done();
+                  })
+              .resume();
+        });
       });
     });
 
@@ -995,6 +1039,24 @@ describe('File', () => {
                 })
             .resume();
       });
+
+      it('should emit errors from the gunzip stream', done => {
+        const error = new Error('Error.');
+        const createGunzipStream = through();
+        createGunzipOverride = () => {
+          setImmediate(() => {
+            createGunzipStream.emit('error', error);
+          });
+          return createGunzipStream;
+        };
+        file.createReadStream()
+            .on('error',
+                (err: Error) => {
+                  assert.strictEqual(err, error);
+                  done();
+                })
+            .resume();
+      });
     });
 
     describe('validation', () => {
@@ -1022,6 +1084,27 @@ describe('File', () => {
         };
       });
 
+      it('should emit errors from the validation stream', done => {
+        const error = new Error('Error.');
+
+        hashStreamValidationOverride = () => {
+          setImmediate(() => {
+            fakeValidationStream.emit('error', error);
+          });
+          return fakeValidationStream;
+        };
+
+        file.requestStream = getFakeSuccessfulRequest(data);
+
+        file.createReadStream()
+            .on('error',
+                (err: Error) => {
+                  assert.strictEqual(err, error);
+                  done();
+                })
+            .resume();
+      });
+
       it('should pass the userProject to getMetadata', done => {
         const fakeOptions = {
           userProject: 'grapce-spaceship-123',