Throttle operator doesn't work correctly with sync duration selector

[fr-ench]

Bug Report

Current Behavior
If a sync duration selector is used in throttle operator, throttle operator won't recognise it has emitted an event and will keep on throttling all other events.

According to throttle documentation duration selector is "A function that receives a value from the source Observable, for computing the silencing duration for each source value, returned as an Observable or a Promise.", hence I would expect it to work with sync streams. Besides, the same behaviour does work as expected with the debounce operator.

Reproduction

https://stackblitz.com/edit/rxjs-w7dbvs

Please see console, to see all both async duration selectors and debounce operator keep on emitting events, while throttle operator with sync duration selector keeps on ignoring them as if no event was emitted by duration selector.

Expected behavior
The throttle operator should work with a sync Observable the same way it does work with async one (and the same way debounce operator does). If sync duration selector is used, throttle should keep on immediately emitting the following events and not ignoring them.

Environment

• Runtime: Chrome
• RxJS version: both ^6.5.2 and 7.0.0-beta.0.

[Andrei0872]

I think the problem is here:

rxjs/src/internal/operators/throttle.ts

Lines 125 to 130 in f5c6332

	private throttle(value: T): void {
	const duration = this.tryDurationSelector(value);
	if (!!duration) {
	this.add(this._throttled = subscribeToResult(this, duration));
	}
	}

Considering your first example:

source.pipe(
  throttle(x => of(true))
).subscribe(x => console.log('a ', x), null, () => console.log('completed'));

duration will be of(true), an observable that synchronously emits and then completes.

subscribeToResult(this, duration) will subscribe to this observable and because everything happens synchronously, it also completes. Therefore, a closed inner subscriber will be assigned to this._throttled.

And here is what happens when a new outer value arrives:

rxjs/src/internal/operators/throttle.ts

Lines 102 to 113 in f5c6332

	protected _next(value: T): void {
	this._hasValue = true;
	this._sendValue = value;
	if (!this._throttled) {
	if (this._leading) {
	this.send();
	} else {
	this.throttle(value);
	}
	}
	}

because this._throttled is not null, nothing will happen, meaning that it can't be emptied out.

A quick fix I think would be this:

private throttle(value: T): void {
  const duration = this.tryDurationSelector(value);
  if (!!duration) {
    const innerSubscriber = subscribeToResult(this, duration);

    !innerSubscriber.closed && this.add(this._throttled = innerSubscriber);
  }
}

[benlesh]

Hi @fr-ench — 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