handle error in pass through stream

remerge · May 8, 2023 · a554efb · a554efb
1 parent 1daeddc
commit a554efb
Show file tree

Hide file tree

Showing 3 changed files with 37 additions and 11 deletions.
diff --git a/build/helper/concurrentLimitRequester.js b/build/helper/concurrentLimitRequester.js
@@ -20,14 +20,20 @@ export function concurrentLimitRequesterFactory(parameters) {
         var queueItem = requestQueue.shift();
         outstandingRequests++;
         var requestId = generateRequestId();
-        console.log("Starting request " + requestId + " (" + outstandingRequests + "/" + concurrentLimit + "): " + queueItem.request.query.queryType);
+        console.log("Starting request " + requestId + " (" + outstandingRequests + "/" + concurrentLimit + "): " + queueItem.request.query.queryType + " from queue");
         var stream = requester(queueItem.request);
         var requestFinishedOnce = getOnceCallback(function () {
-            console.log("Request finished " + requestId);
+            console.log("Request finished " + requestId + " from Queue");
             requestFinished();
         });
         stream.on('error', requestFinishedOnce);
         stream.on('end', requestFinishedOnce);
+        queueItem.stream.on('error', function (error) {
+            requestFinishedOnce(function () {
+                console.log("Error on PassThrough for request " + requestId);
+                stream.emit('error', error);
+            });
+        });
         pipeWithError(stream, queueItem.stream);
     }
     return function (request) {
@@ -56,10 +62,12 @@ export function concurrentLimitRequesterFactory(parameters) {
 }
 function getOnceCallback(callback) {
     var called = false;
-    return function () {
+    return function (optionalCallback) {
+        if (optionalCallback === void 0) { optionalCallback = function () { }; }
         if (!called) {
             called = true;
             callback();
+            optionalCallback();
         }
     };
 }
diff --git a/build/plywood.js b/build/plywood.js
@@ -165,14 +165,20 @@ var concurrentLimitRequesterFactory = exports.concurrentLimitRequesterFactory =
         var queueItem = requestQueue.shift();
         outstandingRequests++;
         var requestId = generateRequestId();
-        console.log("Starting request " + requestId + " (" + outstandingRequests + "/" + concurrentLimit + "): " + queueItem.request.query.queryType);
+        console.log("Starting request " + requestId + " (" + outstandingRequests + "/" + concurrentLimit + "): " + queueItem.request.query.queryType + " from queue");
         var stream = requester(queueItem.request);
         var requestFinishedOnce = getOnceCallback(function () {
-            console.log("Request finished " + requestId);
+            console.log("Request finished " + requestId + " from Queue");
             requestFinished();
         });
         stream.on('error', requestFinishedOnce);
         stream.on('end', requestFinishedOnce);
+        queueItem.stream.on('error', function (error) {
+            requestFinishedOnce(function () {
+                console.log("Error on PassThrough for request " + requestId);
+                stream.emit('error', error);
+            });
+        });
         pipeWithError(stream, queueItem.stream);
     }
     return function (request) {
@@ -201,10 +207,12 @@ var concurrentLimitRequesterFactory = exports.concurrentLimitRequesterFactory =
 }
 function getOnceCallback(callback) {
     var called = false;
-    return function () {
+    return function (optionalCallback) {
+        if (optionalCallback === void 0) { optionalCallback = function () { }; }
         if (!called) {
             called = true;
             callback();
+            optionalCallback();
         }
     };
 }

diff --git a/src/helper/concurrentLimitRequester.ts b/src/helper/concurrentLimitRequester.ts
@@ -52,15 +52,23 @@ export function concurrentLimitRequesterFactory<T>(parameters: ConcurrentLimitRe
     let queueItem = requestQueue.shift();
     outstandingRequests++;
     const requestId = generateRequestId();
-    console.log(`Starting request ${requestId} (${outstandingRequests}/${concurrentLimit}): ${(queueItem.request.query as any).queryType}`);
+    console.log(`Starting request ${requestId} (${outstandingRequests}/${concurrentLimit}): ${(queueItem.request.query as any).queryType} from queue`);
 
     const stream = requester(queueItem.request);
 
     const requestFinishedOnce = getOnceCallback(() => {
-      console.log(`Request finished ${requestId}`);
+      console.log(`Request finished ${requestId} from Queue`);
       requestFinished();
     });
-    stream.on('close', requestFinishedOnce)
+    stream.on('error', requestFinishedOnce);
+    stream.on('end', requestFinishedOnce);
+
+    queueItem.stream.on('error', (error) => {
+      requestFinishedOnce(() => {
+        console.log(`Error on PassThrough for request ${requestId}`)
+        stream.emit('error', error);
+      })
+    });
 
     pipeWithError(stream, queueItem.stream);
   }
@@ -76,7 +84,8 @@ export function concurrentLimitRequesterFactory<T>(parameters: ConcurrentLimitRe
         console.log(`Request finished ${requestId}`);
         requestFinished();
       });
-      stream.on('close', requestFinishedOnce)
+      stream.on('error', requestFinishedOnce);
+      stream.on('end', requestFinishedOnce);
 
       return stream;
     } else {
@@ -93,10 +102,11 @@ export function concurrentLimitRequesterFactory<T>(parameters: ConcurrentLimitRe
 function getOnceCallback(callback: () => void) {
   let called = false;
 
-  return () => {
+  return (optionalCallback = () => {}) => {
     if (!called) {
       called = true;
       callback();
+      optionalCallback();
     }
   };
 }