vadzim/mergeiterator

merges async iterators

mergeiterator

Merges async iterators.

Merges list of async or sync iterables into async one.

It accepts any iterable like arrays, generators, async generators or even promises which resolve to iterables. Any iterator can be infinite, including the list of iterables itself.

import merge from "mergeiterator"

async function DoIt() {
	for await (const v of merge([
		[1, 2, Promise.resolve(3)],
		Promise.resolve([4, 5]),
		(function*() {
			let i = 9
			while (true) {
				yield i++
				yield Promise.resolve(i++)
			}
		})(),
		(async function*() {
			yield 6
			yield await Promise.resolve(7)
			yield Promise.resolve(8)
		})(),
	])) {
		console.log(v)
	}
}

// 1 4 2 9 5 3 10 6 11 7 12 8 13 14 15 16 17 18 19 20 ...

This function guarantees, that if some value is yielded by some of iterables, then that value will be eventually yielded. This is basically about infinite iterables. It also guarantees that the order of values within the same iterable is preserved.

If some iterable yields a promise, its value will be used, not a promise itself.

If some iterable throws an error, that error will be redirected to a caller and other iterables will be closed.

Return values of iterables are discarded.

API

• merge
  • Parameters

merge

Merges async or sync iterables into async one.

Parameters

• sequences AnyIterable<AnyIterable<T>>

Returns AsyncGenerator<T, any, any>

mergeiterator's Issues

enable tests after generators fix

mergeiterator/test/mergeiterator.test.js

Line 439 in aec8e0e

if (+process.version.match(/\d+/)?.[0] >= 14) {

https://bugzilla.mozilla.org/show_bug.cgi?id=1610315

merge() does not propagate iteration end to the children iterators

Example code:

import delay from "delay";
import merge from "mergeiterator";

async function* iterable(name: string, dt: number) {
  try {
    for (let i = 0; ; i++) {
      console.log(`${name}: ${i}`);
      yield `${name}: ${i}`;
      await delay(dt);
    }
  } finally {
    console.log(`Exited ${name}`);
  }
}

async function* caller() {
  //yield* iterable("A", 900);
  yield* merge(iterable("A", 900));
}

async function main() {
  for await (const message of caller()) {
    if (message.includes("4")) {
      break;
    }

    console.log(`Received ${message}`);
  }

  console.log("Finishing");
  await delay(3000);
}

main().catch((e) => console.log(e));

In this example I do a "dummy" merging of 1 iterable for simplicity (but we can merge multiple, the effect persists). The output is:

A: 0
Received A: 0
A: 1
Received A: 1
A: 2
Received A: 2
A: 3
Received A: 3
A: 4
Finishing

Notice that finally {} block in iterable() function was never executed. But it should: replace the call to merge() with yield* iterable("A", 900); to see the correct output (with "Exited A"):

A: 0
Received A: 0
A: 1
Received A: 1
A: 2
Received A: 2
A: 3
Received A: 3
A: 4
Exited A
Finishing

How it works: both for-await and yield* instructions call the source iterator's return() method once the loop is over, and they propagate that signal further up the stack. I think merge() just doesn't do this.