Fix case that backpressure lose consistency on error

Author: Pedro Kehl

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "caminho",
-  "version": "1.7.0",
+  "version": "1.7.1",
   "description": "Tool for creating efficient data pipelines in a JavaScript environment",
   "main": "./dist/cjs/index.js",
   "module": "./dist/esm5/index.js",
@@ -72,4 +72,4 @@
   "dependencies": {
     "rxjs": "^7.8.1"
   }
-}
+}

package.json

@@ -72,7 +72,7 @@ export class Caminho implements CaminhoInterface {
     const loggers = this.getLoggers(generatorParams)
     if (this.options?.maxItemsFlowing) {
       const pendingDataControl = this.pendingDataControl as PendingDataControl
-      this.finalStep = tap(() => pendingDataControl.decrement())
+      this.finalStep = tap({ next: () => pendingDataControl.decrement(), error: () => pendingDataControl.decrement() })
       return wrapGeneratorWithBackPressure(generatorParams, this.options.maxItemsFlowing, pendingDataControl, loggers)
     }
 

src/Caminho.ts

@@ -56,4 +56,13 @@ describe('Error Handling', () => {
 
     await expect(caminho.run()).rejects.toMatchObject({ message: 'Reduce error' })
   })
+
+  test('Should not interfere with the backpressure when error happens', async () => {
+    const operator = jest.fn().mockRejectedValueOnce(new Error('Operator error')).mockResolvedValue(null)
+    const caminho = fromGenerator({ fn: getMockedGenerator([1, 2]), provides: 'number' }, { maxItemsFlowing: 1 })
+      .pipe({ fn: operator })
+
+    await expect(caminho.run()).rejects.toMatchObject({ message: 'Operator error' })
+    expect(caminho.getNumberOfItemsFlowing()).toBe(0)
+  })
 })