fix(timeout): no longer will timeout when receiving the first value s…

…ynchronously (#6865)

* fix(timeout): do not timeout if source emits synchronously when subscribed

Closes #6862

* chore: Improve comments

* chore: minor test cleanup

* chore: minor test cleanup

* chore: Fix broken test that I broke! lol

Co-authored-by: Ben Lesh <ben@benlesh.com>

spec/operators/timeout-spec.ts

@@ -1,8 +1,8 @@
 /** @prettier */
 import { expect } from 'chai';
-import { timeout, mergeMap, take } from 'rxjs/operators';
+import { timeout, mergeMap, take, concatWith } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
-import { TimeoutError, of, Observable } from 'rxjs';
+import { TimeoutError, of, Observable, NEVER } from 'rxjs';
 import { observableMatcher } from '../helpers/observableMatcher';
 
 /** @test {timeout} */
@@ -691,6 +691,15 @@ describe('timeout operator', () => {
         expectSubscriptions(inner.subscriptions).toBe([]);
       });
     });
+
+    it('should not timeout if source emits synchronously when subscribed', () => {
+      rxTestScheduler.run(({ expectObservable, time }) => {
+        const source = of('a').pipe(concatWith(NEVER));
+        const t = time('  ---|');
+        const expected = 'a---';
+        expectObservable(source.pipe(timeout({ first: new Date(t) }))).toBe(expected);
+      });
+    });
   });
 
   it('should stop listening to a synchronous observable when unsubscribed', () => {

src/internal/operators/timeout.ts

@@ -386,10 +386,12 @@ export function timeout<T, O extends ObservableInput<any>, M>(
     );
 
     // Intentionally terse code.
+    // If we've `seen` a value, that means the "first" clause was met already, if it existed.
+    //   it also means that a timer was already started for "each" (in the next handler above).
     // If `first` was provided, and it's a number, then use it.
     // If `first` was provided and it's not a number, it's a Date, and we get the difference between it and "now".
     // If `first` was not provided at all, then our first timer will be the value from `each`.
-    startTimer(first != null ? (typeof first === 'number' ? first : +first - scheduler!.now()) : each!);
+    !seen && startTimer(first != null ? (typeof first === 'number' ? first : +first - scheduler!.now()) : each!);
   });
 }