How is it possible to stop a debounced Rxjs Observable?

Question

I created an observable, which will fire 3 seconds after the last change is made, and calls the publishChange of the service. It works, but I would like to create a doImmediateChange function, which calls publishChange immediately and stops the debounced observable. How is that possible?

My component:

class MyComponent {
    private updateSubject = new Subject<string>();

    ngOnInit() {
        this.updateSubject.pipe(
            debounceTime(3000),
            distinctUntilChanged()
        ).subscribe(val => {
            this.srv.publishChange(val);
        });
    }

    doChange(val: string) {
        this.updateSubject.next(val);
    }

    doImmediateChange(val: string) {

        // Stop the current updateSubject if debounce is in progress and call publish immediately
        // ??
        this.srv.publishChange(val);

    }

}

Answer 1

Use the race operator:

The first observable to complete becomes the only observable subscribed to, so this recursive function will complete after one emission take(1), then resubscribe () => this.raceRecursive().

private timed$ = new Subject<string>();
private event$ = new Subject<string>();

ngOnInit() {
  this.raceRecursive()
}

raceRecursive() {
  race(
    this.timed$.pipe(debounceTime(1000)),
    this.event$
  )
    .pipe(take(1)) // force it to complete
    .subscribe(
      val => console.log(val), // srv call here
      err => console.error(err),
      () => this.raceRecursive() // reset it once complete
    )
}

doChange(val: string) {
  this.timed$.next(val)
}

doImmediateChange(val: string) {
  this.event$.next(val)
}

Answer 2

You can emulate debounceTime using switchMap and delay. Then cancel the inner Observable with takeUntil to prevent a waiting value from being emitted.

private updateSubject = new Subject<string>();
private interrupt = new Subject();

ngOnInit() {
  this.updateSubject.pipe(
    switchMap(val => of(val).pipe(
      delay(3000),
      takeUntil(this.interrupt)
    ))
  ).subscribe(val => publish(val));
}

doChange(val: string) {
  this.updateSubject.next(val);
}

doImmediateChange(val: string) {
  this.interrupt.next();
  publish(val);
}

https://stackblitz.com/edit/rxjs-ya93fb

Answer 3

You can achieve this behavior using debounce and race:
with the code you provided

private destroy$ = new Subject<void>();
private immediate$ = new Subject<void>();
private updateSubject$ = new Subject<string>();

constructor(private srv: PubSubService) {}

ngOnInit() {
  this.updateSubject$.pipe(
      takeUntil(this.destroy$),
      debounce(() => race(timer(3000), this.immediate$))
  ).subscribe(val => {
      this.srv.publishChange(val);
  });
}

doChange(val: string, immediate?: boolean) {
  this.updateSubject$.next(val);
  if (immediate) this.immediate$.next();
}

// don't forget to unsubscribe
ngOnDestroy() {
  this.destroy$.next();
}

emitting an immediate change will replace the previous normal change (that is debounced for 3s) without the delay (thanks to our race observable).

here's a working example