fix(debounceTime): synchronous reentrancy of debounceTime no longer s…

…wallows the second value (#3218)

fixes #2748

spec/operators/debounce-spec.ts

@@ -399,4 +399,20 @@ describe('Observable.prototype.debounce', () => {
         done(new Error('should not be called'));
       });
   });
+
+  it('should debounce correctly when synchronously reentered', () => {
+    const results = [];
+    const source = new Rx.Subject();
+
+    source.debounce(() => Observable.of(null)).subscribe(value => {
+      results.push(value);
+
+      if (value === 1) {
+        source.next(2);
+      }
+    });
+    source.next(1);
+
+    expect(results).to.deep.equal([1, 2]);
+  });
 });

spec/operators/debounceTime-spec.ts

@@ -1,5 +1,7 @@
+import { expect } from 'chai';
 import * as Rx from '../../src/Rx';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
+import { VirtualTimeScheduler } from '../../src/scheduler/VirtualTimeScheduler';
 
 declare const { asDiagram };
 declare const hot: typeof marbleTestingSignature.hot;
@@ -153,4 +155,22 @@ describe('Observable.prototype.debounceTime', () => {
     expectObservable(e1.debounceTime(40, rxTestScheduler)).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should debounce correctly when synchronously reentered', () => {
+    const results = [];
+    const source = new Rx.Subject();
+    const scheduler = new VirtualTimeScheduler();
+
+    source.debounceTime(0, scheduler).subscribe(value => {
+      results.push(value);
+
+      if (value === 1) {
+        source.next(2);
+      }
+    });
+    source.next(1);
+    scheduler.flush();
+
+    expect(results).to.deep.equal([1, 2]);
+  });
 });

src/operators/debounce.ts

@@ -129,6 +129,11 @@ class DebounceSubscriber<T, R> extends OuterSubscriber<T, R> {
         subscription.unsubscribe();
         this.remove(subscription);
       }
+      // This must be done *before* passing the value
+      // along to the destination because it's possible for
+      // the value to synchronously re-enter this operator
+      // recursively if the duration selector Observable
+      // emits synchronously
       this.value = null;
       this.hasValue = false;
       super._next(value);

src/operators/debounceTime.ts

@@ -97,9 +97,15 @@ class DebounceTimeSubscriber<T> extends Subscriber<T> {
     this.clearDebounce();
 
     if (this.hasValue) {
-      this.destination.next(this.lastValue);
+      const { lastValue } = this;
+      // This must be done *before* passing the value
+      // along to the destination because it's possible for
+      // the value to synchronously re-enter this operator
+      // recursively when scheduled with things like
+      // VirtualScheduler/TestScheduler.
       this.lastValue = null;
       this.hasValue = false;
+      this.destination.next(lastValue);
     }
   }