fix(first): will now only emit one value in recursive cases (#2100)

adds guard after first emitted value to prevent reentrant behavior

fixes #2098

Author: benlesh

spec/operators/first-spec.ts

@@ -42,6 +42,20 @@ describe('Observable.prototype.first', () => {
     expectSubscriptions(e1.subscriptions).toBe(sub);
   });
 
+  it('should only emit one value in recursive cases', () => {
+    const subject = new Rx.Subject<number>();
+    const results = [];
+
+    subject.first().subscribe(x => {
+      results.push(x);
+      subject.next(x + 1);
+    });
+
+    subject.next(0);
+
+    expect(results).to.deep.equal([0]);
+  });
+
   it('should propagate error from the source observable', () => {
     const e1 = hot('---^---#');
     const expected =  '----#';

src/operator/first.ts

@@ -85,6 +85,7 @@ class FirstOperator<T, R> implements Operator<T, R> {
 class FirstSubscriber<T, R> extends Subscriber<T> {
   private index: number = 0;
   private hasCompleted: boolean = false;
+  private _emitted: boolean = false;
 
   constructor(destination: Subscriber<R>,
               private predicate?: (value: T, index: number, source: Observable<T>) => boolean,
@@ -137,9 +138,12 @@ class FirstSubscriber<T, R> extends Subscriber<T> {
 
   private _emitFinal(value: any) {
     const destination = this.destination;
-    destination.next(value);
-    destination.complete();
-    this.hasCompleted = true;
+    if (!this._emitted) {
+      this._emitted = true;
+      destination.next(value);
+      destination.complete();
+      this.hasCompleted = true;
+    }
   }
 
   protected _complete(): void {