rx-rpc.ts

import { BehaviorSubject, Observable, Subscription } from 'rxjs';

export class RxRpc<P, R> {
  constructor(private readonly cmd: (params: P) => Observable<R>) {}

  private subscription: Subscription | null = null;

  private readonly _pending$ = new BehaviorSubject(false);
  public readonly pending$ = this._pending$.asObservable();

  private readonly _errorMsg$ = new BehaviorSubject<string | null>(null);
  public readonly errorMsg$ = this._errorMsg$.asObservable();

  private readonly _result$ = new BehaviorSubject<R | void>(undefined);
  public readonly result$ = this._result$.asObservable();

  public execute(params?: P): void;
  public execute(params: P): void {
    this.cancel();

    this._pending$.next(true);

    this.subscription = this.cmd(params).subscribe(
      (result: R) => {
        this._pending$.next(false);
        this._result$.next(result);
      },
      (reason: string) => {
        this._pending$.next(false);
        this._errorMsg$.next(reason);
      },
    );
  }

  public cancel(): void {
    if (this.subscription === null) { return; }

    this.subscription.unsubscribe();
    this.subscription = null;

    this._pending$.next(false);
    this._errorMsg$.next(null);
    this._result$.next(undefined);
  }
}