Add a 'fastDebounce' operator, to be preferred over Observable.deboun…

…ceTime

Summary:
`Observable.debounceTime` is actually quite inefficient with its usage of `setInterval` / `clearInterval`: if you look at a profile, it will always clear and re-create an interval upon receiving a new event.

In contrast, our debounce implementation (like lodash's) re-uses a timer when possible and just resets its timestamp. When the timer fires, we'll create a new timer if necessary.

For very hot codepaths where we debounce things like editor events, every millisecond matters. When features like 'code highlight' debounce events from several streams, this can add up!

See #93 for the investigation.

Reviewed By: matthewwithanm

Differential Revision: D6096145

fbshipit-source-id: 3569e2ce1b7cfc9e693962362ff80583de75e7d5

modules/nuclide-commons/observable.js

@@ -28,6 +28,7 @@ import UniversalDisposable from './UniversalDisposable';
 import invariant from 'assert';
 import {Observable, ReplaySubject} from 'rxjs';
 import {setDifference} from './collection';
+import debounce from './debounce';
 
 /**
  * Splits a stream of strings on newlines.
@@ -358,6 +359,33 @@ export function completingSwitchMap<T, U>(
     });
 }
 
+/**
+ * RxJS's debounceTime is actually fairly inefficient:
+ * on each event, it always clears its interval and [creates a new one][1].
+ * Until this is fixed, this uses our debounce implementation which
+ * reuses a timeout and just sets a timestamp when possible.
+ *
+ * This may seem like a micro-optimization but we often use debounces
+ * for very hot events, like keypresses. Exceeding the frame budget can easily lead
+ * to increased key latency!
+ *
+ * [1]: https://github.com/ReactiveX/rxjs/blob/master/src/operators/debounceTime.ts#L106
+ */
+export function fastDebounce<T>(
+  delay: number,
+): (Observable<T>) => Observable<T> {
+  return (observable: Observable<T>) =>
+    Observable.create(observer => {
+      const debouncedNext = debounce((x: T) => observer.next(x), delay);
+      const subscription = observable.subscribe(
+        debouncedNext,
+        observer.error.bind(observer),
+        observer.complete.bind(observer),
+      );
+      return new UniversalDisposable(subscription, debouncedNext);
+    });
+}
+
 export const microtask = Observable.create(observer => {
   process.nextTick(() => {
     observer.next();

modules/nuclide-commons/spec/observable-spec.js

@@ -12,8 +12,11 @@
 
 import {
   bufferUntil,
-  diffSets,
   cacheWhileSubscribed,
+  completingSwitchMap,
+  concatLatest,
+  diffSets,
+  fastDebounce,
   macrotask,
   microtask,
   nextAnimationFrame,
@@ -22,8 +25,6 @@ import {
   takeWhileInclusive,
   throttle,
   toggle,
-  concatLatest,
-  completingSwitchMap,
 } from '../observable';
 import {Disposable} from 'event-kit';
 import {Observable, Subject} from 'rxjs';
@@ -517,6 +518,58 @@ describe('nuclide-commons/observable', () => {
     });
   });
 
+  describe('fastDebounce', () => {
+    it('debounces events', () => {
+      waitsForPromise(async () => {
+        let nextSpy: jasmine$Spy;
+        const originalCreate = Observable.create.bind(Observable);
+        // Spy on the created observer's next to ensure that we always cancel
+        // the last debounced timer on unsubscribe.
+        spyOn(Observable, 'create').andCallFake(callback => {
+          return originalCreate(observer => {
+            nextSpy = spyOn(observer, 'next').andCallThrough();
+            return callback(observer);
+          });
+        });
+
+        const subject = new Subject();
+        const promise = subject
+          .let(fastDebounce(10))
+          .toArray()
+          .toPromise();
+
+        subject.next(1);
+        subject.next(2);
+        advanceClock(20);
+
+        subject.next(3);
+        advanceClock(5);
+
+        subject.next(4);
+        advanceClock(15);
+
+        subject.next(5);
+        subject.complete();
+        advanceClock(20);
+
+        expect(await promise).toEqual([2, 4]);
+        expect(nextSpy.callCount).toBe(2);
+      });
+    });
+
+    it('passes errors through immediately', () => {
+      let caught = false;
+      Observable.throw(1)
+        .let(fastDebounce(10))
+        .subscribe({
+          error() {
+            caught = true;
+          },
+        });
+      expect(caught).toBe(true);
+    });
+  });
+
   describe('microtask', () => {
     it('is cancelable', () => {
       waitsForPromise(async () => {