feat(rstream): add debounce() sub & tests

packages/rstream/src/index.ts

@@ -22,6 +22,7 @@ export * from "./from/worker";
 
 export * from "./subs/asidechain";
 export * from "./subs/bisect";
+export * from "./subs/debounce";
 export * from "./subs/post-worker";
 export * from "./subs/resolve";
 export * from "./subs/sidechain-partition";

packages/rstream/src/subs/debounce.ts

@@ -0,0 +1,19 @@
+import { fromIterable } from "../from/iterable";
+import { metaStream, MetaStreamOpts } from "../metastream";
+
+/**
+ * Returns a subscription which ignores any intermediate inputs arriving
+ * faster than given `delay` time period.
+ *
+ * @example
+ * ```ts
+ *
+ * ```
+ *
+ * @param delay
+ */
+export const debounce = <T>(delay: number, opts?: Partial<MetaStreamOpts>) =>
+    metaStream((x: T) => fromIterable([x], { delay }), {
+        emitLast: true,
+        ...opts,
+    });

packages/rstream/test/debounce.ts

@@ -0,0 +1,35 @@
+import * as assert from "assert";
+import { debounce, fromIterable } from "../src/index";
+import { TIMEOUT } from "./config";
+
+describe("debounce", () => {
+    it("basic", (done) => {
+        const acc: number[] = [];
+        fromIterable([1, 2, 3], { delay: TIMEOUT })
+            .subscribe(debounce(TIMEOUT * 1.5))
+            .subscribe({
+                next(x) {
+                    acc.push(x);
+                },
+            });
+        setTimeout(() => {
+            assert.deepEqual(acc, [3]);
+            done();
+        }, TIMEOUT * 5);
+    });
+
+    it("no last", (done) => {
+        const acc: number[] = [];
+        fromIterable([1, 2, 3], { delay: TIMEOUT })
+            .subscribe(debounce(TIMEOUT * 1.5, { emitLast: false }))
+            .subscribe({
+                next(x) {
+                    acc.push(x);
+                },
+            });
+        setTimeout(() => {
+            assert.deepEqual(acc, []);
+            done();
+        }, TIMEOUT * 5);
+    });
+});