Немедленное отторжение в Rx

Я ищу оператора для debounce серии событий, скажем, пользовательский клик. Вход и выход должны быть такими:

interval :      ->     <-            ->     <-
in       : 1--2--3-------4--5--5--6-7-8--------
out      : 1-------------4---------------------

Идея похожа на подчеркивание с опцией immediate on http://underscorejs.org/#debounce. Оператор может быть представлен/реализован на любом языке, который поддерживает Reactive Extensions

Изменение: уточнить интервал, скажем, 5 секунд (5 пробелов между двумя стрелками): -> <-

Edit2: более понятная версия: у меня есть пользователь, он несколько раз нажимает кнопку (1, 2, 3); Я хочу поймать первый click (1) и игнорировать остальные. Через некоторое время он устает и отдыхает в течение 7 секунд (что больше, чем 5-секундный интервал между двумя стрелками) и снова нажимаю кнопку (4, 5, 6, 7, 8). Я хочу поймать первую click (4) и игнорируйте все остальное.

Если он нажимает после четвертой стрелки, я тоже хочу поймать этот щелчок.

Edit3: вот изображение , которое можно найти в оригинальной статье

Ответ 1

Изменить: Основываясь на пояснениях, RxJava не имеет оператора для этого типа потока, но может быть составлен из нетривиального набора других операторов:

import java.util.concurrent.TimeUnit;

import rx.Observable;

public class DebounceFirst {

    public static void main(String[] args) {
        Observable.just(0, 100, 200, 1500, 1600, 1800, 2000, 10000)
        .flatMap(v -> Observable.timer(v, TimeUnit.MILLISECONDS).map(w -> v))
        .doOnNext(v -> System.out.println("T=" + v))
        .compose(debounceFirst(500, TimeUnit.MILLISECONDS))
        .toBlocking()
        .subscribe(v -> System.out.println("Debounced: " + v));
    }

    static <T> Observable.Transformer<T, T> debounceFirst(long timeout, TimeUnit unit) {
        return f -> 
            f.publish(g ->
                g.take(1)
                .concatWith(
                    g.switchMap(u -> Observable.timer(timeout, unit).map(w -> u))
                    .take(1)
                    .ignoreElements()
                )
                .repeatWhen(h -> h.takeUntil(g.ignoreElements()))
            );
    }
}

Ответ 2

Нужное поведение - это не то, что делает оператор debounce в Rx.

Это называется throttle, throttleTime или throttleWithTimeout (однако, оно подпадает под категорию операторов debounce). Я не знаю, какой язык вы используете, но в RxJS это выглядит следующим образом:

Смотрите http://reactivex.io/documentation/operators/debounce.html.

Ответ 3

Поскольку debounce() по сути является асинхронным, вам нужно явно вернуть результат в текущий поток.

seriesOfUnfortunateEvents
  .debounce( 14, TimeUnit.MILLISECONDS )
  .observeOn( Schedulers.immediate() )
  .subscribe( v -> yourStuff() );

Ответ 4

Согласно документации, в RxJS есть два оператора debounce. Вас может заинтересовать, в частности, debounceTime.

`debounceTime`

Из документации

Испускает значение из источника Наблюдаемое только после того, как определенный промежуток времени прошел без другого испускания источника.

Пример:

Rx.Observable
    .fromEvent(document.querySelector('button'), 'click')
    .debounceTime(200)
    .mapTo(() => 'clicked!')
    .subscribe(v => console.log(v));

Он выйдет одним нажатием! если кнопка была нажата в заданное время (200 мс в этом примере).

`debounce`

Из документации

Выдает значение из источника Наблюдаемое только через определенное время диапазон, определенный другим Наблюдателем, прошел без другого излучение источника.