From c6a4769673db43f23e3de1bd667115381d56263f Mon Sep 17 00:00:00 2001
From: Stephen <stephenplusplus@users.noreply.github.com>
Date: Tue, 28 Jul 2020 16:57:16 -0400
Subject: [PATCH 1/2] fix: destroy ResourceStream with pre-flight error

---
 src/resource-stream.ts  | 79 ++++++++++++++++++++++-------------------
 test/resource-stream.ts | 12 +++++++
 2 files changed, 55 insertions(+), 36 deletions(-)

diff --git a/src/resource-stream.ts b/src/resource-stream.ts
index 77ee762..823d603 100644
--- a/src/resource-stream.ts
+++ b/src/resource-stream.ts
@@ -59,43 +59,50 @@ export class ResourceStream<T> extends Transform implements ResourceEvents<T> {
 
     this._reading = true;
 
-    this._requestFn(
-      this._nextQuery,
-      (err: Error | null, results: T[], nextQuery: {} | null) => {
-        if (err) {
-          this.destroy(err);
-          return;
-        }
-
-        this._nextQuery = nextQuery;
-
-        if (this._resultsToSend !== Infinity) {
-          results = results.splice(0, this._resultsToSend);
-          this._resultsToSend -= results.length;
-        }
-
-        let more = true;
-
-        for (const result of results) {
-          if (this._ended) {
-            break;
+    // Wrap in a try/catch to catch input linting errors, e.g.
+    // an invalid BigQuery query. These errors are thrown in an
+    // async fashion, which makes them un-catchable by the user.
+    try {
+      this._requestFn(
+        this._nextQuery,
+        (err: Error | null, results: T[], nextQuery: {} | null) => {
+          if (err) {
+            this.destroy(err);
+            return;
           }
-          more = this.push(result);
-        }
-
-        const isFinished = !this._nextQuery || this._resultsToSend < 1;
-        const madeMaxCalls = ++this._requestsMade >= this._maxApiCalls;
-
-        if (isFinished || madeMaxCalls) {
-          this.end();
-        }
-
-        if (more && !this._ended) {
-          setImmediate(() => this._read());
+  
+          this._nextQuery = nextQuery;
+  
+          if (this._resultsToSend !== Infinity) {
+            results = results.splice(0, this._resultsToSend);
+            this._resultsToSend -= results.length;
+          }
+  
+          let more = true;
+  
+          for (const result of results) {
+            if (this._ended) {
+              break;
+            }
+            more = this.push(result);
+          }
+  
+          const isFinished = !this._nextQuery || this._resultsToSend < 1;
+          const madeMaxCalls = ++this._requestsMade >= this._maxApiCalls;
+  
+          if (isFinished || madeMaxCalls) {
+            this.end();
+          }
+  
+          if (more && !this._ended) {
+            setImmediate(() => this._read());
+          }
+  
+          this._reading = false;
         }
-
-        this._reading = false;
-      }
-    );
+      ); 
+    } catch (e) {
+      this.destroy(e);
+    }
   }
 }
diff --git a/test/resource-stream.ts b/test/resource-stream.ts
index 64d1790..970c12a 100644
--- a/test/resource-stream.ts
+++ b/test/resource-stream.ts
@@ -280,5 +280,17 @@ describe('ResourceStream', () => {
 
       assert.strictEqual(stream._reading, false);
     });
+
+    it('should destroy the stream if the request method throws', done => {
+      const error = new Error('Error.');
+      stream._requestFn = () => {
+        throw error;
+      };
+      stream.on('error', err => {
+        assert.strictEqual(err, error);
+        done();
+      });
+      stream._read()
+    });
   });
 });

From d33c9f9584a7203df856195596b23138b50ec373 Mon Sep 17 00:00:00 2001
From: Stephen <stephenplusplus@users.noreply.github.com>
Date: Tue, 28 Jul 2020 17:11:34 -0400
Subject: [PATCH 2/2] lint fix

---
 src/resource-stream.ts  | 18 +++++++++---------
 test/resource-stream.ts |  2 +-
 2 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/src/resource-stream.ts b/src/resource-stream.ts
index 823d603..330948b 100644
--- a/src/resource-stream.ts
+++ b/src/resource-stream.ts
@@ -70,37 +70,37 @@ export class ResourceStream<T> extends Transform implements ResourceEvents<T> {
             this.destroy(err);
             return;
           }
-  
+
           this._nextQuery = nextQuery;
-  
+
           if (this._resultsToSend !== Infinity) {
             results = results.splice(0, this._resultsToSend);
             this._resultsToSend -= results.length;
           }
-  
+
           let more = true;
-  
+
           for (const result of results) {
             if (this._ended) {
               break;
             }
             more = this.push(result);
           }
-  
+
           const isFinished = !this._nextQuery || this._resultsToSend < 1;
           const madeMaxCalls = ++this._requestsMade >= this._maxApiCalls;
-  
+
           if (isFinished || madeMaxCalls) {
             this.end();
           }
-  
+
           if (more && !this._ended) {
             setImmediate(() => this._read());
           }
-  
+
           this._reading = false;
         }
-      ); 
+      );
     } catch (e) {
       this.destroy(e);
     }
diff --git a/test/resource-stream.ts b/test/resource-stream.ts
index 970c12a..27a47f9 100644
--- a/test/resource-stream.ts
+++ b/test/resource-stream.ts
@@ -290,7 +290,7 @@ describe('ResourceStream', () => {
         assert.strictEqual(err, error);
         done();
       });
-      stream._read()
+      stream._read();
     });
   });
 });