Function fromWritable

lib/internal/streams/iter/classic.js:424–737

* Create a stream/iter Writer adapter from a classic Writable (or duck-type). * * Duck-type requirements: write() and on()/off() methods. * Falls back to sensible defaults for missing properties like * writableHighWaterMark, writableLength, writableObjectMode. * @param {object} writable - A cla

(writable, options = kNullPrototype)

 * @returns {object} A stream/iter Writer adapter.
 */
function fromWritable(writable, options = kNullPrototype) {
  if (writable == null ||
      typeof writable.write !== 'function' ||
      typeof writable.on !== 'function') {
    throw new ERR_INVALID_ARG_TYPE('writable', 'Writable', writable);
  }

  validateObject(options, 'options');
  const {
    backpressure = 'strict',
  } = options;
  validateBackpressure(backpressure);

  // The Writer interface is bytes-only. Object-mode Writables expect
  // arbitrary JS values, which is incompatible.
  if (writable.writableObjectMode) {
    throw new ERR_INVALID_STATE(
      'Cannot create a stream/iter Writer from an object-mode Writable');
  }

  // drop-oldest is not supported for classic stream.Writable. The
  // Writable's internal buffer stores individual { chunk, encoding,
  // callback } entries with no concept of batch boundaries. A writev()
  // call fans out into N separate buffer entries, so a subsequent
  // drop-oldest eviction could partially tear apart an earlier atomic
  // writev batch. The PushWriter avoids this because writev occupies a
  // single slot. Supporting drop-oldest here would require either
  // accepting partial writev eviction or adding batch tracking to the
  // buffer -- neither is acceptable without a deeper rework of Writable
  // internals.
  if (backpressure === 'drop-oldest') {
    throw new ERR_INVALID_ARG_VALUE('options.backpressure', backpressure,
                                    'drop-oldest is not supported for classic stream.Writable');
  }

  // Return cached adapter if available. Backpressure policy changes writer
  // behavior, so cache one adapter per policy.
  let cachedByBackpressure = fromWritableCache.get(writable);
  if (cachedByBackpressure !== undefined) {
    const cached = cachedByBackpressure.get(backpressure);
    if (cached !== undefined) return cached;
  } else {
    cachedByBackpressure = new SafeMap();
    fromWritableCache.set(writable, cachedByBackpressure);
  }

  // Fall back to sensible defaults for duck-typed streams that may not
  // expose the full stream.Writable property set.
  const hwm = writable.writableHighWaterMark ?? 16384;
  let totalBytes = 0;

  // Waiters pending on backpressure resolution (block policy only).
  // Multiple un-awaited writes can each add a waiter, so this must be
  // a list. A single persistent 'drain' listener and 'error' listener
  // (installed once lazily) resolve or reject all waiters to avoid
  // accumulating per-write listeners on the stream.
  let waiters = [];
  let listenersInstalled = false;