diff --git a/src/shared/models/rx/cancellable-observable.class.ts b/src/shared/models/rx/cancellable-observable.class.ts new file mode 100644 index 00000000..590b4546 --- /dev/null +++ b/src/shared/models/rx/cancellable-observable.class.ts @@ -0,0 +1,42 @@ +import { BehaviorSubject, Observable, Observer, filter, lastValueFrom, take } from "rxjs"; + +export class CancellableObservable extends Observable { + + private readonly _isCancelled$ = new BehaviorSubject(false); + + constructor(subscriber: (obs: CancellableObserver) => void) { + + super((observer: Observer) => { + + const cancellableObserver: CancellableObserver = Object.assign(observer, { + $isCancelled: () => this.$isCancelled(), + onCancel: (callback: () => void) => this.onCancel(callback) + }); + + subscriber(cancellableObserver); + }); + + lastValueFrom(this).finally(() => this._isCancelled$.complete()); + } + + public cancel(): void { + this._isCancelled$.next(true); + } + + public onCancel(callback: () => void): void { + this.$isCancelled().pipe(filter(isCancelled => isCancelled), take(1)).subscribe(callback); + } + + public $isCancelled(): Observable { + return this._isCancelled$.asObservable(); + } + + public isCancelled(): boolean { + return this._isCancelled$.getValue(); + } +} + +export interface CancellableObserver extends Observer { + readonly $isCancelled: () => Observable; + readonly onCancel: (callback: () => void) => void; +} \ No newline at end of file