node/test/parallel/test-stream-iter-pipeto-writev.js

// Flags: --experimental-stream-iter
'use strict';

// Tests for pipeTo writev/writevSync paths and writeBatchAsyncFallback.

const common = require('../common');
const assert = require('assert');
const { setImmediate: setImmediatePromise } = require('timers/promises');
const { pipeTo, pipeToSync, push, text } = require('stream/iter');

// Multi-chunk batch with writevSync (sync success path)
async function testWritevSyncSuccess() {
  const batches = [];
  const writer = {
    write(chunk) {},
    writevSync(chunks) { batches.push(chunks); return true; },
    writev(chunks) { batches.push(chunks); },
    writeSync(chunk) { return true; },
    endSync() { return 0; },
  };
  // Source that yields multi-chunk batches
  async function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2]), new Uint8Array([3])];
    yield [new Uint8Array([4]), new Uint8Array([5])];
  }
  const total = await pipeTo(source(), writer);
  assert.ok(batches.length > 0);
  // writevSync was used for multi-chunk batches
  assert.ok(batches.some((b) => b.length > 1));
  assert.strictEqual(total, 5);
}

// Multi-chunk batch with writev async (no writevSync)
async function testWritevAsyncFallback() {
  const batches = [];
  const writer = {
    async writev(chunks) { batches.push(chunks); },
    async write(chunk) { batches.push([chunk]); },
    async end() {},
  };
  async function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2]), new Uint8Array([3])];
  }
  await pipeTo(source(), writer);
  assert.ok(batches.length > 0);
  assert.ok(batches.some((b) => b.length > 1));
}

// Multi-chunk batch with synchronous writev success (returns undefined)
async function testWritevSyncUndefinedSuccess() {
  const chunks = [];
  const writer = {
    write(chunk) {
      chunks.push(chunk);
    },
    writev(batch) {
      chunks.push(...batch);
    },
    end() {},
  };
  async function* source() {
    yield [new Uint8Array([65]), new Uint8Array([66])];
  }
  const total = await pipeTo(source(), writer);
  assert.strictEqual(total, 2);
  assert.strictEqual(Buffer.concat(chunks).toString(), 'AB');
}

// writevSync returns false — falls through to async writev
async function testWritevSyncFails() {
  const asyncCalls = [];
  const writer = {
    write() {},
    writevSync() { return false; },
    async writev(chunks) { asyncCalls.push(chunks); },
    writeSync() { return true; },
    endSync() { return 0; },
  };
  async function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2])];
  }
  await pipeTo(source(), writer);
  assert.strictEqual(asyncCalls.length, 1);
  assert.strictEqual(asyncCalls[0].length, 2);
}

// writeSync fails mid-batch — triggers writeBatchAsyncFallback
async function testWriteSyncFailsMidBatch() {
  const asyncWrites = [];
  const writer = {
    writeSync(chunk) {
      // Fail for chunk value 2 — always, including retries
      if (chunk[0] === 2) return false;
      return true;
    },
    async write(chunk) { asyncWrites.push(chunk); },
    async end() {},
  };
  // Single batch with 3 chunks
  async function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2]), new Uint8Array([3])];
  }
  const total = await pipeTo(source(), writer);
  // Chunk 1: writeSync succeeds
  // Chunk 2: writeSync fails → writeBatchAsyncFallback → write() called
  // Chunk 3: writeBatchAsyncFallback retries writeSync → succeeds
  assert.ok(asyncWrites.length >= 1);
  assert.deepStrictEqual(asyncWrites[0], new Uint8Array([2]));
  assert.strictEqual(total, 3);
}

// writeSync always fails — all chunks go through async
async function testWriteSyncAlwaysFails() {
  const asyncWrites = [];
  const writer = {
    writeSync() { return false; },
    async write(chunk) { asyncWrites.push(chunk); },
    async end() {},
  };
  async function* source() {
    yield [new Uint8Array([10]), new Uint8Array([20])];
  }
  const total = await pipeTo(source(), writer);
  assert.strictEqual(asyncWrites.length, 2);
  assert.strictEqual(total, 2);
}

// PushWriter block mode accepts sync writes even when returning false for
// backpressure. pipeTo must wait for drain, not retry the same write.
async function assertPushWriterBlockPipeTo(source, expected, expectedTotal) {
  const { writer, readable } = push({
    budget: 16384,
    backpressure: 'unbounded',
  });

  const pipe = pipeTo(source, writer);
  await setImmediatePromise();
  const data = await text(readable);
  const total = await pipe;

  assert.strictEqual(data, expected);
  assert.strictEqual(total, expectedTotal);
}

async function testPushWriterBlockSyncFalseAccepted() {
  await assertPushWriterBlockPipeTo((async function*() {
    yield [new Uint8Array([97])];
    yield [new Uint8Array([98])];
  })(), 'ab', 2);

  await assertPushWriterBlockPipeTo((async function*() {
    yield [new Uint8Array([97, 98])];
    yield [new Uint8Array([99]), new Uint8Array([100])];
  })(), 'abcd', 4);
}

async function testPipeToSyncPushWriterStrictFalseRejected() {
  const kChunk = new Uint8Array(16384);
  const { writer } = push({ budget: 16384 });

  // Pre-fill the buffer so it's at capacity
  writer.writeSync(kChunk);

  // pipeToSync should throw when writeSync returns false (budget exhausted)
  assert.throws(
    () => pipeToSync([kChunk], writer,
                     { preventClose: true, preventFail: true }),
    { code: 'ERR_OUT_OF_RANGE' },
  );
}

async function testPipeToSyncWritevFalseNotCounted() {
  const writer = {
    writevSync() { return false; },
    writeSync: common.mustNotCall(),
    endSync() { return 0; },
  };
  function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2])];
  }

  // pipeToSync throws when writevSync returns false (budget exhausted)
  assert.throws(
    () => pipeToSync(source(), writer),
    { code: 'ERR_OUT_OF_RANGE' },
  );
}

// pipeToSync with writevSync
async function testPipeToSyncWritev() {
  const batches = [];
  const writer = {
    writevSync(chunks) { batches.push(chunks); },
    writeSync(chunk) { return true; },
    endSync() { return 0; },
  };
  function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2]), new Uint8Array([3])];
    yield [new Uint8Array([4])];
  }
  pipeToSync(source(), writer);
  // Multi-chunk batch should have used writevSync
  assert.ok(batches.some((b) => b.length > 1));
}

// pipeToSync batches plain Uint8Array chunks for writevSync
async function testPipeToSyncPlainChunksWritev() {
  const batches = [];
  const writes = [];
  const writer = {
    writevSync(chunks) { batches.push(chunks); },
    writeSync(chunk) { writes.push(chunk); return true; },
    endSync() { return 0; },
  };
  function* source() {
    yield new Uint8Array([1]);
    yield new Uint8Array([2]);
    yield new Uint8Array([3]);
  }
  const total = pipeToSync(source(), writer);
  assert.strictEqual(total, 3);
  assert.strictEqual(batches.length, 1);
  assert.strictEqual(batches[0].length, 3);
  assert.strictEqual(writes.length, 0);
}

// pipeToSync with writer that has write() and writeSync() — writeSync preferred
async function testPipeToSyncWriteFallback() {
  const syncWrites = [];
  const writer = {
    writeSync(chunk) { syncWrites.push(chunk); return true; },
    write(chunk) { /* should not be called */ },
    endSync() { return 0; },
  };
  function* source() {
    yield [new Uint8Array([1]), new Uint8Array([2])];
  }
  pipeToSync(source(), writer);
  assert.strictEqual(syncWrites.length, 2);
}

Promise.all([
  testWritevSyncSuccess(),
  testWritevAsyncFallback(),
  testWritevSyncUndefinedSuccess(),
  testWritevSyncFails(),
  testWriteSyncFailsMidBatch(),
  testWriteSyncAlwaysFails(),
  testPushWriterBlockSyncFalseAccepted(),
  testPipeToSyncPushWriterStrictFalseRejected(),
  testPipeToSyncWritevFalseNotCounted(),
  testPipeToSyncWritev(),
  testPipeToSyncPlainChunksWritev(),
  testPipeToSyncWriteFallback(),
]).then(common.mustCall());