throttleTime does not ignore subsequent sources if next is in subscribe

[CorentinHardy]

Describe the bug

I recently update from version 6.6 to version 7.5. I found some infinite loop in my code due to throttleTime that didn't work as it used to. We are calling in our subscribe some method that may emit new value on the subscribed observable. throtteTime filtered those values, but it let them pass in version 7.5.6 . Now it behave as it was only synchronous.

Expected behavior

throttleTime should still ignore subsequent sources even if they were emitted in subscribe.

Reproduction code

testThrotteTimeWithNextInSubscribe() {
    const d_1 = new Date().getTime();
    const subject = new Subject<{iteration: number, name: string}>();
    subject.pipe(
      tap((tic) => console.debug('at', new Date().getTime() - d_1, 'tap', tic)),
      throttleTime(10000)
    ).subscribe((me) => {
      console.log('at', new Date().getTime() - d_1, 'subscribe', me);
      if (me.iteration < 10) {
        subject.next({iteration: me.iteration + 1, name: me.name + '-a'});
        subject.next({iteration: me.iteration + 1, name: me.name + '-b'});
      }
    });
    subject.next({iteration: 1, name: 'bug'});
  }

Reproduction URL

No response

Version

7.5.6

Environment

node 14.19.3
typescript 4.6.4

Additional context

No response

[gcamara]

In version 7.5.6 the throttleTime uses throttle operator with a timer.

Checking throttle code, I noticed that in the example above, the subscriber will keep sending values to the source, making the calls to stack and process before subscribing to the timer.

if (hasValue) {
  // Ensure we clear out our value and hasValue flag
  // before we emit, otherwise reentrant code can cause
  // issues here.
  hasValue = false;
  const value = sendValue!;
  sendValue = null;
  // Emit the value.
  subscriber.next(value); // <-- here it'll call send values to the source, which will make it triggers the send again;
  !isComplete && startThrottle(value); // <-- startThrottle won't be called for the next call;
}

Checking if the throttled is set before the next call would make it work properly.

I added this code: hasValue && !throttled && startThrottle(sendValue as any); locally and ran some tests and it worked as before.

But running npm test I get some !!!, which, forgive my ignorance, but I don't know what they mean.

Hope it helps somehow =)

[MatthiasvB]

I just encountered the same issue. I have reproduction code that is somewhat simpler:

const cycle$$ = new Subject<number>();
cycle$$.pipe(throttleTime(1000)).subscribe(val => {
  console.log(val);
  cycle$$.next(val + 1);
});
cycle$$.next(1);

As a workaround, I now use observeOn(asapScheduler) before throttleTime, which fixes the problem.