RxJS поведение дросселя; получить первое значение немедленно

Question

Пример Plunkr: https://plnkr.co/edit/NZwb3ol8CbZFtSc6Q9zm?p=preview

Я знаю, что для rxjs есть 3 метода дросселя (5.0 beta.4):

auditTime(), throttleTime() а также debounceTime()

Я ищу поведение, которое lodash делает по умолчанию на газе:

• 1) Дайте мне первое значение немедленно!
• 2) для последовательных значений удерживайте значения для заданной задержки, а затем отправьте последнее полученное значение
• 3) по истечении времени задержки газа вернитесь в состояние (1)

Теоретически это должно выглядеть так:

inputObservable
  .do(() => cancelPreviousRequest())
  .throttleTime(500)
  .subscribe((value) => doNextRequest(value))

Но

• throttleTime никогда не дает мне последнее значение, если выбрасывается в таймауте газа
• debounceTime не срабатывает сразу
• auditTime не срабатывает сразу

Могу ли я объединить любой из методов rxjs для достижения описанного поведения?

Answer 1

Я взял оператор AuditTime и изменил 2 строки для достижения желаемого поведения.

Новый плункер: https://plnkr.co/edit/4NkXsOeJOSrLUP9WEtp0?p=preview

Оригинал:

• https://github.com/ReactiveX/rxjs/blob/master/src/operator/auditTime.ts

Изменения:

from (auditTime):

protected _next(value: T): void {
  this.value = value;
  this.hasValue = true;
  if (!this.throttled) {
    this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this));
  }
}

clearThrottle(): void {
  const { value, hasValue, throttled } = this;
  if (throttled) {
    this.remove(throttled);
    this.throttled = null;
    throttled.unsubscribe();
  }
  if (hasValue) {
    this.value = null;
    this.hasValue = false;
    this.destination.next(value);
  }
}

to (auditTimeImmediate):

protected _next(value: T): void {
    this.value = value;
    this.hasValue = true;
    if (!this.throttled) {
        // change 1:
        this.clearThrottle();
    }
}

clearThrottle(): void {
    const { value, hasValue, throttled } = this;
    if (throttled) {
        this.remove(throttled);
        this.throttled = null;
        throttled.unsubscribe();
    }
    if (hasValue) {
        this.value = null;
        this.hasValue = false;
        this.destination.next(value);
        // change 2:
        this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this));
    }
}

Поэтому я начинаю тайм-аут после того, как значение было nextредактор

Использование:

inputObservable
  .do(() => cancelPreviousRequest())
  .auditTimeImmediate(500)
  .subscribe((value) => doNextRequest(value))

Answer 2

Для тех, кто ищет это после 2018 года: это было добавлено более года назад, но по какой-то причине документация еще не была обновлена.

RxJS коммит

Вы можете просто передать объект конфигурации throttleTime, По умолчанию { leading: true, trailing: false }, Чтобы добиться поведения, обсуждаемого здесь, вы просто должны установить trailing в true: { leading: true, trailing: true }

Answer 3

Для старых RxJ я написал concatLatest оператор, который делает большую часть того, что вы хотите. С его помощью вы можете получить свое поведение регулирования с помощью этого кода:

const delay = Rx.Observable.empty().delay(500);
inputObservable
    .map(value => Rx.Observable.of(value).concat(delay))
    .concatLatest()
    .subscribe(...);

Вот оператор. Я попытался обновить его для работы с RxJS5:

Rx.Observable.prototype.concatLatest = function () {
    /// <summary>
    /// Concatenates an observable sequence of observable sequences, skipping sequences that arrive while the current sequence is being observed.
    /// If N new observables arrive while the current observable is being observed, the first N-1 new observables will be thrown
    /// away and only the Nth will be observed.
    /// </summary>
    /// <returns type="Rx.Observable"></returns>
    var source = this;

    return Rx.Observable.create(function (observer) {
        var latest,
            isStopped,
            isBusy,
            outerSubscription,
            innerSubscription,
            subscriptions = new Rx.Subscription(function () {
              if (outerSubscription) {
                outerSubscription.unsubscribe();
              }
              if (innerSubscription) {
                innerSubscription.unsubscribe();
              }
            }),
            onError = observer.error.bind(observer),
            onNext = observer.next.bind(observer),
            innerOnComplete = function () {
                var inner = latest;
                if (inner) {
                    latest = undefined;
                    if (innerSubscription) {
                      innerSubscription.unsubscribe();
                    }
                    innerSubscription = inner.subscribe(onNext, onError, innerOnComplete);
                }
                else {
                    isBusy = false;
                    if (isStopped) {
                        observer.complete();
                    }
                }
            };

        outerSubscription = source.subscribe(function (newInner) {
            if (isBusy) {
                latest = newInner;
            }
            else {
                isBusy = true;
                if (innerSubscription) {
                  innerSubscription.unsubscribe();
                }
                innerSubscription = newInner.subscribe(onNext, onError, innerOnComplete);
            }
        }, onError, function () {
            isStopped = true;
            if (!isBusy) {
                observer.complete();
            }
        });

        return subscriptions;
    });
};

А вот обновленный план: https://plnkr.co/edit/DSVmSPRijJwj9msefjRi?p=preview

Примечание. Я обновил вашу версию lodash до последней версии. В lodash 4.7 я переписал операторы throttle/debounce, чтобы исправить некоторые ошибки в крайнем случае. Вы использовали 4.6.1, у которой все еще были некоторые из этих ошибок, хотя я не думаю, что они влияли на ваш тест.