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.

[benlesh]

Hi @CorentinHardy — thank you for taking the time to open this issue and for sharing your thoughtful feedback with the RxJS project. I’m sorry it took us so long to respond. Given how much time has passed and the lack of recent activity, I’m going to close this for now so the tracker reflects the work we’re likely to act on. If this is still relevant on a current RxJS release, please feel welcome to open a new issue with updated context and a reproduction. Thank you again for your contribution and patience.

— Ben