feat(transducers-async): add raf(), sidechain()

- add tests

Author: postspectacular

packages/transducers-async/package.json

@@ -137,6 +137,9 @@
 		"./push": {
 			"default": "./push.js"
 		},
+		"./raf": {
+			"default": "./raf.js"
+		},
 		"./range": {
 			"default": "./range.js"
 		},
@@ -149,6 +152,9 @@
 		"./run": {
 			"default": "./run.js"
 		},
+		"./sidechain": {
+			"default": "./sidechain.js"
+		},
 		"./step": {
 			"default": "./step.js"
 		},

packages/transducers-async/src/index.ts

@@ -11,18 +11,20 @@ export * from "./iterator.js";
 export * from "./map.js";
 export * from "./mapcat.js";
 export * from "./merge.js";
-export * from "./multiplex.js";
 export * from "./multiplex-obj.js";
+export * from "./multiplex.js";
 export * from "./partition.js";
 export * from "./push.js";
+export * from "./raf.js";
 export * from "./range.js";
 export * from "./reduce.js";
 export * from "./repeatedly.js";
 export * from "./run.js";
+export * from "./sidechain.js";
 export * from "./step.js";
 export * from "./sync.js";
 export * from "./take.js";
-export * from "./throttle.js";
 export * from "./throttle-time.js";
+export * from "./throttle.js";
 export * from "./transduce.js";
 export * from "./zip.js";

packages/transducers-async/src/raf.ts

@@ -0,0 +1,39 @@
+import type { Fn } from "@thi.ng/api";
+
+export interface RAFOpts {
+	/**
+	 * If true (default: false), passes the timestamps received
+	 * via `requestAnimationFrame()` as iterator values. If false, a simple
+	 * counter [0..∞) will be emitted.
+	 *
+	 * @defaultValue false
+	 */
+	timestamp: boolean;
+	/**
+	 * Only used if {@link RAFOpts.timestamp} is enabled. If given as
+	 * number, the value will be subtracted from all emitted timestamps. If this
+	 * option is set to true, the timestamps will be automatically zero-adjusted
+	 * such that the first emitted value will be zero. If undefined (default),
+	 * the browser supplied timestamps will be used as is.
+	 */
+	t0: number | boolean;
+}
+
+export async function* raf(opts?: Partial<RAFOpts>): AsyncGenerator<number> {
+	let frame = 0;
+	let t0 = opts?.t0 || 0;
+	while (true) {
+		let resolve: Fn<number, void>;
+		const promise = new Promise<number>(($resolve) => (resolve = $resolve));
+		requestAnimationFrame(resolve!);
+		let t = await promise;
+		if (opts?.timestamp) {
+			if (t0 === true) t0 = t;
+			if (t0) t -= t0;
+		} else {
+			t = frame++;
+		}
+		const cancel = yield t;
+		if (cancel === true) break;
+	}
+}

packages/transducers-async/src/sidechain.ts

@@ -0,0 +1,54 @@
+export interface SidechainOpts {
+	/**
+	 * If true (default), only emits the last received value when the sidechain
+	 * triggers. Otherwise buffers and emits *all* received values since the
+	 * last time the sidechain triggered.
+	 *
+	 * @defaultValue true
+	 */
+	lastOnly: boolean;
+}
+
+export function sidechain<T>(
+	src: AsyncIterable<T>,
+	side: AsyncIterable<boolean>,
+	opts: Partial<SidechainOpts> & { lastOnly: false }
+): AsyncIterableIterator<T[]>;
+export function sidechain<T>(
+	src: AsyncIterable<T>,
+	side: AsyncIterable<boolean>,
+	opts?: Partial<SidechainOpts>
+): AsyncIterableIterator<T>;
+export async function* sidechain<T>(
+	src: AsyncIterable<T>,
+	side: AsyncIterable<boolean>,
+	opts?: Partial<SidechainOpts>
+) {
+	const { lastOnly = true } = opts || {};
+	const $iter = src[Symbol.asyncIterator]();
+	const $side = side[Symbol.asyncIterator]();
+	const promises: Promise<[IteratorResult<any>, boolean?]>[] = [
+		$iter.next().then((res) => [res]),
+		$side.next().then((res) => [res, true]),
+	];
+	let buf: T[] = [];
+	while (true) {
+		const [res, side] = await Promise.any(promises);
+		if (res.done) return;
+		if (side) {
+			promises[1] = $side.next().then((res) => [res, true]);
+			if (!buf.length) continue;
+			if (lastOnly) {
+				yield buf[0];
+				buf.length = 0;
+			} else {
+				yield buf;
+				buf = [];
+			}
+		} else {
+			promises[0] = $iter.next().then((res) => [res]);
+			if (lastOnly) buf[0] = res.value;
+			else buf.push(res.value);
+		}
+	}
+}

packages/transducers-async/test/main.test.ts

@@ -16,6 +16,7 @@ import {
 	range,
 	repeatedly,
 	run,
+	sidechain,
 	step,
 	sync,
 	take,
@@ -197,6 +198,31 @@ test("run", async (done) => {
 	done();
 });
 
+test.only(
+	"sidechain",
+	async (done) => {
+		expect(
+			await push(
+				sidechain(
+					repeatedly((i) => i, 100, 10),
+					repeatedly(() => true, 3, 25)
+				)
+			)
+		).toEqual([0, 2, 4]);
+		expect(
+			await push(
+				sidechain(
+					repeatedly((i) => i, 100, 10),
+					repeatedly(() => true, 3, 25),
+					{ lastOnly: false }
+				)
+			)
+		).toEqual([[0], [1, 2], [3, 4]]);
+		done();
+	},
+	{ retry: 5 }
+);
+
 test("step", async (done) => {
 	expect(await step(mapcat(async (x) => [x, x]))(1)).toEqual([1, 1]);
 	expect(await step(mapcat(async (x) => [x]))(1)).toEqual(1);