Unexpected Timeout when using the timeout operator's first option with a BehaviorSubject or Observable piped through shareReplay #6862
Description
Describe the bug
When using the timeout operator's first option with a BehaviorSubject , the timeout will trigger even though the subscription has received a value from the intial subscribe call.
This example seems a bit contrived (why would we be interested in timing out on a value you know is already there?), but when timeout is used with a subject created through the shareReplay operator, I felt the behavior was strange: On the first subscribe, there probably is nothing to yet replay, and the timeout behaved as I'd expected. When another subscribe occurs, there is probably now an item to be replayed on subscription, and the timeout will trigger subsequently.
Expected behavior
I do not expect the timeout to trigger, as the subscription has already been pushed a value.
Expected and Current Output from Example Program
Running behavior_subject_example
Current Output
# yarn cli behavior
1
-1
Expected Output
# yarn cli behavior
1
2
Running share_replay_example
Current Output
# yarn cli share
First Sub: 1
Second Sub: 1
Second Sub: -1
Expected Output
# yarn cli share
First Sub: 1
Second Sub: 1
Reproduction code
// From main.ts in the linked repo:
import { timeout, of, BehaviorSubject, shareReplay, Observable} from 'rxjs'
const wait = async (ms: number) => {
await new Promise((resolve) => { setTimeout(resolve, ms) })
}
export const behavior_subject_example = async (): Promise<void> => {
const base_subject$ = new BehaviorSubject(1)
const with_timeout$ = base_subject$.pipe(
timeout({first: 500, with: () => of(-1)}),
);
const sub = with_timeout$.subscribe(
console.log
)
await wait(1000)
base_subject$.next(2)
await wait(1000)
sub.unsubscribe()
}
export const share_replay_example = async (): Promise<void> => {
const long_running_observable$ = new Observable((subscriber) => {
setTimeout(() => subscriber.next(1), 1000)
});
const with_replay$ = long_running_observable$.pipe(
shareReplay(1),
)
const with_timeout$ = with_replay$.pipe(
timeout({first: 1500, with: () => of(-1)}),
)
const first_sub$ = with_timeout$.subscribe((x) => console.log(`First Sub: ${x}`))
await wait(2500);
const second_sub$ = with_timeout$.subscribe((x) => console.log(`Second Sub: ${x}`))
await wait(2500);
first_sub$.unsubscribe()
second_sub$.unsubscribe()
}Reproduction URL
https://github.com/Eosis/rxjs-timeout-example
Version
7.5.4
Environment
Node: v15.14.0
Additional context
No response