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