rxjs наблюдаемый debounceTime inner next игнорируется

Question

Кажется, что debounceTime игнорирует внутренние вызовы своих субъектов next метод:

var subject: Subject<number> = new Subject<number>();

subject.pipe(
    tap((a) => console.log("tab:" + a)), 
    debounceTime(300), 
).subscribe((a) => {
    console.log(a);
    subject.next(100)
});

subject.next(19);
subject.next(20);

приведенный выше код должен создавать бесконечный цикл - но это не так:

tab:19
tab:20
20
tab:100

если я добавлю delay(1) к трубе, она будет работать как положено:

subject.pipe(
    tap((a) => console.log("tab:" + a)), 
    debounceTime(300), 
    delay(1)
).subscribe((a) => {
    console.log(a);
    subject.next(100)
});

я что-то упустил?

Редактировать : добавлен пример: https://typescript -fbt2mn.stackblitz.io

Answer 1

Этот тип проблем, когда кажется, что что-то сломано и может быть волшебным образом исправлено с помощью delay(0) или setTimeout, обычно означает, что вы ожидаете, что RxJS будет вести себя асинхронно все время, хотя на самом деле он обычно работает синхронно, если только вынужден поступить иначе.

И это точно ваш случай.

• Когда debounce увеличивает значение, вы попадаете в эту строку https://github.com/ReactiveX/rxjs/blob/5.5.10/src/operators/debounceTime.ts#L100.

• Внутри subscribe вы звоните next, который достигает _next внутри debounceTime https://github.com/ReactiveX/rxjs/blob/5.5.10/src/operators/debounceTime.ts#L84, который устанавливает lastValue и hasValue внутренние переменные.

• Теперь он начинает идти вверх в стеке вызовов и только теперь прекращает обработку этой строки https://github.com/ReactiveX/rxjs/blob/5.5.10/src/operators/debounceTime.ts#L100. В этот момент он продолжает и переопределяет lastValue и hasValue снова https://github.com/ReactiveX/rxjs/blob/5.5.10/src/operators/debounceTime.ts#L101-L102. Это устанавливает их на null и false.

• Затем после 300ms debounceTime хочет выдать его значение, но hasValue равно null https://github.com/ReactiveX/rxjs/blob/5.5.10/src/operators/debounceTime.ts#L99, поскольку оно уже было переопределено.

Кстати, немного более элегантным способом является использование оператора observeOn с планировщиком async, который должен быть более эффективным, чем delay(1).

import { async } from 'rxjs/scheduler/async';

var obs = Observable.of(1, 2, 3);
var subject: Subject<number> = new Subject<number>();

subject.pipe(
    tap((a) => console.log("tab:" + a)), 
    debounceTime(300),
    observeOn(async),
).subscribe((a) => {
    console.log(a);
    subject.next(a+1)
});

Увидимся обновленные демо: https://stackblitz.com/edit/typescript-tsek9s?file=index.ts

Редактировать: Это поведение, вероятно, изменилось в RxJS 6: https://github.com/ReactiveX/rxjs/pull/3218