fix(race): unsubscribe raced observables with immediate scheduler (#2158

)

Fixed the bug where racing an Observable that emits immediately,
with default scheduler, leaves other raced Observables unsubscribed
forever. Now immediate emit makes the race ignore following Observables.

spec/operators/race-spec.ts

@@ -1,4 +1,5 @@
 import {expect} from 'chai';
+import * as sinon from 'sinon';
 import * as Rx from '../../dist/cjs/Rx';
 declare const {hot, cold, expectObservable, expectSubscriptions};
 
@@ -171,4 +172,39 @@ describe('Observable.prototype.race', () => {
       expect(x).to.be.true;
     }, done, done);
   });
+
+  it('should ignore latter observables if a former one emits immediately', () => {
+    const onNext = sinon.spy();
+    const onSubscribe = sinon.spy();
+    const e1 = Observable.of('a'); // Wins the race
+    const e2 = Observable.defer(onSubscribe); // Should be ignored
+
+    e1.race(e2).subscribe(onNext);
+    expect(onNext.calledWithExactly('a')).to.be.true;
+    expect(onSubscribe.called).to.be.false;
+  });
+
+  it('should unsubscribe former observables if a latter one emits immediately', () => {
+    const onNext = sinon.spy();
+    const onUnsubscribe = sinon.spy();
+    const e1 = Observable.never<string>().finally(onUnsubscribe); // Should be unsubscribed
+    const e2 = Observable.of('b'); // Wins the race
+
+    e1.race(e2).subscribe(onNext);
+    expect(onNext.calledWithExactly('b')).to.be.true;
+    expect(onUnsubscribe.calledOnce).to.be.true;
+  });
+
+  it('should unsubscribe from immediately emitting observable on unsubscription', () => {
+    const onNext = sinon.spy();
+    const onUnsubscribe = sinon.spy();
+    const e1 = Observable.never<string>().startWith('a').finally(onUnsubscribe); // Wins the race
+    const e2 = Observable.never<string>(); // Loses the race
+
+    const subscription = e1.race(e2).subscribe(onNext);
+    expect(onNext.calledWithExactly('a')).to.be.true;
+    expect(onUnsubscribe.called).to.be.false;
+    subscription.unsubscribe();
+    expect(onUnsubscribe.calledOnce).to.be.true;
+  });
 });

src/operator/race.ts

@@ -85,14 +85,14 @@ export class RaceSubscriber<T> extends OuterSubscriber<T, T> {
     if (len === 0) {
       this.destination.complete();
     } else {
-      for (let i = 0; i < len; i++) {
+      for (let i = 0; i < len && !this.hasFirst; i++) {
         let observable = observables[i];
         let subscription = subscribeToResult(this, observable, observable, i);
 
         if (this.subscriptions) {
           this.subscriptions.push(subscription);
-          this.add(subscription);
         }
+        this.add(subscription);
       }
       this.observables = null;
     }