Function fromAsyncGen

lib/internal/streams/duplexify.js:211–256

(fn)

};

function fromAsyncGen(fn) {
  let { promise, resolve } = PromiseWithResolvers();
  const ac = new AbortController();
  const signal = ac.signal;
  const value = fn(async function*() {
    while (true) {
      const _promise = promise;
      promise = null;
      const { chunk, done, cb } = await _promise;
      process.nextTick(cb);
      if (done) return;
      if (signal.aborted)
        throw new AbortError(undefined, { cause: signal.reason });
      ({ promise, resolve } = PromiseWithResolvers());
      yield chunk;
    }
  }(), { signal });

  return {
    value,
    write(chunk, encoding, cb) {
      const _resolve = resolve;
      resolve = null;
      _resolve({ __proto__: null, chunk, done: false, cb });
    },
    final(cb) {
      const _resolve = resolve;
      resolve = null;
      _resolve({ __proto__: null, done: true, cb });
    },
    destroy(err, cb) {
      ac.abort(err);

      // If the source async iterator is waiting for the next write/final
      // signal, unblock it so the readable side can observe the abort and
      // finish destroying.
      if (resolve !== null) {
        const _resolve = resolve;
        resolve = null;
        _resolve({ __proto__: null, done: true, cb() {} });
      }

      cb(err);
    },
  };
}

function _duplexify(pair) {
  const r = pair.readable && typeof pair.readable.read !== 'function' ?