concurrentLimitRequest | fix requestFinished fired multiple times on error

src/helper/concurrentLimitRequester.ts

@@ -37,25 +37,30 @@ export function concurrentLimitRequesterFactory<T>(parameters: ConcurrentLimitRe
 
   let requestQueue: QueueItem<T>[] = [];
   let outstandingRequests: int = 0;
-
   function requestFinished(): void {
     outstandingRequests--;
     if (!(requestQueue.length && outstandingRequests < concurrentLimit)) return;
     let queueItem = requestQueue.shift();
     outstandingRequests++;
 
     const stream = requester(queueItem.request);
-    stream.on('error', requestFinished);
-    stream.on('end', requestFinished);
+
+    const requestFinishedOnce = getOnceCallback(requestFinished);
+    stream.on('error', requestFinishedOnce);
+    stream.on('end', requestFinishedOnce);
+
     pipeWithError(stream, queueItem.stream);
   }
 
   return (request: DatabaseRequest<T>) => {
     if (outstandingRequests < concurrentLimit) {
       outstandingRequests++;
       const stream = requester(request);
-      stream.on('error', requestFinished);
-      stream.on('end', requestFinished);
+
+      const requestFinishedOnce = getOnceCallback(requestFinished);
+      stream.on('error', requestFinishedOnce);
+      stream.on('end', requestFinishedOnce);
+
       return stream;
     } else {
       const stream = new PassThrough({ objectMode: true });
@@ -67,3 +72,14 @@ export function concurrentLimitRequesterFactory<T>(parameters: ConcurrentLimitRe
     }
   };
 }
+
+function getOnceCallback(callback: () => void) {
+  let called = false;
+
+  return () => {
+    if (!called) {
+      called = true;
+      callback();
+    }
+  };
+}

test/helper/concurrentLimitRequester.mocha.js

@@ -116,29 +116,43 @@ describe("Concurrent Limit Requester", () => {
         expect(e.message).to.equal('fail');
         expect(nextQuery).to.equal('a');
         nextQuery = 'b';
+        expect(requester.hasQuery('c', 'has c')).to.equal(true);
+        expect(requester.hasQuery('d', 'has d')).to.equal(false);
+
+        requester.reject('b');
       });
 
     let rb = toArray(concurrentLimitRequester({ query: 'b' }))
       .catch((e) => {
         expect(e.message).to.equal('fail');
         expect(nextQuery).to.equal('b');
+
+        expect(requester.hasQuery('d', 'has d')).to.equal(true);
+
         nextQuery = 'c';
-        expect(requester.hasQuery('c', 'has c')).to.equal(true);
         requester.resolve('c');
       });
 
     let rc = toArray(concurrentLimitRequester({ query: 'c' }))
       .then((res) => {
         expect(res).to.deep.equal([1, 2, 3]);
         expect(nextQuery).to.equal('c');
+        nextQuery = 'd';
+        requester.resolve('d');
+      });
+
+      let rd = toArray(concurrentLimitRequester({ query: 'd' }))
+      .then((res) => {
+        expect(res).to.deep.equal([1, 2, 3]);
+        expect(nextQuery).to.equal('d');
       });
 
     expect(requester.hasQuery('a'), 'has a').to.equal(true);
     expect(requester.hasQuery('b'), 'has b').to.equal(true);
     expect(requester.hasQuery('c'), 'has c').to.equal(false);
+    expect(requester.hasQuery('d'), 'has d').to.equal(false);
     requester.reject('a');
-    requester.reject('b');
-    return Promise.all([ra, rb, rc]);
-  });
 
+    return Promise.all([ra, rb, rc, rd]);
+  });
 });