node/test/parallel/test-quic-stream-writer-api.mjs

// Flags: --experimental-quic --experimental-stream-iter --no-warnings

// Test: writer API methods (WRIT-02, WRIT-03, WRIT-04, WRIT-06,
// WRIT-07, WRIT-08, WRIT-12, WRIT-15).
// Uses a single endpoint with multiple streams, one per test.

import { hasQuic, skip, mustCall } from '../common/index.mjs';
import * as assert from 'node:assert';
import { dump } from 'node:stream/iter';

if (!hasQuic) {
  skip('QUIC is not enabled');
}

const { listen, connect } = await import('../common/quic.mjs');
const { bytes } = await import('stream/iter');

const encoder = new TextEncoder();

const totalStreams = 6;
const serverResults = [];
const allDone = Promise.withResolvers();

const serverEndpoint = await listen(mustCall((serverSession) => {
  serverSession.onstream = mustCall(async (stream) => {
    const received = await bytes(stream);
    serverResults.push(received);
    stream.writer.endSync();
    await stream.closed;

    if (serverResults.length === totalStreams) {
      serverSession.close();
      allDone.resolve();
    }
  }, totalStreams);
}));

const clientSession = await connect(serverEndpoint.address);
await clientSession.opened;

// write() async
{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  await w.write(encoder.encode('async write'));
  const n = w.endSync();
  assert.strictEqual(n, 11);
  await dump(stream);
  await stream.closed;
}

// writevSync() vectored write
{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  const result = w.writevSync([
    encoder.encode('hello '),
    encoder.encode('writev'),
  ]);
  assert.strictEqual(result, true);
  const n = w.endSync();
  assert.strictEqual(n, 12);
  await dump(stream);
  await stream.closed;
}

// writev() async vectored write
{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  await w.writev([
    encoder.encode('async '),
    encoder.encode('writev'),
  ]);
  const n = w.endSync();
  assert.strictEqual(n, 12);
  await dump(stream);
  await stream.closed;
}

// end() async close
{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  w.writeSync(encoder.encode('end async'));
  const n = await w.end();
  assert.strictEqual(n, 9);
  await dump(stream);
  await stream.closed;
}

// Web IDL Writer argument conversion
{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  await w.write(42, null);
  await w.writev(new Set([true, { toString: () => 'object' }]));
  assert.throws(
    () => w.write(Symbol('invalid')),
    { code: 'ERR_INVALID_ARG_TYPE' },
  );
  assert.throws(
    () => w.write('invalid options', 1),
    { code: 'ERR_INVALID_ARG_TYPE' },
  );
  assert.strictEqual(w.endSync(), 12);
  // Writes after the writer is closed reject with a TypeError.
  await assert.rejects(w.write('closed'),
                       { code: 'ERR_INVALID_STATE', name: 'TypeError' });
  await assert.rejects(w.writev(['closed']),
                       { code: 'ERR_INVALID_STATE', name: 'TypeError' });
  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
  await stream.closed;
}

{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  // canWrite should be a boolean (false initially before flow
  // control window opens, or true if the window is already open).
  assert.strictEqual(typeof w.canWrite, 'boolean');
  assert.ok(w.canWrite !== null, `canWrite should not be null, got ${w.canWrite}`);
  // drainableProtocol should return null when canWrite is true (has capacity),
  // or a promise when canWrite is false (backpressured). Either way, it
  // should not throw.
  const { drainableProtocol: dp } = await import('stream/iter');
  const drain = w[dp]();
  assert.ok(drain === null || drain instanceof Promise);
  w.writeSync(encoder.encode('capacity'));
  w.endSync();
  await dump(stream);
  await stream.closed;
}

// Return null when errored.
{
  const stream = await clientSession.createBidirectionalStream();
  const w = stream.writer;
  const reason = null;
  w.fail(reason);
  // After fail, canWrite is null.
  assert.strictEqual(w.canWrite, null);
  // drainableProtocol returns null when errored.
  const { drainableProtocol: dp } = await import('stream/iter');
  assert.strictEqual(w[dp](), null);
  // endSync after fail returns -1 (errored).
  assert.strictEqual(w.endSync(), -1);
  // WriteSync after fail returns false.
  assert.strictEqual(w.writeSync(encoder.encode('x')), false);
  // Stored failure takes precedence over per-operation cancellation.
  const signal = AbortSignal.abort('operation cancelled');
  await assert.rejects(
    w.write(encoder.encode('x'), { signal }),
    (error) => error === reason);
  await assert.rejects(w.end({ signal }), (error) => error === reason);
  // Don't await stream.closed here — the reset stream may not trigger
  // server onstream (no data was sent before fail), so the server
  // won't count it. The stream is cleaned up when the session closes.
}

await allDone.promise;
await clientSession.close();
await serverEndpoint.close();

// Verify server received the right data.
const decoder = new TextDecoder();
assert.strictEqual(decoder.decode(serverResults[0]), 'async write');
assert.strictEqual(decoder.decode(serverResults[1]), 'hello writev');
assert.strictEqual(decoder.decode(serverResults[2]), 'async writev');
assert.strictEqual(decoder.decode(serverResults[3]), 'end async');
assert.strictEqual(decoder.decode(serverResults[4]), '42trueobject');
assert.strictEqual(decoder.decode(serverResults[5]), 'capacity');