stream: add more forEach tests

test/parallel/test-stream-forEach.js

@@ -43,6 +43,51 @@ const { setTimeout } = require('timers/promises');
   })().then(common.mustCall());
 }
 
+{
+  // forEach works on an infinite stream
+  const ac = new AbortController();
+  const { signal } = ac;
+  const stream = Readable.from(async function* () {
+    while (true) yield 1;
+  }(), { signal });
+  let i = 0;
+  assert.rejects(stream.forEach(common.mustCall((x) => {
+    i++;
+    if (i === 10) ac.abort();
+    assert.strictEqual(x, 1);
+  }, 10)), { name: 'AbortError' });
+}
+
+{
+  // Emitting an error during `forEach`
+  const stream = Readable.from([1, 2, 3, 4, 5]);
+  assert.rejects(stream.forEach(async (x) => {
+    if (x === 3) {
+      stream.emit('error', new Error('boom'));
+    }
+  }), /boom/).then(common.mustCall());
+}
+
+{
+  // Throwing an error during `forEach` (sync)
+  const stream = Readable.from([1, 2, 3, 4, 5]);
+  assert.rejects(stream.forEach((x) => {
+    if (x === 3) {
+      throw new Error('boom');
+    }
+  }), /boom/).then(common.mustCall());
+}
+
+{
+  // Throwing an error during `forEach` (async)
+  const stream = Readable.from([1, 2, 3, 4, 5]);
+  assert.rejects(stream.forEach(async (x) => {
+    if (x === 3) {
+      return Promise.reject(new Error('boom'));
+    }
+  }), /boom/).then(common.mustCall());
+}
+
 {
   // Concurrency + AbortSignal
   const ac = new AbortController();