Function duplex

lib/internal/streams/iter/duplex.js:27–133

* Create a pair of connected duplex channels for bidirectional communication. * @param {{ highWaterMark?: number, backpressure?: string, signal?: AbortSignal, * a?: object, b?: object }} [options] * @returns {[DuplexChannel, DuplexChannel]}

(options = { __proto__: null })

 * @returns {[DuplexChannel, DuplexChannel]}
 */
function duplex(options = { __proto__: null }) {
  validateObject(options, 'options');
  const { highWaterMark, backpressure, signal, a, b } = options;
  if (a !== undefined) {
    validateObject(a, 'options.a');
  }
  if (b !== undefined) {
    validateObject(b, 'options.b');
  }
  if (signal !== undefined) {
    validateAbortSignal(signal, 'options.signal');
  }

  // Channel A writes to B's readable (A->B direction).
  // Signal is NOT passed to push() -- we handle abort via close() below.
  const { writer: aWriter, readable: bReadable } = push({
    highWaterMark: a?.highWaterMark ?? highWaterMark,
    backpressure: a?.backpressure ?? backpressure,
  });

  // Channel B writes to A's readable (B->A direction)
  const { writer: bWriter, readable: aReadable } = push({
    highWaterMark: b?.highWaterMark ?? highWaterMark,
    backpressure: b?.backpressure ?? backpressure,
  });

  let aClosed = false;
  let bClosed = false;
  // Track active iterators so close() can call .return() on them
  let aReadableIterator = null;
  let bReadableIterator = null;

  const channelA = {
    __proto__: null,
    get writer() { return aWriter; },
    // Wrap readable to track the iterator for cleanup on close()
    get readable() {
      return {
        __proto__: null,
        [SymbolAsyncIterator]() {
          const iter = aReadable[SymbolAsyncIterator]();
          aReadableIterator = iter;
          return iter;
        },
      };
    },
    async close() {
      if (aClosed) return;
      aClosed = true;
      // End the writer (signals end-of-stream to B's readable)
      aWriter.endSync();
      // Stop iteration of this channel's readable
      if (aReadableIterator?.return) {
        await aReadableIterator.return();
        aReadableIterator = null;
      }
    },
    [SymbolAsyncDispose]() {