fix(pairwise): make it recursion-proof (#4743)

spec/operators/pairwise-spec.ts

@@ -1,5 +1,7 @@
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { pairwise } from 'rxjs/operators';
+import { pairwise, take } from 'rxjs/operators';
+import { Subject } from 'rxjs';
+import { expect } from 'chai';
 
 declare function asDiagram(arg: string): Function;
 
@@ -103,4 +105,25 @@ describe('pairwise operator', () => {
     expectObservable(source).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should be recursively re-enterable', () => {
+    const results = new Array<[string, string]>();
+
+    const subject = new Subject<string>();
+
+    subject
+      .pipe(
+        pairwise(),
+        take(3)
+      )
+      .subscribe(pair => {
+        results.push(pair);
+        subject.next('c');
+      });
+
+    subject.next('a');
+    subject.next('b');
+
+    expect(results).to.deep.equal([['a', 'b'], ['b', 'c'], ['c', 'c']]);
+  });
 });

src/internal/operators/pairwise.ts

@@ -70,12 +70,18 @@ class PairwiseSubscriber<T> extends Subscriber<T> {
   }
 
   _next(value: T): void {
+    let pair: [T, T] | undefined;
+
     if (this.hasPrev) {
-      this.destination.next([this.prev, value]);
+      pair = [this.prev, value];
     } else {
       this.hasPrev = true;
     }
 
     this.prev = value;
+
+    if (pair) {
+      this.destination.next(pair);
+    }
   }
 }