Function createNodeWriter

samples/12-nodejs-interop.ts:76–199

* Create a Writer that pipes to a Node.js Writable stream. * * Features: * - Proper backpressure handling (respects drain events) * - Error propagation from Node.js stream to writer * - Proper close/fail handling

(writable: Writable)

 * - Proper close/fail handling
 */
function createNodeWriter(writable: Writable): Writer {
  let error: Error | null = null;
  let drainResolve: (() => void) | null = null;
  let closed = false;
  let byteCount = 0;

  // Listen for errors from the writable
  writable.on('error', (err) => {
    error = err;
    // Resolve any pending drain wait
    if (drainResolve) {
      drainResolve();
      drainResolve = null;
    }
  });

  // Helper to wait for drain, optionally cancellable via signal
  const waitForDrain = (signal?: AbortSignal): Promise<void> => {
    return new Promise((resolve, reject) => {
      if (signal?.aborted) {
        reject(signal.reason);
        return;
      }
      drainResolve = resolve;
      const onDrain = () => {
        signal?.removeEventListener('abort', onAbort);
        drainResolve = null;
        resolve();
      };
      const onAbort = () => {
        writable.removeListener('drain', onDrain);
        drainResolve = null;
        reject(signal!.reason);
      };
      writable.once('drain', onDrain);
      signal?.addEventListener('abort', onAbort, { once: true });
    });
  };

  return {
    get desiredSize(): number | null {
      return closed || error ? null : 16384; // Reasonable default
    },

    async write(chunk: Uint8Array | string, options?: WriteOptions): Promise<void> {
      if (error) throw error;
      if (closed) throw new Error('Writer is closed');

      const bytes = typeof chunk === 'string' ? textEncoder.encode(chunk) : chunk;
      const buffer = Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength);

      const canContinue = writable.write(buffer);
      byteCount += buffer.length;

      if (!canContinue) {
        await waitForDrain(options?.signal);
        if (error) throw error;
      }