stream: fixup stream/iter tests

Author: jasnell

test/parallel/test-stream-iter-broadcast-backpressure.js

@@ -57,20 +57,82 @@ async function testBlockBackpressure() {
   // Next write should block
   let writeResolved = false;
   const writePromise = writer.write('b').then(() => { writeResolved = true; });
-  await new Promise((r) => setTimeout(r, 10));
+  await new Promise(setImmediate);
   assert.strictEqual(writeResolved, false);
 
-  // Drain consumer
+  // Drain consumer to unblock the pending write
   const iter = consumer[Symbol.asyncIterator]();
-  await iter.next();
-  await new Promise((r) => setTimeout(r, 10));
+  const first = await iter.next();
+  assert.strictEqual(first.done, false);
+  await new Promise(setImmediate);
   assert.strictEqual(writeResolved, true);
+
   writer.endSync();
+  // Drain remaining data and verify completion
+  const second = await iter.next();
+  assert.strictEqual(second.done, false);
+  await writePromise;
+}
+
+// Verify block backpressure data flows correctly end-to-end
+async function testBlockBackpressureContent() {
+  const { writer, broadcast: bc } = broadcast({
+    highWaterMark: 1,
+    backpressure: 'block',
+  });
+  const consumer = bc.push();
+
+  writer.writeSync('a');
+  const writePromise = writer.write('b');
+  await new Promise(setImmediate);
+
+  // Read all and verify content
+  const iter = consumer[Symbol.asyncIterator]();
+  const first = await iter.next();
+  assert.strictEqual(first.done, false);
+  const firstStr = new TextDecoder().decode(first.value[0]);
+  assert.strictEqual(firstStr, 'a');
+
   await writePromise;
+  writer.endSync();
+
+  const second = await iter.next();
+  assert.strictEqual(second.done, false);
+  const secondStr = new TextDecoder().decode(second.value[0]);
+  assert.strictEqual(secondStr, 'b');
+
+  const done = await iter.next();
+  assert.strictEqual(done.done, true);
+}
+
+// Writev async path
+async function testWritevAsync() {
+  const { writer, broadcast: bc } = broadcast({ highWaterMark: 10 });
+  const consumer = bc.push();
+
+  await writer.writev(['hello', ' ', 'world']);
+  await writer.end();
+
+  const data = await text(consumer);
+  assert.strictEqual(data, 'hello world');
+}
+
+// endSync returns the total byte count
+async function testEndSyncReturnValue() {
+  const { writer, broadcast: bc } = broadcast({ highWaterMark: 10 });
+  bc.push(); // Need a consumer to write to
+
+  writer.writeSync('hello'); // 5 bytes
+  writer.writeSync(' world'); // 6 bytes
+  const total = writer.endSync();
+  assert.strictEqual(total, 11);
 }
 
 Promise.all([
   testDropOldest(),
   testDropNewest(),
   testBlockBackpressure(),
+  testBlockBackpressureContent(),
+  testWritevAsync(),
+  testEndSyncReturnValue(),
 ]).then(common.mustCall());

test/parallel/test-stream-iter-broadcast-basic.js

@@ -55,11 +55,12 @@ async function testConsumerCount() {
   bc.push();
   assert.strictEqual(bc.consumerCount, 2);
 
-  // Consume c1 to completion (it returns immediately since no data has been
-  // pushed and we haven't ended yet - but we'll cancel to detach)
   bc.cancel();
 
-  // After cancel, consumers are detached
+  // After cancel, consumer count drops to 0
+  assert.strictEqual(bc.consumerCount, 0);
+
+  // Consumers are detached and yield nothing
   const batches = [];
   for await (const batch of c1) {
     batches.push(batch);
@@ -103,7 +104,7 @@ async function testWriterEnd() {
 
   await writer.write('data');
   const totalBytes = await writer.end();
-  assert.ok(totalBytes > 0);
+  assert.strictEqual(totalBytes, 4); // 'data' = 4 UTF-8 bytes
 
   const data = await text(consumer);
   assert.strictEqual(data, 'data');
@@ -175,6 +176,9 @@ async function testFailDetachesConsumers() {
   await writer.write('data');
   await writer.fail(new Error('writer failed'));
 
+  // After fail, consumers are detached
+  assert.strictEqual(bc.consumerCount, 0);
+
   // Both consumers should see the error
   await assert.rejects(
     async () => {

test/parallel/test-stream-iter-broadcast-from.js

@@ -20,9 +20,10 @@ async function testBroadcastFromAsyncIterable() {
 
 async function testBroadcastFromNonArrayChunks() {
   // Source that yields single Uint8Array chunks (not arrays)
+  const enc = new TextEncoder();
   async function* singleChunkSource() {
-    yield new TextEncoder().encode('hello');
-    yield new TextEncoder().encode(' world');
+    yield enc.encode('hello');
+    yield enc.encode(' world');
   }
   const { broadcast: bc } = Broadcast.from(singleChunkSource());
   const consumer = bc.push();
@@ -98,10 +99,11 @@ async function testBroadcastFromCancelWhileBlocked() {
   // Create a slow async source that blocks between yields
   let sourceFinished = false;
   async function* slowSource() {
-    yield [new TextEncoder().encode('chunk1')];
+    const enc = new TextEncoder();
+    yield [enc.encode('chunk1')];
     // Simulate a long delay - the cancel should unblock this
     await new Promise((resolve) => setTimeout(resolve, 10000));
-    yield [new TextEncoder().encode('chunk2')];
+    yield [enc.encode('chunk2')];
     sourceFinished = true;
   }
 

test/parallel/test-stream-iter-consumers-bytes.js

@@ -21,46 +21,41 @@ const {
 // =============================================================================
 
 async function testBytesSyncBasic() {
-  const source = fromSync('hello');
-  const data = bytesSync(source);
+  const data = bytesSync(fromSync('hello'));
   assert.deepStrictEqual(data, new TextEncoder().encode('hello'));
 }
 
 async function testBytesSyncLimit() {
-  const source = fromSync('hello world');
   assert.throws(
-    () => bytesSync(source, { limit: 3 }),
+    () => bytesSync(fromSync('hello world'), { limit: 3 }),
     { name: 'RangeError' },
   );
 }
 
 async function testBytesAsync() {
-  const source = from('hello-async');
-  const data = await bytes(source);
+  const data = await bytes(from('hello-async'));
   assert.deepStrictEqual(data, new TextEncoder().encode('hello-async'));
 }
 
 async function testBytesAsyncLimit() {
-  const source = from('hello world');
   await assert.rejects(
-    () => bytes(source, { limit: 3 }),
+    () => bytes(from('hello world'), { limit: 3 }),
     { name: 'RangeError' },
   );
 }
 
 async function testBytesAsyncAbort() {
   const ac = new AbortController();
   ac.abort();
-  const source = from('data');
   await assert.rejects(
-    () => bytes(source, { signal: ac.signal }),
-    (err) => err.name === 'AbortError',
+    () => bytes(from('data'), { signal: ac.signal }),
+    { name: 'AbortError' },
   );
 }
 
 async function testBytesEmpty() {
-  const source = from([]);
-  const data = await bytes(source);
+  const data = await bytes(from([]));
+  assert.ok(data instanceof Uint8Array);
   assert.strictEqual(data.byteLength, 0);
 }
 
@@ -69,17 +64,15 @@ async function testBytesEmpty() {
 // =============================================================================
 
 async function testArrayBufferSyncBasic() {
-  const source = fromSync(new Uint8Array([1, 2, 3]));
-  const ab = arrayBufferSync(source);
+  const ab = arrayBufferSync(fromSync(new Uint8Array([1, 2, 3])));
   assert.ok(ab instanceof ArrayBuffer);
   assert.strictEqual(ab.byteLength, 3);
   const view = new Uint8Array(ab);
   assert.deepStrictEqual(view, new Uint8Array([1, 2, 3]));
 }
 
 async function testArrayBufferAsync() {
-  const source = from(new Uint8Array([10, 20, 30]));
-  const ab = await arrayBuffer(source);
+  const ab = await arrayBuffer(from(new Uint8Array([10, 20, 30])));
   assert.ok(ab instanceof ArrayBuffer);
   assert.strictEqual(ab.byteLength, 3);
   const view = new Uint8Array(ab);
@@ -96,8 +89,7 @@ async function testArraySyncBasic() {
     yield new Uint8Array([2]);
     yield new Uint8Array([3]);
   }
-  const source = fromSync(gen());
-  const chunks = arraySync(source);
+  const chunks = arraySync(fromSync(gen()));
   assert.strictEqual(chunks.length, 3);
   assert.deepStrictEqual(chunks[0], new Uint8Array([1]));
   assert.deepStrictEqual(chunks[1], new Uint8Array([2]));

test/parallel/test-stream-iter-consumers-merge.js

@@ -39,10 +39,7 @@ async function testMergeTwoSources() {
 }
 
 async function testMergeSingleSource() {
-  const source = from('only-one');
-  const merged = merge(source);
-
-  const data = await text(merged);
+  const data = await text(merge(from('only-one')));
   assert.strictEqual(data, 'only-one');
 }
 
@@ -59,8 +56,7 @@ async function testMergeWithAbortSignal() {
   const ac = new AbortController();
   ac.abort();
 
-  const source = from('data');
-  const merged = merge(source, { signal: ac.signal });
+  const merged = merge(from('data'), { signal: ac.signal });
 
   await assert.rejects(
     async () => {
@@ -69,7 +65,7 @@ async function testMergeWithAbortSignal() {
         assert.fail('Should not reach here');
       }
     },
-    (err) => err.name === 'AbortError',
+    { name: 'AbortError' },
   );
 }
 
@@ -91,10 +87,11 @@ async function testMergeSyncSources() {
 
 async function testMergeSourceError() {
   async function* goodSource() {
-    yield [new TextEncoder().encode('a')];
+    const enc = new TextEncoder();
+    yield [enc.encode('a')];
     // Slow so the bad source errors first
     await new Promise((r) => setTimeout(r, 50));
-    yield [new TextEncoder().encode('b')];
+    yield [enc.encode('b')];
   }
 
   async function* badSource() {
@@ -129,17 +126,19 @@ async function testMergeConsumerBreak() {
   for await (const _ of merge(source1(), source2())) {
     break; // Break after first batch
   }
-  await new Promise((r) => setTimeout(r, 10));
-  // At least one source should be cleaned up
-  assert.strictEqual(source1Return || source2Return, true);
+  // Give async cleanup a tick to complete
+  await new Promise(setImmediate);
+  // Both sources should be cleaned up
+  assert.strictEqual(source1Return && source2Return, true);
 }
 
 async function testMergeSignalMidIteration() {
   const ac = new AbortController();
   async function* slowSource() {
-    yield [new TextEncoder().encode('a')];
+    const enc = new TextEncoder();
+    yield [enc.encode('a')];
     await new Promise((r) => setTimeout(r, 100));
-    yield [new TextEncoder().encode('b')];
+    yield [enc.encode('b')];
   }
   const merged = merge(slowSource(), { signal: ac.signal });
   const iter = merged[Symbol.asyncIterator]();

test/parallel/test-stream-iter-consumers-tap.js

@@ -4,11 +4,15 @@
 const common = require('../common');
 const assert = require('assert');
 const {
+  from,
+  fromSync,
   pull,
+  pullSync,
   push,
   tap,
   tapSync,
   text,
+  textSync,
 } = require('stream/iter');
 
 // =============================================================================
@@ -77,8 +81,50 @@ async function testTapInPipeline() {
   assert.strictEqual(seen[0], 'hello');
 }
 
+// Tap callback error propagates through async pipeline
+async function testTapAsyncErrorPropagation() {
+  const badTap = tap(() => { throw new Error('tap error'); });
+  await assert.rejects(async () => {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of pull(from('hello'), badTap)) { /* consume */ }
+  }, { message: 'tap error' });
+}
+
+// TapSync callback error propagates through sync pipeline
+function testTapSyncErrorPropagation() {
+  const badTap = tapSync(() => { throw new Error('tapSync error'); });
+  assert.throws(() => {
+    // eslint-disable-next-line no-unused-vars
+    for (const _ of pullSync(fromSync('hello'), badTap)) { /* consume */ }
+  }, { message: 'tapSync error' });
+}
+
+// TapSync in a pullSync pipeline passes through data and flush
+function testTapSyncInPipeline() {
+  const seen = [];
+  let sawFlush = false;
+  const observer = tapSync((chunks) => {
+    if (chunks === null) {
+      sawFlush = true;
+    } else {
+      for (const chunk of chunks) {
+        seen.push(new TextDecoder().decode(chunk));
+      }
+    }
+  });
+
+  const data = textSync(pullSync(fromSync('hello'), observer));
+  assert.strictEqual(data, 'hello');
+  assert.strictEqual(seen.length, 1);
+  assert.strictEqual(seen[0], 'hello');
+  assert.strictEqual(sawFlush, true);
+}
+
 Promise.all([
   testTapSync(),
   testTapAsync(),
   testTapInPipeline(),
+  testTapAsyncErrorPropagation(),
+  testTapSyncErrorPropagation(),
+  testTapSyncInPipeline(),
 ]).then(common.mustCall());