Deliver the first item immediately, 'debounce' following items

Question

Consider the following use case:

• need to deliver first item as soon as possible
• need to debounce following events with 1 second timeout

I ended up implementing custom operator based on OperatorDebounceWithTime then using it like this

.lift(new CustomOperatorDebounceWithTime<>(1, TimeUnit.SECONDS, Schedulers.computation()))

CustomOperatorDebounceWithTime delivers the first item immediately then uses OperatorDebounceWithTime operator's logic to debounce later items.

Is there an easier way to achieve described behavior? Let's skip the compose operator, it doesn't solve the problem. I'm looking for a way to achieve this without implementing custom operators.

Answer 1

Update:
From @lopar's comments a better way would be:

Observable.from(items).publish(publishedItems -> publishedItems.limit(1).concatWith(publishedItems.skip(1).debounce(1, TimeUnit.SECONDS)))

Would something like this work:

String[] items = {"one", "two", "three", "four", "five"};
Observable<String> myObservable = Observable.from(items);
Observable.concat(
  myObservable.first(), 
  myObservable.skip(1).debounce(1, TimeUnit.SECONDS)
).subscribe(s -> 
  System.out.println(s));

Answer 2

The answers by @LortRaydenMK and @lopar are best, but I wanted to suggest something else in case it happened to work better for you or for someone in a similar situation.

There's a variant of debounce() that takes a function that decides how long to debounce this particular item for. It specifies this by returning an observable that completes after some amount of time. Your function could return empty() for the first item and timer() for the rest. Something like (untested):

String[] items = {"one", "two", "three", "four", "five", "six"};
Observable.from(items)
    .debounce(item -> item.equals("one")
            ? Observable.empty()
            : Observable.timer(1, TimeUnit.SECONDS));

The trick is that this function would have to know which item is the first. Your sequence might know that. If it doesn't, you might have to zip() with range() or something. Better in that case to use the solution in the other answer.

Answer 3

A simple solution using RxJava 2.0, translated from the answer for the same question for RxJS, which combines throttleFirst and debounce, then removes duplicates.

private <T> ObservableTransformer<T, T> debounceImmediate() {
    return observable  -> observable.publish(p -> 
        Observable.merge(p.throttleFirst(1, TimeUnit.SECONDS), 
            p.debounce(1, TimeUnit.SECONDS)).distinctUntilChanged());
} 

@Test
public void testDebounceImmediate() {
    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(LocalDateTime.now() + " T=" + v))
            .compose(debounceImmediate())
            .blockingSubscribe(v -> System.out.println(LocalDateTime.now() + " Debounced: " + v));
}

The approach of using limit() or take() doesn't seem to handle long lived data flows, where I might want to continually observe, but still act immediately for the first event seen for a time.

Answer 4

The LordRaydenMK and lopar's answer has a problem: you always lose the second item. I supose that no one realeased this before because if you have a debounce you normally has a lot of events and the second is discarted with the debounce anyways. The correct way to never lose an event is:

observable
    .publish(published ->
        published
            .limit(1)
            .concatWith(published.debounce(1, TimeUnit.SECONDS)));

And don't worry, you are not going to get any duplicated event. If you aren't sure about it you can run this code and check it yourself:

Observable.just(1, 2, 3, 4)
    .publish(published ->
        published
            .limit(1)
            .concatWith(published))
    .subscribe(System.out::println);

Answer 5

Use the version of debounce that takes a function and implement the function in this way:

    .debounce(new Func1<String, Observable<String>>() {
        private AtomicBoolean isFirstEmission = new AtomicBoolean(true);
        @Override
        public Observable<String> call(String s) {
             // note: standard debounce causes the first item to be
             // delayed by 1 second unnecessarily, this is a workaround
             if (isFirstEmission.getAndSet(false)) {
                 return Observable.just(s);
             } else {
                 return Observable.just(s).delay(1, TimeUnit.SECONDS);
             }
        }
    })

The first item emits immediately. Subsequent items are delayed by a second. If a delayed observable doesn't terminate before the following item arrives, it's cancelled, so the expected debounce behavior is fulfilled.

Answer 6

Kotlin extension functions based on @lopar's comment:

fun <T> Flowable<T>.debounceImmediate(timeout: Long, unit: TimeUnit): Flowable<T> {
    return publish {
        it.take(1).concatWith(it.debounce(timeout, unit))
    }
}

fun <T> Observable<T>.debounceImmediate(timeout: Long, unit: TimeUnit): Observable<T> {
    return publish {
        it.take(1).concatWith(it.debounce(timeout, unit))
    }
}

Answer 7

Ngrx - rxjs solution, split the pipe to two

onMyAction$ = this.actions$
    .pipe(ofType<any>(ActionTypes.MY_ACTION);

lastTime = new Date();

@Effect()
onMyActionWithAbort$ = this.onMyAction$
    .pipe(
        filter((data) => { 
          const result = new Date() - this.lastTime > 200; 
          this.lastTime = new Date(); 
          return result; 
        }),
        switchMap(this.DoTheJob.bind(this))
    );

@Effect()
onMyActionWithDebounce$ = this.onMyAction$
    .pipe(
        debounceTime(200),
        filter(this.preventDuplicateFilter.bind(this)),
        switchMap(this.DoTheJob.bind(this))
    );

Answer 8

My solution for Dart:

extension StreamExt<T> on Stream<T> {
  Stream<T> immediateDebounce(Duration duration) {
    var lastEmit = 0;
    return debounce((event) {
      if (_now - lastEmit < duration.inMilliseconds) {
        lastEmit = _now;
        return Stream.value(event).delay(duration);
      } else {
        lastEmit = _now;
        return Stream.value(event);
      }
    });
  }
}

int get _now =>  DateTime.now().millisecondsSinceEpoch;

Answer 9

To prevent making dual subscriptions using this:

    const debouncedSkipFirstStream$ = stream$.pipe(
        map((it, index) => ({ it, index })),
        debounce(({ index }) => (
            index ? new Promise(res => setTimeout(res, TimeUnit.SECONDS))
                : Rx.of(true))),
        map(({ it }) => it),
    );

if using split solution you will see 'run' print twice

x = rxjs.Observable.create(o=>{
    console.info('run');
    o.next(1);
    o.next(2);
});
a = x.pipe(rxjs.operators.take(1));
b = x.pipe(rxjs.operators.skip(1), rxjs.operators.debounceTime(60));
rxjs.concat(a, b).subscribe(console.log);

Answer 10

I went with

Flowable.concat(

    flowable // emits immediately
        .take(1)
        .skipWhile { it.isEmpty() },

    flowable // same flowable, but emits with delay and debounce
        .debounce(2, TimeUnit.SECONDS)
)
    .distinctUntilChanged()

Answer 11

If someone looking for this in 2021:

@OptIn(FlowPreview::class)
fun <T> Flow<T>.debounceImmediate(timeMillis: Long): Flow<T> =
    withIndex()
        .onEach { if (it.index != 0) delay(timeMillis) }
        .map { it.value }

Usage:

authRepository.login(loginDto)
                    .debounceImmediate(10000)

Answer 12

After reading this article I ended up using the throttleLatest operator to get very similar behaviour to the immediate debounce I was looking for.

The following code will instantly emit the first item and then check for new items every 500ms. Only the latest event received within that 500ms window will be sent out.

observable.throttleLatest(500, TimeUnit.MILLISECONDS)

Answer 13

   view.clicks()
            .throttleFirst(2, TimeUnit.SECONDS)
            .subscribe {
                println("Clicked button")
            }

I found this the simplest way to do that. clicks() come from rx view binding. Add this dependency to get view observable

 implementation 'com.jakewharton.rxbinding4:rxbinding:4.0.0

Answer 14

For those who are trying to solve the same problem using Kotlin Flow:

fun <T> Flow<T>.throttleFirst(timeout: Duration): Flow<T> {
    var job = Job().apply { complete() }
    return onCompletion { job.cancel() }.run {
        flow {
            coroutineScope {
                collect { value ->
                    if (!job.isActive) {
                        emit(value)
                        job = launch { delay(timeout.inWholeMilliseconds) }
                    }
                }
            }
        }
    }
}

Example:

flow {
    emit(1)
    delay(90)
    emit(2)
    delay(90)
    emit(3)
    delay(1010)
    emit(4)
    delay(1010)
    emit(5)
}.throttleFirst(1.seconds).collect { ... }
// 1, 4, 5

• Ref

Answer 15

const changes = this.form.valueChanges; // For example

const changesDebounced = changes.pipe(
    exhaustMap(change => merge(
        of(change),
        changes.pipe(startWith(change), debounceTime(500), take(1)),
    )),
    distinctUntilChanged(),
);

changesDebounced will emit immediately on changes. If subsequent changes occur, they are debounced (by 500ms). After no further values have been emitted for 500ms, changesDebounced will again emit immediately on the next change.

Using exhaustMap on the changes observable will ignore incoming changes while we execute the debounce logic.

Then we merge two observables: of(change) which will emit the change immediately, and also the debounced changes. Use take(1) so that we return control to the outer observable after the first debounced value is emitted. This lets us listen for changes again without debouncing.

Also notice that the debounced observable uses startsWith(change). This is in case only one change is emitted within the debounce interval, meaning there are no other changes to debounce.

Lastly, we need distinctUntilChanged(). This is again in case only one change is emitted during the debounce interval. In that scenario, the debounced observable will emit the same value as the of(change) observable, so we can ignore the redundant emitted value.