rxjs: cancelling a debounced observable

Question

I have an observable Subject that emits some changes with debouncing:

someSubject.pipe(
  debounceTime(5000),
).subscribe(response => {
  console.log('Value is', response);
})

Now, I need a Stop button somewhere on the screen that would cancel my debounced emit. So I create a button:

const stopObs = new Subject();
...
<button onClick={() => stopObs.next()}>Stop</button>

and modify my subscription like so:

someSubject.pipe(
  debounceTime(5000),
  takeUntil(stopObs),
).subscribe(response => {
  console.log('Value is', response);
})

This works fine, after hitting "Stop" I stop getting values in console, but there is a problem: the observable is stopped forever. And I need it to be able to emit new values, I only need to cancel already started debounced emits.

My first thought was to create a new subject and use repeatWhen:

const startObs = new Subject();
...
<button onClick={() => startObs.next()}>Start</button>

...
someSubject.pipe(
  debounceTime(5000),
  takeUntil(stopObs),
  repeatWhen(() => startObs)
).subscribe(response => {
  console.log('Value is', response);
})

But there's another problem: if I hit "Start" button more than one time and emit more than one value to startObs, then I start getting multiple console.log's for single debounced value!

So is there a way to cancel only debounced emits without stopping the entire observable?

It looks like you could just use repeat() instead of repeatWhen() since you're using just Subject so it sohouldn't matter that you'll stay subscribed to the source Observable. — martin

Andrei Gătej Andrei Gătej · Accepted Answer · 2020-11-07T21:26:31

Since debounceTime is just

const duration = timer(dueTime, scheduler);
return debounce(() => duration);

I think you can solve the problem like this:

someSubject.pipe(
  debounce(() => timer(5000).pipe(takeUntil(stopObs))),
)

If you want to send the last value when the timer is cancelled due to stopObs, you could try this:

someSubject.pipe(
  debounce(
    () => timer(5000)
      .pipe(
        takeUntil(stopObs),
        isEmpty(),
      )
  ),
)

isEmpty() will emit true immediately before a complete notification, which is what debounce needs in order to send the last received value. If the timer completes without stopObs's involvement, isEmpty will emit false instead of true, but this still works well for debounce, since it only needs a value from the inner observable.

rxjs: cancelling a debounced observable

1 Answers