feat: destroy stream on error

src/methods/chunk.ts

@@ -1,28 +1,9 @@
-import { Transform } from "stream";
 import { TransformTyped, TransformTypedOptions } from "../types";
+import { chunkMap } from "./chunkMap";
 
 export function chunk<R>(
   size: number,
   options: TransformTypedOptions<R, R[]> = {}
 ): TransformTyped<R, R[]> {
-  let chunk: R[] = [];
-  return new Transform({
-    objectMode: true,
-    ...options,
-    transform(item, _encoding, callback) {
-      chunk.push(item);
-      if (chunk.length >= size) {
-        this.push(chunk);
-        chunk = [];
-      }
-      callback();
-    },
-    flush(callback) {
-      if (chunk.length > 0) {
-        this.push(chunk);
-        chunk = [];
-      }
-      callback();
-    }
-  });
+  return chunkMap(size, (chunk) => chunk, options);
 }

src/methods/chunkMap.ts

@@ -13,21 +13,31 @@ export function chunkMap<T, R>(
     objectMode: true,
     ...options,
     async transform(item, _encoding, callback) {
-      chunk.push(item);
-      if (chunk.length >= size) {
-        const result = await method(chunk, index++);
-        this.push(result);
-        chunk = [];
+      try {
+        chunk.push(item);
+        if (chunk.length >= size) {
+          const result = await method(chunk, index++);
+          this.push(result);
+          chunk = [];
+        }
+        callback();
+      } catch (err) {
+        callback(err);
+        this.destroy();
       }
-      callback();
     },
     async flush(callback) {
-      if (chunk.length > 0) {
-        const result = await method(chunk, index++);
-        this.push(result);
-        chunk = [];
+      try {
+        if (chunk.length > 0) {
+          const result = await method(chunk, index++);
+          this.push(result);
+          chunk = [];
+        }
+        callback();
+      } catch (err) {
+        callback(err);
+        this.destroy();
       }
-      callback();
     }
   });
 }

src/methods/filter.ts

@@ -15,6 +15,7 @@ export function filter<T>(
         callback(undefined, take ? chunk : undefined);
       } catch (err) {
         callback(err);
+        this.destroy();
       }
     }
   });

src/methods/map.ts

@@ -17,6 +17,7 @@ export function map<T, R>(
         callback();
       } catch (err) {
         callback(err);
+        this.destroy();
       }
     }
   });

src/methods/reduce.ts

@@ -17,6 +17,7 @@ export function reduce<T, R>(
         callback();
       } catch (err) {
         callback(err);
+        this.destroy();
       }
     },
     flush(callback) {

src/methods/transform.spec.ts

@@ -67,4 +67,32 @@ describe('transform', () => {
       done();
     });
   });
+
+  it('handles errors by closing stream', (done) => {
+    const check = jest.fn();
+
+    from([1, 2, 3])
+    .pipe(transform(() => {
+      throw new Error('test')
+    }))
+    .on('error', check)
+    .on('close', () => {
+      expect(check.mock.calls.length).toEqual(1);
+      done();
+    });
+  });
+
+  it('handles errors by closing stream', (done) => {
+    const check = jest.fn();
+
+    from([1, 2, 3])
+    .pipe(transform(async () => {
+      throw new Error('test')
+    }))
+    .on('error', check)
+    .on('close', () => {
+      expect(check.mock.calls.length).toEqual(1);
+      done();
+    });
+  });
 });

src/methods/transform.ts

@@ -37,6 +37,7 @@ export function transform<In, Out>(
         callback();
       } catch (err) {
         callback(err);
+        this.destroy();
       }
     },
     async flush(callback) {
@@ -50,6 +51,7 @@ export function transform<In, Out>(
         callback();
       } catch (err) {
         callback(err);
+        this.destroy();
       }
     }
   });