fix(race): ignore latter sources after first complete or error (#4809)

cartant · benlesh · commit f31c3df01b52 · 2019-06-04T15:28:28.000-07:00
* test(race): add failing tests * fix(race): ignore sources after first complete/error Closes #4808 BREAKING CHANGE: `race()` will no longer subscribe to subsequent observables if a provided source synchronously errors or completes. This means side effects that might have occurred during subscription in those rare cases will no longer occur.
diff --git a/spec/operators/race-spec.ts b/spec/operators/race-spec.ts
@@ -1,7 +1,7 @@
 import { expect } from 'chai';
 import * as sinon from 'sinon';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { NEVER, of, race as staticRace, timer, defer, Observable } from 'rxjs';
+import { EMPTY, NEVER, of, race as staticRace, timer, defer, Observable, throwError } from 'rxjs';
 import { race, mergeMap, map, finalize, startWith } from 'rxjs/operators';
 
 /** @test {race} */
@@ -184,6 +184,28 @@ describe('race operator', () => {
     expect(onSubscribe.called).to.be.false;
   });
 
+  it('should ignore latter observables if a former one completes immediately', () => {
+    const onComplete = sinon.spy();
+    const onSubscribe = sinon.spy();
+    const e1 = EMPTY; // Wins the race
+    const e2 = defer(onSubscribe); // Should be ignored
+
+    e1.pipe(race(e2)).subscribe({ complete: onComplete });
+    expect(onComplete.calledWithExactly()).to.be.true;
+    expect(onSubscribe.called).to.be.false;
+  });
+
+  it('should ignore latter observables if a former one errors immediately', () => {
+    const onError = sinon.spy();
+    const onSubscribe = sinon.spy();
+    const e1 = throwError('kaboom'); // Wins the race
+    const e2 = defer(onSubscribe); // Should be ignored
+
+    e1.pipe(race(e2)).subscribe({ error: onError });
+    expect(onError.calledWithExactly('kaboom')).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();
diff --git a/src/internal/observable/race.ts b/src/internal/observable/race.ts
@@ -137,4 +137,14 @@ export class RaceSubscriber<T> extends OuterSubscriber<T, T> {
 
     this.destination.next(innerValue);
   }
+
+  notifyComplete(innerSub: InnerSubscriber<T, T>): void {
+    this.hasFirst = true;
+    super.notifyComplete(innerSub);
+  }
+
+  notifyError(error: any, innerSub: InnerSubscriber<T, T>): void {
+    this.hasFirst = true;
+    super.notifyError(error, innerSub);
+  }
 }

Original file line number	Diff line number	Diff line change
`@@ -137,4 +137,14 @@ export class RaceSubscriber<T> extends OuterSubscriber<T, T> {`
`137`	`137`
`138`	`138`	`this.destination.next(innerValue);`
`139`	`139`	`}`
	`140`	`+`
	`141`	`+ notifyComplete(innerSub: InnerSubscriber<T, T>): void {`
	`142`	`+ this.hasFirst = true;`
	`143`	`+ super.notifyComplete(innerSub);`
	`144`	`+ }`
	`145`	`+`
	`146`	`+ notifyError(error: any, innerSub: InnerSubscriber<T, T>): void {`
	`147`	`+ this.hasFirst = true;`
	`148`	`+ super.notifyError(error, innerSub);`
	`149`	`+ }`
`140`	`150`	`}`