rxjs-pausable-operator

rxjs-pausable-operator.ts

class PausableObservable<T> extends Observable<T> {
  private pauser: BehaviorSubject<boolean>;

  pause(){
    this.pauser.next(true);
  }

  resume(){
    this.pauser.next(false);
  }
}

function pausable() {
  return function pauseFn<T>(source: Observable<T>): PausableObservable<T> {
    const pausableProto = PausableObservable.prototype;

    const pauser = new BehaviorSubject(false);
    const newSource = pauser.pipe(
      switchMap(paused => paused ? NEVER : source.pipe(materialize())),
      dematerialize()
    )

    const pausable = Object.create(newSource, {
      pause: { value: pausableProto.pause },
      resume: { value: pausableProto.resume },
      pauser: { value: pauser }
    });
    return pausable as PausableObservable<T>;
  };
}