feat(transducers-async): add SyncOpts, update sync()

- add tests

Author: postspectacular

Diff for: packages/transducers-async/src/merge.ts

@@ -16,8 +16,7 @@ export async function* merge<T>(
 	let n = iters.length;
 	const $remove = (id: number) => {
 		iters.splice(id, 1);
-		n--;
-		if (!n) return true;
+		if (!--n) return true;
 		for (let i = id; i < n; i++) iters[i].id--;
 	};
 	// array of in-flight promises

Diff for: packages/transducers-async/src/sync.ts

@@ -4,6 +4,26 @@ export type SyncSources<T extends Record<NumOrString, any>> = {
 	[id in keyof T]: AsyncIterable<T[id]>;
 };
 
+export interface SyncOpts {
+	/**
+	 * If true, {@link sync} waits for new values from *all* remaining inputs
+	 * before a new tuple is produced. If false, that synchronization only
+	 * happens for the very first tuple and afterwards any changed input will
+	 * trigger a tuple update.
+	 *
+	 * @defaultValue false
+	 */
+	reset: boolean;
+	/**
+	 * Only used if {@link SyncOpts.reset} is disabled. If true (default: false)
+	 * *no* input synchronization (waiting for values) is applied and
+	 * {@link sync} will emit potentially partially populated tuple objects for
+	 * each received input value. However, as with the default behavior, tuples
+	 * will retain the most recent consumed value from other inputs.
+	 */
+	mergeOnly: boolean;
+}
+
 /**
  * Async iterator version of [thi.ng/rstream's sync()
  * construct](https://docs.thi.ng/umbrella/rstream/functions/sync.html).
@@ -12,10 +32,20 @@ export type SyncSources<T extends Record<NumOrString, any>> = {
  * Also see {@link merge} for an alternative way of merging.
  *
  * @param src
+ * @param opts
  */
+export function sync<T extends Record<NumOrString, any>>(
+	src: SyncSources<T>,
+	opts?: Partial<SyncOpts> & { mergeOnly: true }
+): AsyncIterableIterator<Partial<T>>;
+export function sync<T extends Record<NumOrString, any>>(
+	src: SyncSources<T>,
+	opts?: Partial<SyncOpts>
+): AsyncIterableIterator<T>;
 export async function* sync<T extends Record<NumOrString, any>>(
-	src: SyncSources<T>
-): AsyncIterableIterator<T> {
+	src: SyncSources<T>,
+	opts?: Partial<SyncOpts>
+) {
 	let iters = <{ key: keyof T; id: number; iter: AsyncIterator<any> }[]>(
 		Object.entries(src).map(([key, v], id) => ({
 			key,
@@ -26,38 +56,57 @@ export async function* sync<T extends Record<NumOrString, any>>(
 	let n = iters.length;
 	const $remove = (id: number) => {
 		iters.splice(id, 1);
-		n--;
-		if (!n) return true;
+		if (!--n) return true;
 		for (let i = id; i < n; i++) iters[i].id--;
 	};
-	// wait for all sources
-	const initial = await Promise.all(iters.map(({ iter }) => iter.next()));
-	// keep active iterators only, update successive IDs
-	for (let i = 0; i < n; ) {
-		if (initial[i].done) {
-			initial.splice(i, 1);
-			if ($remove(i)) return;
-		} else i++;
-	}
-	// build initial tuple
-	const tuple = initial.reduce(
-		(acc, x, i) => ((acc[iters[i].key] = x.value), acc),
-		<T>{}
-	);
-	yield { ...tuple };
-	// array of in-flight promises
-	const promises = iters.map((iter) =>
-		iter.iter.next().then((res) => ({ iter, res }))
-	);
-	while (true) {
-		const { iter, res } = await Promise.any(promises);
-		if (res.done) {
-			promises.splice(iter.id, 1);
-			if ($remove(iter.id)) return;
+	const $initial = async () => {
+		// wait for all sources
+		const res = await Promise.all(iters.map(({ iter }) => iter.next()));
+		// keep active iterators only, update successive IDs
+		for (let i = 0; i < n; ) {
+			if (res[i].done) {
+				res.splice(i, 1);
+				if ($remove(i)) return;
+			} else i++;
+		}
+		// build tuple
+		return res.reduce(
+			(acc, x, i) => ((acc[iters[i].key] = x.value), acc),
+			<T>{}
+		);
+	};
+	if (opts?.reset) {
+		let tuple: T | undefined;
+		let curr: T | undefined;
+		while ((curr = await $initial())) {
+			tuple = { ...tuple, ...curr };
+			yield tuple;
+		}
+	} else {
+		let tuple: T | undefined;
+		if (opts?.mergeOnly) {
+			tuple = <T>{};
 		} else {
-			tuple[iter.key] = res.value;
+			tuple = await $initial();
+			if (!tuple) return;
 			yield { ...tuple };
-			promises[iter.id] = iter.iter.next().then((res) => ({ res, iter }));
+		}
+		// array of in-flight promises
+		const promises = iters.map((iter) =>
+			iter.iter.next().then((res) => ({ iter, res }))
+		);
+		while (true) {
+			const { iter, res } = await Promise.any(promises);
+			if (res.done) {
+				promises.splice(iter.id, 1);
+				if ($remove(iter.id)) return;
+			} else {
+				tuple[iter.key] = res.value;
+				yield { ...tuple };
+				promises[iter.id] = iter.iter
+					.next()
+					.then((res) => ({ res, iter }));
+			}
 		}
 	}
 }

Diff for: packages/transducers-async/test/main.test.ts

@@ -240,6 +240,48 @@ test(
 	{ retry: 5 }
 );
 
+test("sync (reset)", async (done) => {
+	expect(
+		await push(
+			sync(
+				{
+					a: repeatedly((i) => i, 2),
+					b: repeatedly((i) => i, 4),
+				},
+				{ reset: true }
+			)
+		)
+	).toEqual([
+		{ a: 0, b: 0 },
+		{ a: 1, b: 1 },
+		{ a: 1, b: 2 },
+		{ a: 1, b: 3 },
+	]);
+	done();
+});
+
+test("sync (merge only)", async (done) => {
+	expect(
+		await push(
+			sync(
+				{
+					a: repeatedly((i) => i, 2),
+					b: repeatedly((i) => i, 4),
+				},
+				{ mergeOnly: true }
+			)
+		)
+	).toEqual([
+		{ a: 0 },
+		{ a: 0, b: 0 },
+		{ a: 1, b: 0 },
+		{ a: 1, b: 1 },
+		{ a: 1, b: 2 },
+		{ a: 1, b: 3 },
+	]);
+	done();
+});
+
 test("take", async (done) => {
 	expect(
 		await transduce(