How is it possible to stop a debounced Rxjs Observable?

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


You could supply a value specific debounce time with every value and use debounce with timer to change the debounce time for values dynamically.

private updateSubject = new Subject<{ value: any, debounceTime: number}>();

ngOnInit() {
  updateSubject.pipe(
    debounce(({ debounceTime }) => timer(debounceTime)),
    pluck('value')
  ).subscribe(val => publish(val));
}

doChange(value: string) {
  updateSubject.next({ value, debounceTime: 3000 });
}

doImmediateChange(value: string) {
  updateSubject.next({ value, debounceTime: 0 });
}

This doesn't directly stop the debounced Observable but let's you "overwrite" a waiting value with a new one being emitted with zero delay.

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

(user733421 didn't seem to want to add a complete solution so I expanded the approach)


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)
}

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