feat: Add debounceSync.

Author: cartant

source/operators/debounceSync-spec.ts

@@ -0,0 +1,49 @@
+/**
+ * @license Use of this source code is governed by an MIT-style license that
+ * can be found in the LICENSE file at https://github.com/cartant/rxjs-etc
+ */
+/*tslint:disable:no-unused-expression rxjs-no-ignored-subscription*/
+
+import { marbles } from "rxjs-marbles";
+import { debounceSync } from "./debounceSync";
+
+describe("debounceSync", () => {
+  it(
+    "should debounce synchronous values",
+    marbles((m) => {
+      const source = m.cold("   (ab)-(cd)---(ef)----|");
+      const sourceSubs = "      ^-------------------!";
+      const expected = m.cold(" b----d------f-------|");
+
+      const destination = source.pipe(debounceSync());
+      m.expect(destination).toBeObservable(expected);
+      m.expect(source).toHaveSubscriptions(sourceSubs);
+    })
+  );
+
+  it(
+    "should debounce values with synchronous completion",
+    marbles((m) => {
+      const source = m.cold("   (ab)-(cd)---(ef|)");
+      const sourceSubs = "      ^-----------!";
+      const expected = m.cold(" b----d------(f|)");
+
+      const destination = source.pipe(debounceSync());
+      m.expect(destination).toBeObservable(expected);
+      m.expect(source).toHaveSubscriptions(sourceSubs);
+    })
+  );
+
+  it(
+    "should debounce synchronous streams",
+    marbles((m) => {
+      const source = m.cold("   (abcdef|)");
+      const sourceSubs = "      (^!)";
+      const expected = m.cold(" (f|)");
+
+      const destination = source.pipe(debounceSync());
+      m.expect(destination).toBeObservable(expected);
+      m.expect(source).toHaveSubscriptions(sourceSubs);
+    })
+  );
+});

source/operators/debounceSync.ts

@@ -0,0 +1,42 @@
+/**
+ * @license Use of this source code is governed by an MIT-style license that
+ * can be found in the LICENSE file at https://github.com/cartant/rxjs-etc
+ */
+
+import {
+  asapScheduler,
+  MonoTypeOperatorFunction,
+  Observable,
+  Subscription,
+} from "rxjs";
+
+export function debounceSync<T>(): MonoTypeOperatorFunction<T> {
+  return (source) =>
+    new Observable<T>((observer) => {
+      let actionSubscription: Subscription | undefined;
+      let actionValue: T | undefined;
+      const rootSubscription = new Subscription();
+      rootSubscription.add(
+        source.subscribe({
+          complete: () => {
+            if (actionSubscription) {
+              observer.next(actionValue);
+            }
+            observer.complete();
+          },
+          error: (error) => observer.error(error),
+          next: (value) => {
+            actionValue = value;
+            if (!actionSubscription) {
+              actionSubscription = asapScheduler.schedule(() => {
+                observer.next(actionValue);
+                actionSubscription = undefined;
+              });
+              rootSubscription.add(actionSubscription);
+            }
+          },
+        })
+      );
+      return rootSubscription;
+    });
+}

source/operators/index.ts

@@ -9,6 +9,7 @@ export * from "./bufferRecent";
 export * from "./concatIfEmpty";
 export * from "./concatTap";
 export * from "./debounceAfter";
+export * from "./debounceSync";
 export * from "./debounceTimeSubsequent";
 export * from "./defaultObservableIfEmpty";
 export * from "./deferFinalize";