pause/resume a timer Observable

Question

I'm building a simple stopwatch with angular/rxjs6, I can start the timer but I can't pause/resume it.

  source: Observable<number>;
  subscribe: Subscription;

  start() {
    this.source = timer(0, 1000);
    this.subscribe = this.source
      .subscribe(number => {
        this.toSeconds = number % 60;
        this.toMinutes = Math.floor(number / 60);
        this.toHours = Math.floor(number / (60 * 60));

        this.seconds = (this.toSeconds < 10 ? '0' : '') + this.toSeconds;
        this.minutes = (this.toMinutes < 10 ? '0' : '') + this.toMinutes;
        this.hours = (this.toHours < 10 ? '0' : '') + this.toHours;
    });
  }

  pause() {
    this.subscribe.unsubscribe(); // not working
  }

after doing lot of searching, I found that I should use switchMap operator to accomplish that, but I'm new to rxjs and don't know how to do it the right way.

Any help would be much appreciated.

Answer 1

For the benefit of anyone searching, this was my implementation of a pausable rxjs timer. It takes a pause observable so that you can pause/un-pause at will.

Usage:

pausableTimer(1000, this.myPauseObservable)
  .subscribe(() => { /* do your thing */ };

Code:

import { BehaviorSubject, combineLatest, Observable, timer } from 'rxjs';
import { delayWhen, startWith } from 'rxjs/operators';

export function pausableTimer(
  delayMs: number = 0,
  pause: Observable<boolean>
): Observable<number> {
  return new Observable<number>((observer: any) => {
    // The sequence of delays we'd like to listen to.
    // Start off with zero & paused = false so that combineLatest always fires at least once.
    const delays = new BehaviorSubject(0);
    pause = pause.pipe(startWith(false));

    // Piped delays that emit at the time given.
    const temporalDelays = delays.pipe(delayWhen((val) => timer(val)));

    let startTime = Date.now();
    let totalPauseTimeMs = 0;
    let lastPaused: number | undefined;

    // This listens to pause/unpause events, as well as timers.
    const sub = combineLatest([pause, temporalDelays]).subscribe(
      ([paused, delay]) => {
        if (paused) {
          // If we're paused, we never want to complete, even if the timer expires.
          if (lastPaused === undefined) {
            lastPaused = Date.now();
          }

          return;
        }

        // If we've just un-paused, add on the paused time.
        if (lastPaused !== undefined) {
          totalPauseTimeMs += Date.now() - lastPaused;
          lastPaused = undefined;
        }

        // Look at how much time has expired.
        const totalElapsed = Date.now() - startTime;
        const remainingTime = delayMs + totalPauseTimeMs - totalElapsed;

        if (remainingTime <= 0) {
          // We're done!
          observer.next(totalElapsed);
          observer.complete();
        } else {
          // We're not done.  If there's not already a timer running, start a new one.
          const lastTimerFinished = delay === delays.value;
          if (lastTimerFinished) {
            delays.next(remainingTime);
          }
        }
      }
    );

    observer.add(() => sub.unsubscribe());
  });
}

Answer 2

I've faced the same problem today (when implementing Tetris clone with Angular). Here is what I ended up with:

import { Subject, timer } from 'rxjs';

export class Timer {
  private timeElapsed = 0;
  private timer = null;
  private subscription = null;

  private readonly step: number;

  update = new Subject<number>();

  constructor(step: number) {
    this.timeElapsed = 0;
    this.step = step;
  }

  start() {
    this.timer = timer(this.step, this.step);
    this.subscription = this.timer.subscribe(() => {
      this.timeElapsed = this.timeElapsed + this.step;
      this.update.next(this.timeElapsed);
    });
  }

  pause() {
    if (this.timer) {
      this.subscription.unsubscribe();
      this.timer = null;
    } else {
      this.start();
    }
  }

  stop() {
    if (this.timer) {
      this.subscription.unsubscribe();
      this.timer = null;
    }
  }
}

And in my game service I use it like this:

  init() {
    this.timer = new Timer(50);
    this.timer.start();
    this.timer.update.subscribe(timeElapsed => {
      if (timeElapsed % 1000 === 0) {
        this.step(); // step() runs one game iteration
      }
    });
  }

  togglePause() {
    this.timer.pause();
  }

N.B.: I'm new to Angular/RxJS, so I'm not sure if the code above is good. But it works.

Answer 3

This is a node.js snippet using rxjs 6. Timer events will be emited unless p is pressed. When pressed again the emissions continue (ctrl-c will exit).

Internally, actually a new timer is started, when the pauser emits false. Hence we're prepending (concat) the pauser with a false emission to start the first timer. The 2 scan operators manage the state (pause toggler + counter) of the chain.

import { timer, concat, NEVER, of, fromEvent } from 'rxjs';
import { scan, tap, filter, switchMap } from 'rxjs/operators';
import { emitKeypressEvents } from 'readline';

process.stdin.setRawMode(true);
emitKeypressEvents(process.stdin);

const keypresses$ = fromEvent(process.stdin, 'keypress', (_, key) => key);
const pauser$ = keypresses$.pipe(
  tap(key => {
    if (key && key.ctrl && key.name == 'c') {
      process.exit(0);
    }
  }),
  filter(key => key.name === 'p'),
  scan(acc => !acc, false),
);

const starter$ = of(false);
concat(starter$, pauser$)
  .pipe(
    switchMap(stopped => (stopped ? NEVER : timer(0, 1000))),
    scan(acc => acc + 1, 0),
  )
  .subscribe(console.log);

Answer 4

I have never used the timer() function but what you could do is set a flag like this.

  source: Observable<number>;
  subscribe: Subscription;
  timerPaused: boolean = false;


  start() {
    this.seconds = 0;
    this.minutes = 0;
    this.hours = 0;
    this.time = 0;

    this.source = timer(0, 1000);
    this.subscribe = this.source
      .subscribe(number => {
        if(!this.timerPaused) {
          this.toSeconds = this.time % 60;
          this.toMinutes = Math.floor(this.time / 60);
          this.toHours = Math.floor(this.time / (60 * 60));

          this.seconds = (this.toSeconds < 10 ? '0' : '') + this.toSeconds;
          this.minutes = (this.toMinutes < 10 ? '0' : '') + Math.floor(number / 60);
          this.hours = (this.toHours < 10 ? '0' : '') + Math.floor(number / (60 * 60));
          this.time += 1000
        }
      });
  }

  onPause() {
    this.timerPaused = true;
  }

  onResume() {
    this.timerPaused = false;
  }

  ngOnDestroy() {
    this.subscribe.unsubscribe();
  }