DelayQueue failed: behavior breaking change after RxJS from v6 to v7

[huan]

The following code work as expected with RxJS@6:

rx-queue/src/delay-queue.ts

Lines 35 to 38 in 79f8ebb

	concatMap(args => concat(
	of(args), // emit first item right away
	EMPTY.pipe(delay(this.period)), // delay next item
	)),

It will add a delay between each item.

However, the above code does not work anymore after we upgrade to RxJS@7, it likes that all delay does not take effect anymore: all items will be emitted without any delay.

Problem: breaking changes from RxJS

After performed my google-fu, I found there's a "behavior of delay inconsistent with completions" (ReactiveX/rxjs#4249) created by @benlesh which talked about this behavior and there's also a PR named "fix(delay): emit complete notification as soon as possible" (ReactiveX/rxjs#4444) has implemented this change.

So that's the reason why our code break after upgrade to RxJS v7.

Solution

Use the following code to emit complete after the delay this.period.

concatMap(x => concat(
  of(x),
  timer(this.period).pipe(skip(1)),
)),