Function merge

lib/internal/streams/iter/consumers.js:395–514

* Merge multiple async iterables by yielding values in temporal order. * @param {...(AsyncIterable |object)} args * @returns {AsyncIterable }

(...args)

 * @returns {AsyncIterable<Uint8Array[]>}
 */
function merge(...args) {
  let sources;
  let options;

  if (args.length > 0 && isMergeOptions(args[args.length - 1])) {
    options = args[args.length - 1];
    sources = ArrayPrototypeSlice(args, 0, -1);
  } else {
    sources = args;
  }

  if (options?.signal !== undefined) {
    validateAbortSignal(options.signal, 'options.signal');
  }

  // Normalize each source via from()
  const normalized = ArrayPrototypeMap(sources, (source) => from(source));

  return {
    __proto__: null,
    async *[SymbolAsyncIterator]() {
      const signal = options?.signal;

      signal?.throwIfAborted();

      if (normalized.length === 0) return;

      if (normalized.length === 1) {
        for await (const batch of normalized[0]) {
          signal?.throwIfAborted();
          yield batch;
        }
        return;
      }

      // Multiple sources - use a ready queue so that batches that settle
      // between consumer pulls are drained synchronously without an extra
      // async tick per batch. Each source has at most one pending .next()
      // at a time. Every batch from every source is preserved.
      const ready = [];
      let activeCount = normalized.length;
      let waitResolve = null;

      // Called when a source's .next() settles. Pushes the result into
      // the ready queue and wakes the consumer if it's waiting.
      const onSettled = (iterator, result) => {
        if (result.done) {
          activeCount--;
        } else {
          ArrayPrototypePush(ready, result.value);
          // Immediately request the next value from this source
          // (at most one pending .next() per source)
          PromisePrototypeThen(
            iterator.next(),
            (r) => onSettled(iterator, r),
            (err) => {
              ArrayPrototypePush(ready, { __proto__: null, error: err });
              if (waitResolve) {